如何解决如何在远程进程之间使用 Python3.9.2 multiprocessing.SyncManager？
我在 relevant Python 3.9.2 documentation regarding the synchronization of data among remote processes 上模拟了以下测试程序。不过，据我所知，它实际上并没有用，所以我认为有些事情我不知道。该文档并未明确说明如何在远程进程之间部署 SyncManager 对象，但它们毕竟是 BaseManager 子类的实例，因此必须假设相同的技术应该起作用。
在下面的代码之后是 shell 输出，显示了三个并发调用，大概说明了我遇到的问题。尽管正在与服务器建立连接，但 dict 并未同步。问题是：为什么？
#!/usr/bin/env python3
# <zteeq.py>
import multiprocessing as mp
import multiprocessing.shared_memory as sm
import multiprocessing.managers as mgrs
import os,sys
#######################################################
class CatalogManager( mgrs.SyncManager): pass
CatalogManager.register( 'get_catalog',dict,mgrs.DictProxy)
#######################################################
class ShareCatalog():
#######################################################
def __init__( self,catalogManagerAddress,catalogManagerAuthkey,**kwargs
):
self.catalogManagerAddress = catalogManagerAddress
self.catalogManagerAuthkey = catalogManagerAuthkey
#######################################################
def start( self):
self.catalogManager = CatalogManager(
self.catalogManagerAddress,self.catalogManagerAuthkey,)
try:
self.catalogManager.connect()
print( 'connected self.catalogManager')
except ConnectionRefusedError:
catalogManagerServer = self.catalogManager.get_server()
print( 'starting self.catalogManager')
catalogManagerServer.serve_forever()
self.catalog = self.catalogManager.get_catalog()
###
print( 'pid %d: first stop: %r' % ( os.getpid(),str( self.catalog)))
input()
###
if 'streams' not in self.catalog:
print( 'adding streams')
self.catalog[ 'streams'] = {}
###
print( 'pid %d: second stop: %r' % ( os.getpid(),str( self.catalog)))
input()
###
#######################################################
if __name__ == '__main__':
mp.set_start_method( 'spawn')
shareCatalog = ShareCatalog(
( '127.0.1.1',43210),b'abc',)
shareCatalog.start()
#</zteeq.py>
在第一个 shell 中，SyncManager 服务器启动：
# ./zteeq.py
starting self.catalogManager
让它继续运行，我在第二个 shell 中再次启动程序：
# ./zteeq.py
connected self.catalogManager
pid 2486196: first stop: '{}'
adding streams
pid 2486196: second stop: "{'streams': {}}"
到目前为止，一切都很好。我让它继续运行并第三次调用。但是第三次调用对第二次调用所做的事情一无所知；共享字典中没有“流”键：
# ./zteeq.py
connected self.catalogManager
pid 2492338: first stop: '{}'
我错过了什么？
（Python 3.9.2） (Linux 5.10.0-4-amd64 #1 SMP Debian 5.10.19-1 (2021-03-02) x86_64 GNU/Linux)
备注：一般来说，文档似乎假设所有 SyncManager 对象都将通过名为“multiprocessing.Manager()”的快捷方式创建，该快捷方式不提供远程套接字通信的规范。我假设这些对象会被所有将使用它的进程分叉继承，正如我迄今为止发现的所有示例所示。但这不是我想要做的。
解决方法
我做出的一些推论被证明是不正确的。我希望下面的解决方案看起来不那么笨拙和多余，但我认为最好以这种方式发布它，因为冗余本身是提供信息的。 （冗余的必要性让我感到惊讶，我仍在考虑它。）文档建议将包含的对象置于非托管状态，然后简单地调整托管容器以告诉管理器更新客户端只是行不通.我不知道为什么；可能是另一个错误的推论或误解。无论如何，以下确实有效。
#!/usr/bin/env python3
import multiprocessing as mp
import multiprocessing.shared_memory as sm
import multiprocessing.managers as mgrs
import os,sys
#######################################################
#######################################################
class ShareCatalog():
#######################################################
def __init__( self,catalogManagerAddress,catalogManagerAuthkey,**kwargs
):
self.catalogManagerAddress = catalogManagerAddress
self.catalogManagerAuthkey = catalogManagerAuthkey
#######################################################
def start( self,server):
class CatalogManager( mgrs.SyncManager): pass
if server:
catalogDict = {
'streams': {},}
CatalogManager.register( 'get_catalog',lambda:catalogDict,mgrs.DictProxy)
CatalogManager.register( 'get_streams',lambda:catalogDict[ 'streams'],mgrs.DictProxy)
self.catalogManager = CatalogManager(
self.catalogManagerAddress,self.catalogManagerAuthkey,)
catalogManagerServer = self.catalogManager.get_server()
print( 'starting self.catalogManager')
catalogManagerServer.serve_forever()
else: ## not server
CatalogManager.register( 'get_catalog')
CatalogManager.register( 'get_streams')
self.catalogManager = CatalogManager(
self.catalogManagerAddress,)
self.catalogManager.connect()
print( 'connected self.catalogManager')
self.catalog = self.catalogManager.get_catalog()
self.streams = self.catalogManager.get_streams()
###
ctr = -1
while True:
print( 'pid %d: first stop: %r' % ( os.getpid(),str( self.catalog)))
input()
ctr += 1
self.catalog.setdefault(
( os.getpid(),ctr,),None,)
self.streams[ ctr] = None
print( 'pid %d: second stop: %r' % ( os.getpid(),str( self.catalog)))
input()
###
#######################################################
if __name__ == '__main__':
mp.set_start_method( 'spawn')
shareCatalog = ShareCatalog(
( '127.0.1.1',43210),b'abc',)
shareCatalog.start( eval( sys.argv[ 1]))
和以前一样，第一次调用启动服务器：
# ./zteeq.py True ## True means "be the server"
starting self.catalogManager
第二次调用：
# ./zteeq.py False
connected self.catalogManager
pid 2767634: first stop: "{'streams': {}}"
pid 2767634: second stop: "{'streams': {0: None},(2767634,0): None}"
pid 2767634: first stop: "{'streams': {0: None},0): None}"
pid 2767634: second stop: "{'streams': {0: None,1: None},0): None,1): None}"
让第二次调用继续运行，这是第三次调用：
# ./zteeq.py False
connected self.catalogManager
pid 2767704: first stop: "{'streams': {0: None,1): None}"
pid 2767704: second stop: "{'streams': {0: None,1): None,(2767704,0): None}"
pid 2767704: first stop: "{'streams': {0: None,0): None}"
回到第二次调用，我们看到按 Enter 后第三次调用的变化：
# ./zteeq.py False
connected self.catalogManager
pid 2767634: first stop: "{'streams': {}}"
pid 2767634: second stop: "{'streams': {0: None},1): None}"
pid 2767634: first stop: "{'streams': {0: None,0): None}"
无论如何都足够好用！
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。
