使用multiprocessing在Python进程之间共享数据的问题

Question

我看过一些关于这个的帖子，所以我知道这是相当简单的，但我似乎短。我不确定是否需要创建一个工作池，或者使用Queue类。基本上，我希望能够创建几个进程，每个进程都自主地行动(这就是为什么它们继承自Agent超类)。

在主循环的随机滴答声中，我想要更新每个Agent。我在主循环和代理的运行循环中使用具有不同值的time.sleep来模拟不同的处理器速度。

这是我的Agent超类:

# Generic class to handle mpc of each agent
class Agent(mpc.Process):
  # initialize agent parameters
  def __init__(self,):
    # init mpc
    mpc.Process.__init__(self)
    self.exit = mpc.Event()
  # an agent's main loop...generally should be overridden
  def run(self):
    while not self.exit.is_set():
      pass
    print "You exited!"
  # safely shutdown an agent
  def shutdown(self):
    print "Shutdown initiated"
    self.exit.set()
  # safely communicate values to this agent
  def communicate(self,value):
    print value

特定代理的子类(模拟HVAC系统):

class HVAC(Agent):
  def __init__(self, dt=70, dh=50.0):
    super(Agent, self).__init__()
    self.exit = mpc.Event()
    self.__pref_heating     = True
    self.__pref_cooling     = True
    self.__desired_temperature = dt
    self.__desired_humidity    = dh
    self.__meas_temperature = 0
    self.__meas_humidity    = 0.0
    self.__hvac_status      = "" # heating, cooling, off
    self.start()
  def run(self): # handle AC or heater on 
    while not self.exit.is_set():
      ctemp = self.measureTemp()
      chum  = self.measureHumidity()
      if (ctemp < self.__desired_temperature):
        self.__hvac_status = 'heating'
        self.__meas_temperature += 1
      elif (ctemp > self.__desired_temperature):
        self.__hvac_status = 'cooling'
        self.__meas_temperature += 1
      else:
        self.__hvac_status = 'off'
      print self.__hvac_status, self.__meas_temperature

      time.sleep(0.5)

    print "HVAC EXITED"
  def measureTemp(self):
    return self.__meas_temperature
  def measureHumidity(self):
    return self.__meas_humidity
  def communicate(self,updates):
    self.__meas_temperature = updates['temp']
    self.__meas_humidity    = updates['humidity']
    print "Measured [%d] [%f]" % (self.__meas_temperature,self.__meas_humidity)

我的主循环:

if __name__ == "__main__":
  print "Initializing subsystems"
  agents = {}
  agents['HVAC'] = HVAC()
  # Run simulation
  timestep = 0
  while timestep < args.timesteps:
    print "Timestep %d" % timestep
    if timestep % 10 == 0:
      curr_temp = random.randrange(68,72)
      curr_humidity = random.uniform(40.0,60.0)
      agents['HVAC'].communicate({'temp':curr_temp, 'humidity':curr_humidity})
    time.sleep(1)
    timestep += 1
  agents['HVAC'].shutdown()
  print "HVAC process state: %d" % agents['HVAC'].is_alive()

所以问题是，每当我在主循环中运行agents['HVAC'].communicate(x)时，我可以看到在其run循环中传递到HVAC子类的值(因此它正确打印接收到的值)。但是，该值永远不会被成功存储。

典型的输出是这样的:

Initializing subsystems
Timestep 0
Measured [68] [56.948675]
heating 1
heating 2
Timestep 1
heating 3
heating 4
Timestep 2
heating 5
heating 6

现实中，一旦Measured[68]出现，内部存储值就应该更新为输出68(不是加热1、加热2等)。所以实际上，暖通空调本身。__meas_temperature没有被正确更新。

---

编辑:经过一番研究，我意识到我不一定了解幕后发生的事情。每个子进程都使用自己的虚拟内存块进行操作，并且完全从以这种方式共享的任何数据中抽象出来，因此传入值将不起作用。我的新问题是，我不一定确定如何与多个进程共享全局值。

我正在看队列或JoinableQueue包，但我不一定确定如何将队列传递到我拥有的超类设置类型(特别是mpc.Process.__init__(self)调用)。

一个侧面的问题是，如果我可以有多个代理读取值的队列，而不把它拉出队列?例如，如果我想与多个代理共享temperature值，那么Queue是否可以实现此目的?

Pipe v Queue

Answer 1

假设您需要以下内容，这里有一个建议的解决方案:

• 一个集中的管理器/主进程，控制工人的生命周期
• 工作进程做一些独立的事情，然后报告结果给管理器和其他进程

在我展示它之前，为了记录，我想说一般来说，除非你是CPU限制，multiprocessing不是真正合适的，主要是因为增加的复杂性，你可能会更好地使用不同的高级异步框架。另外，你应该使用python 3，它好多了!

也就是说，multiprocessing.Manager使得使用multiprocessing很容易做到这一点。我已经在python 3中这样做了，但我不认为任何东西不应该在python 2中"正常工作"，但我还没有检查。

from ctypes import c_bool
from multiprocessing import Manager, Process, Array, Value
from pprint import pprint
from time import sleep, time

class Agent(Process):
    def __init__(self, name, shared_dictionary, delay=0.5):
        """My take on your Agent.
        Key difference is that I've commonized the run-loop and used
        a shared value to signal when to stop, to demonstrate it.
        """
        super(Agent, self).__init__()
        self.name = name
        # This is going to be how we communicate between processes.
        self.shared_dictionary = shared_dictionary
        # Create a silo for us to use.
        shared_dictionary[name] = []
        self.should_stop = Value(c_bool, False)
        # Primarily for testing purposes, and for simulating 
        # slower agents.
        self.delay = delay
    def get_next_results(self):
        # In the real world I'd use abc.ABCMeta as the metaclass to do 
        # this properly.
        raise RuntimeError('Subclasses must implement this')
    def run(self):
        ii = 0
        while not self.should_stop.value:
            ii += 1
            # debugging / monitoring
            print('%s %s run loop execution %d' % (
                type(self).__name__, self.name, ii))
            next_results = self.get_next_results()
            # Add the results, along with a timestamp.
            self.shared_dictionary[self.name] += [(time(), next_results)]
            sleep(self.delay)
    def stop(self):
        self.should_stop.value = True
        print('%s %s stopped' % (type(self).__name__, self.name))

class HVACAgent(Agent):
    def get_next_results(self):
        # This is where you do your work, but for the sake of
        # the example just return a constant dictionary.
        return {'temperature': 5, 'pressure': 7, 'humidity': 9}

class DumbReadingAgent(Agent):
    """A dumb agent to demonstrate workers reading other worker values."""
    def get_next_results(self):
        # get hvac 1 results:
        hvac1_results = self.shared_dictionary.get('hvac 1')
        if hvac1_results is None:
            return None
        return hvac1_results[-1][1]['temperature']
# Script starts.
results = {}
# The "with" ensures we terminate the manager at the end.
with Manager() as manager:
    # the manager is a subprocess in its own right. We can ask
    # it to manage a dictionary (or other python types) for us
    # to be shared among the other children.
    shared_info = manager.dict()
    hvac_agent1 = HVACAgent('hvac 1', shared_info)
    hvac_agent2 = HVACAgent('hvac 2', shared_info, delay=0.1)
    dumb_agent = DumbReadingAgent('dumb hvac1 reader', shared_info)
    agents = (hvac_agent1, hvac_agent2, dumb_agent)
    list(map(lambda a: a.start(), agents))
    sleep(1)
    list(map(lambda a: a.stop(), agents))
    list(map(lambda a: a.join(), agents))
    # Not quite sure what happens to the shared dictionary after
    # the manager dies, so for safety make a local copy.
    results = dict(shared_info)
pprint(results)