并行任务执行完毕后，状态为何始终未更新为finished？

我用以下代码运行并行任务，从输出中能看到任务已完成，但status_dict中对应任务的状态始终为False，预期任务完成后状态应更新为True，但实际一直保持False。

控制台输出 #

----------123
{'JOB4': False, 'JOB11': False, 'JOB23': False}
----
JOB4 False
Starting job JOB11...
Starting job JOB4...
Starting job JOB23...
----
JOB11 False
----
JOB23 False
----
JOB4 False
Finished job JOB23.
Finished job JOB4.
Finished job JOB11.
----
JOB11 False
----
JOB23 False
----
JOB4 False
----
JOB11 False

原代码 #

import multiprocessing, time

# run jobs parallel
def run_job(jn, par):
    print(f"Starting job {jn}...")
    time.sleep(3)
    print(f"Finished job {jn}.")


if __name__ == '__main__':    
    list_jobs = ["JOB4", "JOB11", "JOB23"]
    processes = []

    status_dict = multiprocessing.Manager().dict()
    for jn in list_jobs:
        par = {}
        par["sleep"] = 5
        status_dict[jn] = False
        p = multiprocessing.Process(target=run_job, args=(jn, par))
        processes.append(p)
        p.start()

    
    print("------------123")
    print(status_dict)
    while not all(status_dict.values()):
        for job_id, status in status_dict.items():
            print("----")
            print(job_id, status)
            time.sleep(1)


    print("ENDE")

解决方案 #

问题核心是子进程从未更新共享字典的状态：虽然用multiprocessing.Manager().dict()创建了跨进程共享的字典，但原代码中run_job函数在任务结束后没有对status_dict做任何修改，所以主进程看到的始终是初始的False。

只需要两步修改：

1. 将status_dict作为参数传递给run_job函数
2. 在任务完成后，将对应任务的状态设为True

修改后的代码 #

import multiprocessing, time

# run jobs parallel
def run_job(jn, par, status_dict):
    print(f"Starting job {jn}...")
    time.sleep(3)
    print(f"Finished job {jn}.")
    # 任务完成后更新状态
    status_dict[jn] = True


if __name__ == '__main__':    
    list_jobs = ["JOB4", "JOB11", "JOB23"]
    processes = []

    status_dict = multiprocessing.Manager().dict()
    for jn in list_jobs:
        par = {}
        par["sleep"] = 5
        status_dict[jn] = False
        # 把status_dict传给子进程
        p = multiprocessing.Process(target=run_job, args=(jn, par, status_dict))
        processes.append(p)
        p.start()

    
    print("------------123")
    print(status_dict)
    while not all(status_dict.values()):
        for job_id, status in status_dict.items():
            print("----")
            print(job_id, status)
            time.sleep(1)


    print("ENDE")

修改后，子进程在任务结束时会主动更新共享字典的状态，主进程的循环就能检测到状态变化，最终正常退出循环并输出ENDE。

内容的提问来源于stack exchange，提问作者olli