1

我需要实现一个带有主进程的系统，该主进程管理执行其他任务的从进程。我有两种不同的从属类型，每个从属需要 6 个实例。我写了一些可以工作的东西，但是它会杀死每个进程并在任务完成时启动一个新进程。这是不可取的，因为产生新进程的成本很高。我更愿意让每个从站作为一个进程运行，并在它完成时得到通知，然后用新的输入再次运行它。

我当前的伪代码如下。它并不完美；我正在考虑它，因为我没有实际的代码。

# SlaveTypeB is pretty much the same.
class SlaveTypeA(multiprocessing.Process):
    def __init__(self, val):
        self.value = val
        self.result = multiprocessing.Queue(1)
        self.start()
    def run(self):
        # In real life, run does something that takes a few seconds.
        sleep(2)
        # For SlaveTypeB, assume it writes self.val to a file instead of incrementing
        self.result.put(self.val + 1)
    def getResult(self):
        return self.result.get()[0]


if __name__ == "__main__":
    MAX_PROCESSES = 6
    # In real life, the input will grow as the while loop is being processed
    input = [1, 4, 5, 6, 9, 6, 3, 3]
    aProcessed = []
    aSlaves = []
    bSlaves = []

    while len(input) > 0 or len(aProcessed) > 0:
        if len(aSlaves) < MAX_PROCESSES and len(input) > 0:
            aSlaves.append(SlaveTypeA(input.pop(0))
        if len(bSlaves) < MAX_PROCESSES and len(aProcessed) > 0 :
            bSlaves.append(SlaveTypeB(aProcesssed.pop(0))
        for aSlave in aSlaves:
            if not aSlave.isAlive():
                aProcessed = aSlave.getResult()
                aSlaves.remove(aSlave)
        for bSlave in bSlaves:
            if not bSlave.isAlive():
                bSlaves.remove(bSlave)

我怎样才能使 aSlaves 和 bSlaves 中的进程不会被杀死和重生。我想我可以使用管道，但我不确定如何判断进程何时完成阻塞而无需等待。

编辑 我使用管道重写了它，它解决了我无法保持进程运行的问题。仍然希望输入有关执行此操作的最佳方法的信息。我省略了 slaveB 部分，因为只有一种工作类型可以简化问题。

class Slave(Process)
    def __init__(self, id):
        # Call super init, set id, set idlestate = true, etc
        self.parentCon, self.childCon = Pipe()
        self.start()

    def run(self):
        while True:
            input = self.childCon.recv()
            # Do something here in real life
            sleep(2)
            self.childCon.send(input + 1)

   #def isIdle/setIdle():
       # Getter/setter for idle

   def tryGetResult(self):            
       if self.parentCon.poll():
           return self.parentCon.recv()
       return False

   def process(self, input):
       self.parentConnection.send(input)

if __name__ == '__main__'
    MAX_PROCESSES = 6
    jobs = [1, 4, 5, 6, 9, 6, 3, 3]
    slaves = []
    for int i in range(MAX_PROCESSES):
        slaves.append(Slave(i))
    while len(jobs) > 0:
        for slave in slaves:
            result = slave.tryGetResult()
            if result:
                # Do something with result
                slave.setIdle(True)
            if slave.isIdle():
                slave.process(jobs.pop())
                slave.setIdle(False)

编辑 2 知道了，请参阅下面的答案。

4

2 回答 2

0

创建两个队列？喜欢worktodoA并且worktodoB让您的员工在等待将某些东西放入队列时闲置，如果放在那里的项目让我们说“退出”，他们会退出吗？

否则你应该给tMCs 评论一个机会

于 2012-09-26T06:40:00.143 回答
0

看起来使用 SyncManager 是这种情况下的最佳选择。

class Master(SyncManager):
    pass

input = [1, 4, 5, 6, 9, 6, 6, 3, 3]
def getNextInput():
    # Check that input isn't empty first
    return input.pop()

if __name__ == "__main__":
    MAX_PROCESSES = 6
    Master.register("getNextInput", getNextInput)
    m = Master(('localhost', 5000))
    m.start()
    for i in range(MAX_PROCESSES):
        Slave()
    while True:
        pass

class Slave(Process):
    def __init__(self):
        multiprocessing.Process.__init__(self)
        self.start()
    def run(self):
        Master.register("getNextInput", getNextInput)
        m = Master(('localhost', 5000))
        m.connect()
        while True:
            input = m.getNextInput()
            # Check for None first
            self.process(input)
    def process(self):
        print "Processed " + str(input)
于 2013-10-08T19:57:13.527 回答