管道卡住了——堆栈溢出工作没有其他解决方案

Question

（更新）我正在构建一个模块来分发基于代理的模型，其想法是将模型拆分为多个进程，然后当代理到达边界时，它们被传递给处理该区域的处理器。我可以设置进程并在没有通信的情况下工作，但无法让数据通过管道并更新另一个处理器上的模型段。

我已经在 stackoverflow 上尝试了解决方案，并构建了一个简单版本的模型。一旦我将模型对象放入管道中，模型就会挂起（它适用于 python 标准数据类型）。简单的版本只是来回传递代理。

 from pathos.multiprocessing import ProcessPool
 from pathos.helpers import mp
 import copy



 class TestAgent:

 "Agent Class-- Schedule iterates through each agent and \
  executes step function"

 def __init__(self, unique_id, model):
      self.unique_id = unique_id
      self.model = model
      self.type = "agent"

 def step(self):

       pass 
       #print ('     ', self.unique_id, "I have stepped")

class TestModel:

   "Model Class iterates through schedule and executes step function for \
   each agent"

   def __init__(self):
       self.schedule = []
       self.pipe = None
       self.process = None

       for i in range(1000):
           a = TestAgent(i, self)
           self.schedule.append(a)


   def step(self):


       for a in self.schedule:
           a.step()

if __name__ == '__main__':

   pool = ProcessPool(nodes=2)

   #create instance of model
   test_model = TestModel()
   #create copies of model to be run on 2 processors
   test1 = copy.deepcopy(test_model)
   #clear schedule
   test1.schedule = []
   #Put in only half the schedule
   for i in range(0,500):
       test1.schedule.append(test_model.schedule[i])  
   #Give process tracker number
   test1.process = 1
   #repeat for other processor
   test2= copy.deepcopy(test_model)
   test2.schedule = []
   for i in range(500,1000):
       test2.schedule.append(test_model.schedule[i])
   test2.process = 2

   #create pipe
   end1, end2 = mp.Pipe()

   #Main run function for each process
   def run(model, pipe):

      for i in range(5):
          print (model.process)#, [a.unique_id for a in model.schedule])
          model.step() # IT HANGS AFTER INITIAL STEP
          print ("send")
          pipe.send(model.schedule)
          print ("closed")
          sched = pipe.recv()
          print ("received")
          model.schedule = sched



   pool.map(run, [test1, test2], [end1,end2])

代理应切换处理器并执行其打印功能。 （我的下一个问题是同步处理器，使它们保持在每一步，但一次只做一件事。）

Answer 1

我让它工作了。我超出了 python (8192) 中的管道缓冲区限制。如果代理持有模型的副本作为属性，则尤其如此。下面是上述代码的工作版本，它一次通过一个代理。它使用 Pimpler 来获取所有代理的大小。

from pathos.multiprocessing import ProcessPool
from pathos.helpers import mp
import copy

# do a blocking map on the chosen function

class TestAgent:

"Agent Class-- Schedule iterates through each agent and \
executes step function"

   def __init__(self, unique_id, model):
       self.unique_id = unique_id
       self.type = "agent"

   def step(self):
       pass 


class TestModel:

   "Model Class iterates through schedule and executes step function for \
   each agent"

   def __init__(self):
       from pympler import asizeof 

       self.schedule = []
       self.pipe = None
       self.process = None
       self.size = asizeof.asizeof


       for i in range(1000):
           a = TestAgent(i, self)
           self.schedule.append(a)


   def step(self):


       for a in self.schedule:
           a.step()

if __name__ == '__main__':

   pool = ProcessPool(nodes=2)

   #create instance of model
   test_model = TestModel()
   #create copies of model to be run on 2 processors
   test1 = copy.deepcopy(test_model)
   #clear schedule
   test1.schedule = []
   #Put in only half the schedule
   for i in range(0,500):
       test1.schedule.append(test_model.schedule[i])  
   #Give process tracker number
   test1.process = 1
   #repeat for other processor
   test2= copy.deepcopy(test_model)
   test2.schedule = []
   for i in range(500,1000):
       test2.schedule.append(test_model.schedule[i])
   test2.process = 2

   #create pipe
   end1, end2 = mp.Pipe()

   #Main run function for each process
   def run(model, pipe):

      for i in range(5):
        agents = []
        print (model.process, model.size(model.schedule) ) 
        model.step() # IT HANGS AFTER INITIAL STEP
        #agent_num = list(model.schedule._agents.keys())
        for agent in model.schedule[:]:
            model.schedule.remove(agent)
            pipe.send(agent)
            agent = pipe.recv()
            agents.append(agent)
        print (model.process, "all agents received")
        for agent in agents: 
            model.schedule.append(agent)

        print (model.process, len(model.schedule))



   pool.map(run, [test1, test2], [end1,end2])

Mike McKerns 和 Thomas Moreau -- 感谢您帮助我走上正确的道路。