如何诊断Python multiprocessing.Pool 中无响应的进程

import datetime
import multiprocessing
import time
import random

def my_function(a, b, c_list):
    """
    一个模拟耗时任务的函数。
    c_list 是一个共享列表，用于跟踪正在进行的任务。
    """
    pid = multiprocessing.current_process().pid
    print(f"Process {pid}: Starting task {a}")
    c_list.append(a) # 标记任务开始
    try:
        # 模拟一个可能耗时很长或卡住的操作
        # 例如，这里我们模拟某些任务会随机卡住
        if a % 10 == 0 and random.random() < 0.5: # 模拟5%的概率卡住
            print(f"Process {pid}: Task {a} is intentionally stuck...")
            while True: # 模拟无限循环
                time.sleep(100)
        else:
            time.sleep(random.uniform(1, 5)) # 模拟正常任务耗时1-5秒
            result = (f"result_for_{a}", [f"data_{i}" for i in range(a)])
            print(f"Process {pid}: Finished task {a}")
            return result
    finally:
        # 确保任务完成或异常后从共享列表中移除
        if a in c_list:
            c_list.remove(a)

def main():
    start_time = datetime.datetime.now()
    large_list_a = list(range(20))
    large_list_b = [2] * 20

    # 使用Manager来创建一个可以在进程间共享的列表
    manager = multiprocessing.Manager()
    shared_list_c = manager.list()

    print("Starting multiprocessing pool...")
    with multiprocessing.Pool(processes=5) as pool:
        # 提交任务到进程池
        async_results = pool.starmap_async(my_function, [(a, b, shared_list_c) for a, b in zip(large_list_a, large_list_b)])

        while not async_results.ready():
            current_time = datetime.datetime.now()
            elapsed_time = current_time - start_time
            print(f"Elapsed: {elapsed_time.total_seconds():.2f}s, Active tasks in shared list: {list(shared_list_c)}")

            # 检查是否超时，这里模拟一个较短的超时，以便观察
            if elapsed_time > datetime.timedelta(seconds=20):
                print("Monitoring timeout reached. Attempting to diagnose stalled processes.")
                break # 跳出循环，进入诊断阶段

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

        try:
            # 尝试获取结果，设置一个更长的超时以捕获实际的TimeoutError
            out_tuple_list = async_results.get(timeout=30)
            print("All tasks completed successfully.")
            print(f"Results: {len(out_tuple_list)} items.")
        except multiprocessing.TimeoutError:
            print("\n!!! multiprocessing.TimeoutError: Results not ready within the specified timeout.")
            print("Initiating diagnosis of pool processes...")

            # 诊断：查找仍在运行的进程
            stalled_processes = list(filter(lambda p: p.exitcode is None, pool._pool))
            if stalled_processes:
                print(f"Found {len(stalled_processes)} potentially stalled processes:")
                for p in stalled_processes:
                    print(f"  Process Name: {p.name}, PID: {p.pid}, Exitcode: {p.exitcode}")
            else:
                print("No processes found with exitcode == None. All workers appear to have finished or exited.")

            # 可以选择终止池中的所有进程
            print("Terminating the pool to clean up...")
            pool.terminate() # 强制终止所有工作进程
            pool.join()      # 等待所有工作进程终止
            print("Pool terminated.")
        except Exception as e:
            print(f"An unexpected error occurred: {e}")
            pool.terminate()
            pool.join()

    print("Main function finished.")
    manager.shutdown() # 关闭Manager

if __name__ == '__main__':
    main()

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

Found 1 potentially stalled processes:
  Process Name: SpawnPoolWorker-2, PID: 12345, Exitcode: None