使用队列的Python多处理死锁

我有一个如下所示的 python 程序。

from multiprocessing import Lock, Process, Queue, current_process 
import time 
 
lock = Lock() 
 
 
def do_job(tasks_to_accomplish, tasks_that_are_done): 
    while not tasks_to_accomplish.empty(): 
        task = tasks_to_accomplish.get() 
        print(task) 
        lock.acquire() 
        tasks_that_are_done.put(task + ' is done by ' + current_process().name) 
        lock.release() 
        time.sleep(1) 
    return True 
 
 
def main(): 
    number_of_task = 10 
    number_of_processes = 4 
    tasks_to_accomplish = Queue() 
    tasks_that_are_done = Queue() 
    processes = [] 
 
    for i in range(number_of_task): 
        tasks_to_accomplish.put("Task no " + str(i)) 
 
    # creating processes 
    for w in range(number_of_processes): 
        p = Process(target=do_job, args=(tasks_to_accomplish, tasks_that_are_done)) 
        processes.append(p) 
        p.start() 
 
 
    # completing process 
    for p in processes: 
        p.join() 
 
    # print the output 
    while not tasks_that_are_done.empty(): 
        print(tasks_that_are_done.get()) 
 
    return True 
 
 
if __name__ == '__main__': 
    main() 

有时程序运行完美，但有时会卡住，无法完成。手动退出时，会产生如下错误。

$ python3 multiprocessing_example.py  
Task no 0 
Task no 1 
Task no 2 
Task no 3 
Task no 4 
Task no 5 
Task no 6 
Task no 7 
Task no 8 
Task no 9 
^CProcess Process-1: 
Traceback (most recent call last): 
  File "multiprocessing_example.py", line 47, in <module> 
    main() 
  File "multiprocessing_example.py", line 37, in main 
    p.join() 
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/process.py", line 121, in join 
    res = self._popen.wait(timeout) 
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/popen_fork.py", line 51, in wait 
    return self.poll(os.WNOHANG if timeout == 0.0 else 0) 
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/popen_fork.py", line 29, in poll 
    pid, sts = os.waitpid(self.pid, flag) 
KeyboardInterrupt 
Traceback (most recent call last): 
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/process.py", line 249, in _bootstrap 
    self.run() 
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/process.py", line 93, in run 
    self._target(*self._args, **self._kwargs) 
  File "multiprocessing_example.py", line 9, in do_job 
    task = tasks_to_accomplish.get() 
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/queues.py", line 94, in get 
    res = self._recv_bytes() 
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/connection.py", line 216, in recv_bytes 
    buf = self._recv_bytes(maxlength) 
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/connection.py", line 407, in _recv_bytes 
    buf = self._recv(4) 
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/connection.py", line 379, in _recv 
    chunk = read(handle, remaining) 
KeyboardInterrupt 

有人能告诉我这个程序有什么问题吗？我正在使用 python 3.6。

请您参考如下方法:

Note: Lock is not needed around a Queue.

    lock.acquire() 
    tasks_that_are_done.put(task + ' is done by ' + current_process().name) 
    lock.release() 

Queue
The Queue class in this module implements all the required locking semantics.

Question: ... what is the issue with the program?

您正在使用 Queue.empty() 和 Queue.get()， 这会导致调用 join() 时出现死锁，因为无法保证 empty() 状态不会直到 get() 才改变 正在到达。

Deadlock prone:

while not tasks_to_accomplish.empty(): 
    task = tasks_to_accomplish.get() 

不使用empty/get，配对使用例如:

import queue 
while True: 
    try: 
        task = tasks_to_accomplish.get_nowait() 
    except queue.Empty: 
        break 
    else: 
        # Handle task here 
        ... 
        tasks_to_accomplish.task_done()