如何在不传递引用的情况下在 Python 中使用 SyncManager 跨进程共享列表

Posted 2023-02-16

【中文标题】如何在不传递引用的情况下在 Python 中使用 SyncManager 跨进程共享列表

【英文标题】：How do I share a list across processes with SyncManager in Python without passing references

【发布时间】：2020-11-24 01:08:24

【问题描述】：

在 python 中，多处理模块提供了可以在进程之间生成共享列表/字典的管理器。

但是，如果访问管理器的进程不是子进程，而是通过Manager.connect 连接到管理器，那么我在使用这些共享对象时会遇到问题。

这是一个非常基本的示例：我正在尝试创建一个可由一组进程访问的共享列表。对于这个例子，我只是在两个终端窗口中启动了两次相同的代码：

import os, time
from multiprocessing.managers import SyncManager
            
def main() -> None:
    print(f"I am process os.getpid()")
    print(f"Starting proxy server...")
 
    manager = SyncManager(address=("127.0.0.1", 8000), authkey=b"noauth")
    try:
        manager.start() # will start the sync process if it doesn't exist
    except:
        manager.connect() # if it does already exist, connect to it instead
 
    print(f"Proxy server started/connected")
 
    # would like to generate a shared list that each process can access.
    sharedList = manager.list() # this generates a new list, so each process gets their own, which is not what I want
 
    sharedList.append(os.getpid())
 
    time.sleep(20)
 
if __name__ == '__main__':
    main()

Pythons documentation on using remote managers 似乎与我正在寻找的相似，但没有关于如何获得 Manager.list 或 Manager.dict 共享的信息。

注意：我也非常乐意共享一个命名空间对象。

【问题讨论】：

我的想法是你必须在两位经理之间建立某种沟通渠道。这个 SO question 似乎解释了如何做到这一点：***.com/questions/4467626/…。完成此操作后，我想知道您是否可以让一侧的经理创建一个共享列表，然后通过通信渠道将其（实际上是对它的引用）发送给另一位经理。然后，另一个 Manager 应该为该对象创建一个您可以在该端使用的代理。

【参考方案1】：

这就是我最终解决问题的方法。您需要手动生成一个拥有共享列表的管理器进程。

import multiprocessing
from multiprocessing import process
import os, time, sys
from multiprocessing.managers import SyncManager, ListProxy
from queue import Queue

class SharedStorage(SyncManager):
    pass

def ManagerProcess():
    sys.stdout = open(os.devnull, 'w') 
    sys.stderr = open(os.devnull, 'w') 

    l = list()

    SharedStorage.register('get_list', lambda: l, ListProxy)
    try: 
        ss = SharedStorage(address=("127.0.0.1", 8000), authkey=b"noauth")
        ss.get_server().serve_forever()
    except OSError:
        # failed to listen on port - already in use.
        pass

def main() -> None:
    print(f"I am process os.getpid()")
    print(f"Starting proxy server...")

    mainProcess = multiprocessing.Process(target=ManagerProcess, daemon=True)
    mainProcess.start()

    SharedStorage.register('get_list')
    manager = SharedStorage(address=("127.0.0.1", 8000), authkey=b"noauth")
    manager.connect()

    print(f"Proxy server started/connected")

    # required - see https://bugs.python.org/issue7503
    multiprocessing.current_process().authkey = b"noauth"

    # get reference to the shared list object
    shared_list = manager.get_list()

    shared_list.append(os.getpid())

    for i in shared_list:
        print(i)

    time.sleep(20)

if __name__ == '__main__':
    main()

这可以安全地运行多次，因为后续进程产生的管理器进程将在无法侦听端口后退出。