Networkx als Aufgabenwarteschlange?

Question

Ich habe eine gerichtete azyklische Grafik in networkx. Jeder Knoten stellt eine Aufgabe dar, und die Vorgänger eines Knotens sind Aufgabenabhängigkeiten (eine bestimmte Aufgabe kann erst ausgeführt werden, wenn ihre Abhängigkeiten ausgeführt wurden).

Ich möchte das Diagramm in einer asynchronen Aufgabenwarteschlange "ausführen", ähnlich wie celery bietet (damit ich Jobs nach ihrem Status abfragen, Ergebnisse abrufen usw.). Sellerie bietet nicht die Möglichkeit, DAGs zu erstellen (soweit ich weiß) und die Möglichkeit haben, zu einem task zu gehen, sobald alle Abhängigkeiten abgeschlossen sind, wäre entscheidend (eine DAG kann mehrere Pfade haben und selbst wenn eine Aufgabe langsam ist/Blockierung, kann es möglich sein, zu anderen Aufgaben weiterzugehen etc).

Gibt es einfache Beispiele, wie ich das erreichen oder vielleicht sogar networkx mit celery integrieren könnte?

Answer 1

Ich denke, diese Funktion kann helfen:

# 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)