Networkx как очередь задач?

Question

У меня есть ориентированный ациклический график в networkx. Каждый узел представляет задачу, а предшественники узлов - зависимые от задачи (данная задача не может выполняться до тех пор, пока не будут выполнены ее «зависимости»).

Я хотел бы «выполнить» график в асинхронной очереди задач, аналогично тому, что предлагает celery (чтобы я мог опросить задания для их статуса, получить результаты и т. Д.). Celery не дает возможности создавать DAG (насколько мне известно) и иметь возможность перейти на task, как только все зависимости будут завершены, будет иметь решающее значение (у DAG может быть несколько путей и даже если одна задача медленная/блокирование, возможно, можно перейти к другим задачам и т. д.).

Есть ли какие-либо простые примеры того, как я мог бы достичь этого, или, возможно, интегрировать networkx с celery?

Answer 1

Я думаю, что эта функция может помочь:

# The graph G is represened by a dictionnary following this pattern: 
    # G = { vertex: [ (successor1: weight1), (successor2: weight2),... ] } 
    def progress (G, start): 
    Q = [ start ] # contain tasks to execute 
    done = [ ] # contain executed tasks 
    while len (Q) > 0: # still there tasks to execute ? 
     task = Q.pop(0) # pick up the oldest one 
     ready = True 
     for T in G:  # make sure all predecessors are executed 
      for S, w in G[T]: 
       if S == task and and S not in done:# found not executed predecessor 
       ready = False 
       break 
      if not ready : break 
     if not ready: 
      Q.appen(task) # the task is not ready for execution 
     else: 
      execute(task) 
      done.appen(task) # execute the task 
      for S, w in G[task]:# and explore all its successors 
       Q.append(S)