Python multiprocessing.Pool进程状态诊断与超时排查

import datetime
import multiprocessing
import time
import random

def my_cool_function(a, b, shared_list):
    """
    一个模拟长时间运行且可能卡顿的函数。
    a: 任务ID
    b: 额外参数
    shared_list: 模拟共享列表，用于标记任务开始和结束
    """
    print(f"Task {a} started at {datetime.datetime.now()}")
    # 模拟任务开始，将任务ID添加到共享列表
    # 注意：在实际的多进程环境中，共享列表需要通过Manager来管理
    # 这里为了演示，我们假设shared_list是一个Manager管理的列表
    # shared_list.append(a)

    try:
        # 模拟长时间运行，某些任务可能故意延长或卡住
        if a % 5 == 0: # 模拟每5个任务中有一个耗时特别长
            time.sleep(10)
        elif a % 7 == 0: # 模拟每7个任务中有一个可能卡住
            print(f"Task {a} is simulating a long wait...")
            time.sleep(20) # 更长时间的等待
        else:
            time.sleep(random.uniform(0.5, 2))

        # 模拟返回大量数据
        return (f"Result for {a}", [f"data_{i}" for i in range(1000)])
    finally:
        # 模拟任务结束，从共享列表中移除任务ID
        # shared_list.remove(a)
        print(f"Task {a} finished at {datetime.datetime.now()}")

def main_debug_pool():
    start_time = datetime.datetime.now()
    # 使用Manager创建共享列表，以便子进程可以访问和修改
    manager = multiprocessing.Manager()
    l = manager.list() # 真实的共享列表

    large_list_a = list(range(20)) # 示例任务数量
    large_list_b = [2] * 20

    print("Starting multiprocessing pool...")
    with multiprocessing.Pool(processes=4) as pool: # 使用较小的进程数以便观察
        # 提交异步任务
        out_results = pool.starmap_async(my_cool_function, [(a, b, l) for (a, b) in zip(large_list_a, large_list_b)])

        print("\nMonitoring pool readiness...")
        while not out_results.ready():
            current_time = datetime.datetime.now()
            elapsed_time = current_time - start_time
            print(f"[{elapsed_time}] Pool not ready. Checking active processes...")

            # 核心诊断逻辑：通过exitcode过滤仍在运行的进程
            # pool._pool 是一个内部属性，包含了所有工作进程的Process对象
            active_processes = list(filter(lambda p: p.exitcode is None, pool._pool))

            if active_processes:
                print(f"  Currently {len(active_processes)} processes are active:")
                for p in active_processes:
                    print(f"    - Process Name: {p.name}, PID: {p.pid}, Alive: {p.is_alive()}, ExitCode: {p.exitcode}")
            else:
                print("  No processes appear active, but pool not ready. This is unusual, investigate queue issues.")

            time.sleep(2) # 每2秒检查一次

        print("\nPool is ready. Attempting to get results...")
        try:
            # 尝试获取结果，设置一个可能导致超时的短时间
            out_tuple_list = out_results.get(timeout=5)
            print("Results obtained successfully!")
            # print("Sample result:", out_tuple_list[0])
        except multiprocessing.TimeoutError:
            print("\n!!! multiprocessing.TimeoutError occurred when getting results !!!")
            print("Final check for active processes after timeout:")
            # 再次检查活跃进程，以确认超时原因
            active_processes = list(filter(lambda p: p.exitcode is None, pool._pool))
            if active_processes:
                print(f"  Still {len(active_processes)} processes active (likely the cause of timeout):")
                for p in active_processes:
                    print(f"    - Process Name: {p.name}, PID: {p.pid}, Alive: {p.is_alive()}, ExitCode: {p.exitcode}")
            else:
                print("  No active processes found, but timeout still occurred. This might indicate issues with result queue or internal Pool state.")
        except Exception as e:
            print(f"\nAn unexpected error occurred: {e}")
        finally:
            # 确保Manager的资源被清理
            manager.shutdown()

if __name__ == '__main__':
    main_debug_pool()

active_processes = list(filter(lambda p: p.exitcode is None, pool._pool))