跨進程共享多處理同步原語

Question

（Python 3.4，Linux）。

我有一個主進程「P」，它分叉了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

下面是一個示例通過在一個孩子中創建一個multiprocessing.managers.BaseManager，然後從所有其他孩子連接到該經理，可以做到這一點。請注意，它需要將家長的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! 

Answer 2

是否可以簡單地捕獲進程的ID和manaully打電話給你的行動只是其中之一？像這樣？

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()