'multiprocessing.resource_sharer' 中的 AttributeError 'DupFd' | Python多处理+线程

Question

我正在尝试在执行 I/O 绑定任务的多个 threading.Thread(s) 和执行 CPU 绑定任务的多个 multiprocessing.Process(es) 之间进行通信。每当一个线程为一个进程找到工作时，它将被放在一个multiprocessing.Queue 上，以及一个multiprocessing.Pipe(duplex=False) 的发送端。然后，这些进程会尽其所能，并通过管道将结果发送回线程。这个过程似乎在大约 70% 的情况下有效，另外 30% 我收到了AttributeError: Can't get attribute 'DupFd' on <module 'multiprocessing.resource_sharer' from '/usr/lib/python3.5/multiprocessing/resource_sharer.py'>

复制：

import multiprocessing
import threading
import time

def thread_work(work_queue, pipe):
    while True:
        work_queue.put((threading.current_thread().name,  pipe[1]))
        received = pipe[0].recv()
        print("{}: {}".format(threading.current_thread().name, threading.current_thread().name == received))
        time.sleep(0.3)

def process_work(work_queue):
    while True:
        thread, pipe = work_queue.get()
        pipe.send(thread)

work_queue = multiprocessing.Queue()
for i in range(0,3):
    receive, send = multiprocessing.Pipe(duplex=False)
    t = threading.Thread(target=thread_work, args=[work_queue, (receive, send)])
    t.daemon = True
    t.start()

for i in range(0,2):
    p = multiprocessing.Process(target=process_work, args=[work_queue])
    p.daemon = True
    p.start()

time.sleep(5)

我查看了多处理source code，但不明白为什么会发生此错误。 我尝试使用queue.Queue，或带有duplex=True（默认）的管道，但在错误中找不到模式。有人知道如何调试吗？

Answer 1

您在这里分叉了一个已经多线程的主进程。众所周知，这通常是有问题的。

它实际上很容易出现问题（不仅仅是在 Python 中）。规则是“在你分叉之后线程，而不是之前”。否则，线程执行器使用的锁将在进程间重复。如果其中一个进程在拥有锁时死掉了，那么使用该锁的所有其他进程都将死锁 -Raymond Hettinger。

您得到的错误的触发显然是管道文件描述符的复制在子进程中失败。

要解决此问题，只要您的主进程仍然是单线程的，就创建您的子进程，或者使用另一个 start_method 来创建新进程，如“spawn”（Windows 上的默认值）或“forkserver”，如果有的话。

分叉服务器

当程序启动并选择forkserver启动方式时，启动一个服务器进程。从那时起，每当需要一个新进程时，父进程都会连接到服务器并请求它派生一个新进程。 fork 服务器进程是单线程的，因此使用 os.fork() 是安全的。不会继承不必要的资源。

在支持通过 Unix 管道传递文件描述符的 Unix 平台上可用。 docs

您可以指定另一个 start_method：

multiprocessing.set_start_method(方法) 设置应该用于启动子进程的方法。方法可以是“fork”、“spawn”或“forkserver”。

请注意，这应该最多调用一次，并且应该在主模块的 if name == 'ma​​in' 子句中保护它。 docs

有关特定 start_methods（在 Ubuntu 18.04 上）的基准测试，请查看 here。