python - 使用python多处理与函数共享信号量

Question

我想通过将信号量传递给函数来在 python multiprocessing.Pool 中同步对共享资源的访问。这是一些伪代码。

def do_work(payload, semaphore):
    with semaphore:
        access_the_shared_resource(payload)

do_work 函数是在一个库中定义的，因此我无法在我的本地范围内定义该函数可以继承的信号量。我也不能使用 functools.partial 传递信号量，因为 multiprocessing 试图 pickle 不允许的信号量。似乎有效的是使用 multiprocessing.Manager 创建一个 Semaphore 的代理:

manager = multiprocessing.Manager()
semaphore = manager.Semaphore()

with multiprocessing.Pool() as pool:
    results = pool.map(functools.partial(do_work, semaphore=semaphore), payloads)

这是最好的方法还是我错过了明显的解决方案？

Answer 1

您唯一的其他选择是使用 initializer 和 initargs 在池创建时将常规 multiprocessing.Semaphore 传递给每个工作进程，并将其用作全局变量:

semaphore = None
def do_work(payload):
    with semaphore:
        return payload

def init(sem):
    global semaphore
    semaphore = sem

if __name__ == "__main__":
    sem = multiprocessing.Semaphore()
    with multiprocessing.Pool(initializer=init, initargs=(sem,)) as p:
        results = p.map(do_work, payloads)

您在父进程中定义的全局semaphore 变量将被设置为每个子进程中的multiprocessing.Semaphore()。

使用 manager.Semaphore() 很好，尽管它确实需要产生一个额外的 Python 进程。