跨进程共享多处理同步原语 [英] Sharing a multiprocessing synchronization primitive across processes

问题描述

(Python 3.4，Linux).

(Python 3.4, Linux).

我有一个主进程"P"，它分叉了8个进程("C1"至"C8").我想创建multiprocessing.Barrier，以确保所有8个子进程在某个时刻都保持同步.

I have a main process 'P', which forks 8 processes ('C1' through 'C8'). I want to create multiprocessing.Barrier that ensures all the 8 child processes are in sync at a certain point.

如果我在父进程中定义了同步原语，那么一切都会很好，这样当我分叉子进程时，它会被正确继承:

Everything works fine if I define the synchronization primitive in the parent process, so that when I fork the child processes it is properly inherited:

import multiprocessing as mp
barrier = mp.Barrier(8)

def f():
  # do something
  barrier.wait()
  # do more stuff

def main():
  for i in range(8):
    p = mp.Process(target = f)
    p.start()

if __name__ == '__main__':
  main()

但是，在我的情况下，直到子进程启动后，我才知道创建Barrier对象所需的详细信息(我不知道我想作为其action参数传递的参数).因此，我想在一个子进程中创建Barrier，但是我不知道如何使它可用于其他子进程.由于子进程中的8个Barrier对象是彼此完全独立的，因此以下操作当然不会起作用:

But in my case, I do not know the details required to create the Barrier object until after the child processes start (I don't know the argument I want to pass as its action parameter). Therefore, I want to create Barrier in one of the child processes but I don't know how to make it available to the other child processes. The following won't work of course because the 8 Barrier objects in the child process are completely independent from each other:

import multiprocessing as mp

def f():
  global barrier
  # do something
  barrier = mp.Barrier(8)
  barrier.wait()
  # do more stuff

def main():
  for i in range(8):
    p = mp.Process(target = f)
    p.start()

if __name__ == '__main__':
  main()

我当时正在考虑在其中一个子进程中创建barrier，然后使用multiprocessing.Queue(或者如果Queue不接受Barrier对象，则使用multiprocessing.Manager().Barrier)将其传递给其他子进程.但是，即使这行得通，我也不知道如何确保只有一个进程将同步原语(的7个副本)实际put真正地放入队列，而其他进程仅get. (当然，我可以在父进程中创建另一个同步原语来做到这一点，但毕竟我还是可以重构代码以在父进程中创建原始的Barrier.)

I was thinking to create barrier in one of the child processes and pass it to the others using multiprocessing.Queue (or if Queue doesn't accept Barrier objects, using multiprocessing.Manager().Barrier). However, even if this works, I don't know how to ensure only one process actually puts the (7 copies of) synchronization primitives onto the queue, while the others will only get them. (Of course, I can create a yet another synchronization primitive in the parent process just to do that, but then I might as well refactor my code to create the original Barrier in the parent process after all.)

推荐答案

是否可以简单地捕获进程的ID并仅在其中一个中手动调用您的操作?像这样吗?

Would it be possible to simply capture the id's of the processes and manaully call your action in only one of them? Something like this?

import multiprocessing as mp
barrier = mp.Barrier(8)

def f():
  # create action
  def action():
      print("action was run on process {}.".format(id))

  # do something
  print("Hello from process {}.".format(id))
  id = barrier.wait()
  if id == 0:
      action()
  barrier.wait()

  # Do more stuff

def main():
  for i in range(8):
    p = mp.Process(target = f)
    p.start()

if __name__ == '__main__':
  main()

这篇关于跨进程共享多处理同步原语的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！