2017-04-13 134 views
0

我刚刚学会了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”]获得元素,有什么问题呢?

+0

我不知道'AllQueue'定义。另请编辑您的问题以删除冗余元素:是否需要所有进口? – quamrana

+0

我试图把AllQueue的定义从主要的,或使其成为全局变量,结果是一样的...... – Coneain

+0

当你使用'ProcessPoolExecutor'或'multiprocessing.Pool'时,AllQueue似乎超出了范围。您需要通过'map'调用将'AllQueue'作为参数传递。 – quamrana

回答

0

这是通过字典中的任务的例子:

from concurrent.futures import ProcessPoolExecutor 
from multiprocessing import Manager 


def addSigal(target, information, q): 
    print(target,information,q) 
    q[target]=(information) 
    print("Succeed in sending msg to "+target) 
    print(target+' now has ',q[target]) 


if __name__=='__main__': 
    m = Manager() 
    AllQueue = m.dict() 
    AllQueue.update({'A':0,'B':1}) 
    with ProcessPoolExecutor() as pool: 
     pool.map(addSigal,'AB', [1, 2],[AllQueue,AllQueue]) 
    print(AllQueue) 
+0

是的,你是对的。这适用于一对一,但是多对一的情况如何?一个人收到一个数字序列而不是一个数字,字典的价值只是一个数字,恐怕这还不够... – Coneain

+0

嗯,我认为它是由你来工作。我已经证明,主要问题是不同的进程有自己的变量副本,解决方法是发送任何全局通过参数。你只需要继续这样做。 – quamrana

+0

好的,就是这一点!我知道了。谢谢! – Coneain