DAG并行处理

Question

我正在努力弄清楚如何并行处理一个有向无圈图。每个节点只有在其所有输入节点都已预先处理后才能“执行”。假设有一个具有以下接口的类Task：

class Task(object):
    result = None
    def inputs(self):
        ''' List all requirements of the task. '''
        return ()
    def run(self):
        pass

除了一种方法之外，我想不出一种方法来处理这个结构可以异步表示的图形，同时使用最大数量的工作人员。

我认为，通过为每个任务创建一个线程，等待所有输入被处理，就可以实现最优处理。但是，立即为每个任务生成一个线程，而不是连续地生成一个线程(即，当任务准备被处理时)，对我来说这听起来不是一个好主意。

import threading
class Runner(threading.Thread):
    def __init__(self, task):
        super(Runner, self).__init__()
        self.task = task
        self.start()
    def run(self):
        threads = [Runner(r) for r in self.task.inputs()]
        [t.join() for t in threads]
        self.task.run()

有没有一种更理想地模仿这种行为的方法？此外，这种方法目前还没有实现一种限制一次运行任务数量的方法。

Answer 1

有一个主线程将项目推送到队列中，一旦它们准备好被处理。然后让一组工作人员监听要处理的任务的队列。(Python在 module中提供了一个同步队列，在Python3中重命名为小写queue )。

主服务器首先创建一个从依赖项到依赖任务的映射。没有任何依赖关系的每个任务都可以进入队列。每次任务完成时，主程序都会使用字典来确定存在哪些依赖任务，如果现在满足了它们的所有依赖项，则将它们放入队列中。

Answer 2

芹菜(http://www.celeryproject.org/)是Python的主要任务管理工具。它应该能帮到你的。