我刚刚学会了python多处理。我想制作一个模型来模拟在网络中发送和接收消息的过程。有向图描述了两个节点之间的关系,而一个字典描述了两个节点之间的通信。该字典的值的数据类型是队列。但是,我遇到了一些错误:python在多处理中共享dict()中共享队列()
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import Manager
PoolGroup=[('R1','R2','R3'),('N1','N2','N3'),('E1','E2','E3')]
PoolElement=['R1','R2','R3','N1','N2','N3','E1','E2','E3']
graph={'R1':['N1','N2','N3'],
'R2':['N1','N2','N3'],
'R3':['N1','N2','N3'],
'N1':['E1','E2','E3'],
'N2':['E1','E2','E3'],
'N3':['E1','E2','E3'],
'E1':[],
'E2':[],
'E3':[]}
def addSigal(target,information):
AllQueue[target].put(information)
print("Succeed in sending msg to "+target)
print(target+' now has ',AllQueue[target].qsize(),' signals')
def pool1function(name,information):
targetlist=list(graph[name])
print(name+" send information to "+str(targetlist))
with ProcessPoolExecutor() as pool1:
pool1.map(addSigal,targetlist,[information]*3)
if __name__=='__main__':
m=Manager()
AllQueue=m.dict()
AllQueue.update({PE:m.Queue() for PE in PoolElement})
with ProcessPoolExecutor() as pool:
pool.map(pool1function,PoolGroup[0],[1,2,3])
不幸的是,结果刚刚出现:
R1 send information to ['N1', 'N2', 'N3']
R2 send information to ['N1', 'N2', 'N3']
R3 send information to ['N1', 'N2', 'N3']
这意味着信息不被发送到相应的节点。所以,我检查AllQueue并发现了一些奇怪的:当我打印AllQueue [“R1”],它表明:
RemoteError:
---------------------------------------------------------------------------
Unserializable message: ('#RETURN', <queue.Queue object at 0x10edd8dd8>)
---------------------------------------------------------------------------
我还没有把或AllQueue [“R1”]获得元素,有什么问题呢?
我不知道'AllQueue'定义。另请编辑您的问题以删除冗余元素:是否需要所有进口? – quamrana
我试图把AllQueue的定义从主要的,或使其成为全局变量,结果是一样的...... – Coneain
当你使用'ProcessPoolExecutor'或'multiprocessing.Pool'时,AllQueue似乎超出了范围。您需要通过'map'调用将'AllQueue'作为参数传递。 – quamrana