python - 跨进程共享多处理同步原语

Question

(Python 3.4、Linux)。

我有一个主进程“P”，它 fork 8 个进程(“C1”到“C8”)。我想创建 multiprocessing.Barrier 以确保所有 8 个子进程在特定点同步。

如果我在父进程中定义同步原语，一切都会正常工作，这样当我派生子进程时，它就会被正确继承:

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 对象彼此完全独立:

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 个副本)同步原语放入队列，而其他进程只会get 他们。 (当然，我可以在父进程中创建另一个同步原语来做到这一点，但我也可以重构我的代码，毕竟在父进程中创建原始的 Barrier。)

Answer 1

这是一个示例，说明如何通过在一个 child 中创建一个 multiprocessing.managers.BaseManager，然后从所有其他 child 连接到该管理器来实现这一点。请注意，出于同步目的，它需要将 multiprocessing.Lock 从父级传递给所有子级，您提到过您希望避免这种情况。不过，我不确定是否还有其他选择。

import multiprocessing as mp
from multiprocessing.managers import BaseManager

class MyManager(BaseManager):
    pass

def f(lock):
  # do something
  with lock:
      try:
          MyManager.register('get_barrier')
          m = MyManager(address=('localhost', 5555), authkey=b'akey')
          m.connect()
          b = m.get_barrier()
          print("Got the barrier from the manager")
      except OSError as e:
          # We are the first. Create the manager, register
          # a mp.Barrier instance with it, and start it up.
          print("Creating the manager...")
          b = mp.Barrier(8)
          MyManager.register('get_barrier', callable=lambda:b)
          m = MyManager(address=('localhost', 5555), authkey=b'akey')
          m.start()
  b.wait()
  print("Done!")
  # do more stuff

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

if __name__ == '__main__':
  main()

输出:

Creating the manager...
Got the barrier from the manager
Got the barrier from the manager
Got the barrier from the manager
Got the barrier from the manager
Got the barrier from the manager
Got the barrier from the manager
Got the barrier from the manager
Done!
Done!
Done!
Done!
Done!
Done!
Done!
Done!