Python Sellerie - wie auf alle Teilaufgaben in Akkord warten

Question

Ich bin Unit-Test Sellerie Aufgaben. Ich habe Kettenaufgaben, die auch Gruppen haben, also entsteht ein Akkord.

Der Test sollte wie folgt aussehen:

• Lauf Sellerie Aufgabe (Verzögerung)
• warten Aufgabe und alle Unteraufgaben
• assert

Ich habe versucht, die folgenden:

def wait_for_result(result): 
    result.get() 
    for child in result.children or list(): 
     if isinstance(child, GroupResult): 
      # tried looping over task result in group 
      # until tasks are ready, but without success 
      pass 
     wait_for_result(child) 

Dies schafft eine de adlock, chord_unlock wird für immer wiederholt. Ich bin nicht an Aufgabenergebnissen interessiert. Wie kann ich warten, bis alle Teilaufgaben beendet sind?

Answer 1

Haben Sie versucht, Akkord + Rückruf?

http://docs.celeryproject.org/en/latest/userguide/canvas.html#chords

>>> callback = tsum.s() 
>>> header = [add.s(i, i) for i in range(100)] 
>>> result = chord(header)(callback) 
>>> result.get() 
9900 

Answer 2

Obwohl dies eine alte Frage, ich wollte nur teilen, wie ich von dem Deadlock Problem losgeworden, falls es jemanden hilft.

Wie die Sellerie Logs sagt, nie get() innerhalb einer Aufgabe verwenden. Dies wird in der Tat eine Sackgasse schaffen.

Ich habe eine ähnliche Reihe von Sellerie Aufgaben, die Kette von Gruppenaufgaben enthält, so dass es ein Akkord. Ich rufe diese Aufgaben mit Tornado auf, indem ich eine HTTP-Anfrage mache. Also, was ich tat, war so etwas wie dieses:

@task 
def someFunction(): 
    .... 


@task 
def someTask(): 
    .... 


@task 
def celeryTask(): 
    groupTask = group([someFunction.s(i) for i in range(10)]) 

    job = (groupTask| someTask.s()) 

    return job 

Wenn celeryTask() durch Tornado genannt wird, wird die Kette starten Ausführung & die UUID someTask() wird in job statt. Es wird wie etwas aussehen

AsyncResult: 765b29a8-7873-4b28-B05C-7e19c33e950c

Diese UUID zurückgegeben wird und die celeryTask() verlässt, bevor auch die Kette (im Idealfall) beginnt mit der Ausführung, damit Raum verlassen für ein anderer Prozess zum Ausführen.

Ich habe dann die Tornado-Ebene verwendet, um den Status der Aufgabe zu überprüfen. Details zur Tornado-Schicht finden Sie in dieser stackoverflow question