multiprocessing race condition on flushing stdout, deadlocks child on exit

Bắt đầu bằng cách chạy bản tái hiện multiprocessing được cung cấp trên Linux với các phiên bản Python được báo cáo, sau đó kiểm tra multiprocessing Queue và đường dẫn thoát của tiến trình con, bao gồm các đường dẫn multiprocessing/process.py và multiprocessing/popen_fork.py được hiển thị trong traceback. Công việc được xem là hoàn tất khi các worker thực hiện join một cách ổn định sau khi các mục trong Queue được lấy ra, với coverage hồi quy cho trường hợp đã tái hiện.

I experienced deadlocks when using the logging example for multiprocessing, where a QueueHandler is used.

I found that sometimes a multiprocessing process is not terminating, when it has put an element in a Queue,
even if the parent process runs a thread to empty the queue and successfully retrieved the item.

The worker looks like this:

def worker(q):
    q.put(current_process().name)
    return

and while this works:

workers = []
for i in range(15):
    p = Process(target=worker, args=(logq,), name=f"Worker {i+1}")
    workers.append(p)

for p in workers:
    p.start()
    time.sleep(.1)

removing the sleep leads to a very high propability of deadlocking when I then try to join the processes, e.g,:

for w in workers:
    print(f'trying to join on {w.name}, alive={w.is_alive()}, exitcode={w.exitcode}', w.name, w.is_alive(), w.exitcode)
    w.join()

The Process is still marked as alive, hitting Ctrl+C gives this:

trying to join on Worker 14, alive=True, exitcode=None Worker 14 True None

^CTraceback (most recent call last):
  File "/home/fls/pybug/deadlock.py", line 43, in <module>
    w.join()
  File "/usr/lib/python3.10/multiprocessing/process.py", line 149, in join
    res = self._popen.wait(timeout)
  File "/usr/lib/python3.10/multiprocessing/popen_fork.py", line 43, in wait
    return self.poll(os.WNOHANG if timeout == 0.0 else 0)
  File "/usr/lib/python3.10/multiprocessing/popen_fork.py", line 27, in poll
    pid, sts = os.waitpid(self.pid, flag)
KeyboardInterrupt

Can reproduce using Python 3.9.10 and 3.10.4 on Linux:

import time
import threading
from multiprocessing import Process, Queue, current_process

def logger_thread(q: Queue):
    while True:
        record = q.get()
        if record is None:
            break
        print("logrecord: ", record)

def worker(q):
    q.put(current_process().name)
    return

logq = Queue()

lp = threading.Thread(target=logger_thread, args=(logq,), daemon=True)
lp.start()

workers = []
for i in range(15):
    p = Process(target=worker, args=(logq,), name=f"Worker {i+1}")
    workers.append(p)

print("starting workers")

for p in workers:
    p.start()
    # no deadlock when added:
    # time.sleep(.1)

print("waiting a bit")
time.sleep(1)
print("trying to join workers")

for w in workers:
    print(f'trying to join on {w.name}, alive={w.is_alive()}, exitcode={w.exitcode}', w.name, w.is_alive(), w.exitcode)
    w.join()
    print(f'joined on {w.name}', w.name)

logq.put(None)
lp.join()