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

几年后，再次回顾这个问题时，我建议所有走过这条路的人都看看拓扑排序算法。在算法的每一步中，您都会查看图中入度为零 (0) 的所有节点。所有这样的节点可以被并行处理。以这种方式限制并行处理的节点数量变得非常简单，因为您可以简单地决定不将所有这些节点推送到工作队列中。从图中删除完全处理的节点，这应该留下入度为零 (0) 的新节点，除非图中存在循环。