深入理解 multiprocessing.Pool：诊断未完成任务的进程

# 假设 pool 是一个 multiprocessing.Pool 实例
# 并且已经捕获了 TimeoutError

active_or_stuck_processes = list(filter(lambda p: p.is_alive() and p.exitcode is None, pool._pool))

if active_or_stuck_processes:
    print(f"发现 {len(active_or_stuck_processes)} 个仍在运行或可能挂起的进程:")
    for p in active_or_stuck_processes:
        print(f"  - 进程名称: {p.name}, PID: {p.pid}, Exitcode: {p.exitcode}")
else:
    print("未发现仍在运行或挂起的进程，可能在检查时已退出。")

import multiprocessing
import time
import random

def worker_function(task_id, duration):
    """
    模拟一个可能长时间运行或挂起的任务。
    如果 duration 为负数，模拟一个长时间挂起的任务。
    """
    process_name = multiprocessing.current_process().name
    print(f"[{process_name}] Task {task_id} started (expected duration: {duration}s)")
    try:
        if duration < 0:
            # 模拟一个非常长的操作，导致外部超时
            time.sleep(300)
            return f"Task {task_id} unexpectedly long"
        time.sleep(duration)
        print(f"[{process_name}] Task {task_id} finished")
        return f"Task {task_id} completed successfully"
    except Exception as e:
        print(f"[{process_name}] Task {task_id} failed with {e}")
        # 重新抛出异常，让进程退出码反映问题
        raise

def run_pool_example():
    num_tasks = 10
    pool_size = 3
    tasks_data = []
    # 创建正常任务
    for i in range(num_tasks - 1):
        tasks_data.append((i, random.uniform(1, 2))) # 1到2秒的随机任务
    # 模拟一个会挂起的任务
    tasks_data.append((num_tasks - 1, -1)) # 持续时间为负数表示挂起

    print(f"--- 启动 Pool，共 {pool_size} 个进程，处理 {num_tasks} 个任务 ---")

    with multiprocessing.Pool(processes=pool_size) as pool:
        async_result = pool.starmap_async(worker_function, tasks_data)

        try:
            # 设置一个较短的超时时间来触发 TimeoutError
            print("\n--- 尝试获取结果 (超时10秒) ---")
            results = async_result.get(timeout=10)
            print("\n所有任务成功完成:")
            for res in results:
                print(f"- {res}")
        except multiprocessing.TimeoutError:
            print("\n>>> 捕获到 multiprocessing.TimeoutError！Pool 未在规定时间内完成。")
            print(">>> 开始诊断未完成的进程...")

            # 诊断步骤：检查 pool._pool 中的进程状态
            print("\n--- 检查 Pool 内部进程状态 ---")
            active_or_stuck_processes = []
            for p in pool._pool:
                print(f"  - 进程名称: {p.name}, PID: {p.pid}, is_alive(): {p.is_alive()}, exitcode: {p.exitcode}")
                if p.is_alive() and p.exitcode is None:
                    active_or_stuck_processes.append(p)

            if active_or_stuck_processes:
                print(f"\n发现 {len(active_or_stuck_processes)} 个仍在运行或可能挂起的进程:")
                for p in active_or_stuck_processes:
                    print(f"  - 进程名称: {p.name}, PID: {p.pid}")
            else:
                print("\n未发现仍在运行或挂起的进程，可能是在检查时已退出或已完成。")

            # 在实际应用中，这里可能需要调用 pool.terminate() 来强制关闭进程
            # pool.terminate()
            # pool.join()
        except Exception as e:
            print(f"\n发生未知错误: {e}")

    print("\n--- 主程序执行完毕 ---")

if __name__ == '__main__':
    run_pool_example()