eine Aufgabe Laufen, nachdem alle Aufgaben abgeschlossen wurden

Question

ich eine Anwendung Ich schreibe die eine Reihe von Aufgaben parallel ausgeführt werden muss und dann eine einzelne Aufgabe mit den Ergebnissen aller Aufgaben ausführen:

@celery.task 
def power(value, expo): 
    return value ** expo 

@celery.task 
def amass(values): 
    print str(values) 

Es ist ein sehr konstruiertes und vereinfachtes Beispiel, aber hoffentlich kommt der Punkt gut rüber. Grundsätzlich habe ich viele Elemente, die durch power laufen müssen, aber ich möchte nur amass auf die Ergebnisse von allen Aufgaben ausführen. All dies sollte asynchron passieren, und ich brauche nichts zurück von der amass Methode.

Kann jemand in Sellerie einrichten, so dass alles asynchron ausgeführt wird und ein einziger Rückruf mit einer Liste der Ergebnisse aufgerufen wird, nachdem alles gesagt und getan ist?

ich Setup habe dieses Beispiel mit einem chord als Alexander Afanasiev laufen empfohlen:

from time import sleep 

import random 

tasks = [] 

for i in xrange(10): 
    tasks.append(power.s((i, 2))) 
    sleep(random.randint(10, 1000)/1000.0) # sleep for 10-1000ms 

callback = amass.s() 

r = chord(tasks)(callback) 

Leider ist in dem obigen Beispiel alle Aufgaben in tasks gestartet werden, nur dann, wenn die chord Methode aufgerufen wird. Gibt es eine Möglichkeit, dass jede Aufgabe einzeln gestartet werden kann und ich dann einen Rückruf zu der Gruppe hinzufügen kann, die ausgeführt wird, wenn alles fertig ist?

Answer 1

Hier ist eine Lösung, die für meine Zwecke gearbeitet:

tasks.py:

from time import sleep 

import random 

@celery.task 
def power(value, expo): 
    sleep(random.randint(10, 1000)/1000.0) # sleep for 10-1000ms 
    return value ** expo 

@celery.task 
def amass(results, tasks): 
    completed_tasks = [] 
    for task in tasks: 
     if task.ready(): 
      completed_tasks.append(task) 
      results.append(task.get()) 

    # remove completed tasks 
    tasks = list(set(tasks) - set(completed_tasks)) 

    if len(tasks) > 0: 
     # resend the task to execute at least 1 second from now 
     amass.delay(results, tasks, countdown=1) 
    else: 
     # we done 
     print results 

Use Case:

tasks = [] 

for i in xrange(10): 
    tasks.append(power.delay(i, 2)) 

amass.delay([], tasks) 

Was sollte diese tun ist Start alle Aufgaben so schnell wie möglich asynchron. Sobald sie alle in die Warteschlange eingereiht wurden, wird die Aufgabe amass ebenfalls in die Warteschlange gestellt. Die Sammelaufgabe wird sich so lange selbst aktualisieren, bis alle anderen Aufgaben abgeschlossen sind.

Answer 2

Sellerie hat plenty of tools für die meisten Workflows, die Sie sich vorstellen können.

Es scheint, dass Sie chord verwenden müssen. Hier ist ein Zitat von docs:

A chord is just like a group but with a callback. A chord consists of a header group and a body, where the body is a task that should execute after all of the tasks in the header are complete.

Answer 3

Die Antwort, dass @ alexander-Afanasiev haben Sie im Wesentlichen richtig sind: einen Akkord verwenden.

Ihr Code ist OK, aber tasks.append(power.s((i, 2))) führt die Teilaufgabe nicht aus, sondern fügt nur subtasks zu einer Liste hinzu. Es ist chord(...)(...) die eine, die so viele Nachrichten an den Broker senden wie Teilaufgaben, die Sie in tasks Liste definiert haben, plus eine weitere Nachricht für die Rückruf-Teilaufgabe. Wenn Sie chord aufrufen, wird es zurückgegeben, sobald es möglich ist.

Wenn Sie wissen möchten, wann der Akkord fertig ist, können Sie wie bei einer einzelnen Aufgabe unter Verwendung von r.ready() in Ihrem Sample nach Abschluss suchen.

Answer 4

einen Blick auf diese Schnipsel aus Ihrer Frage machen, sieht es aus wie Sie einen list als Akkord-Header sind vorbei, anstatt ein group:

from time import sleep 
import random 

tasks = [] 

for i in xrange(10): 
    tasks.append(power.s((i, 2))) 
    sleep(random.randint(10, 1000)/1000.0) # sleep for 10-1000ms 

callback = amass.s() 

r = chord(tasks)(callback) 

die list zu einem group Konvertieren in das Verhalten führen sollte Sie erwarten:

... 

callback = amass.s() 

tasks = group(tasks) 

r = chord(tasks)(callback)