Sellerie - Verkettung Gruppen und Teilaufgaben. -> out-of-order execution

Wenn ich etwas wie das folgende

group1 = group(task1.si(), task1.si(), task1.si())
group2 = group(task2.si(), task2.si(), task2.si())

workflow = chain(group1, group2, task3.si())

Die intuitive interpretation ist, dass task3 sollte nur ausgeführt werden, wenn alle Aufgaben in der Gruppe 2 abgeschlossen haben.

In der Realität, Aufgabe 3 ausgeführt, während Gruppe1 hat begonnen, ist aber noch nicht abgeschlossen noch.

Was mache ich falsch?

  • Ein update auf diese neue version von Sellerie?
InformationsquelleAutor w-- | 2013-02-27
  1. 18

    Ich habe das gleiche Problem mit Sellerie, versucht zu haben einen workflow, bei dem der erste Schritt "spawn eine million Aufgaben". Versucht Gruppen von Gruppen, Arbeitsschritte, irgendwann mein step2 kicks off, bevor Sie Schritt 1 über.

    Lange Geschichte kurz ich vielleicht eine Lösung gefunden haben mit der Verwendung von Akkorden und einem dämlichen finisher:

    @celery.task
def chordfinisher( *args, **kwargs ):
  return "OK"

    Nichts zu tun viel, aber es ermöglicht mir, dies zu tun:

    tasks = []
for id in ids:
    tasks.append( mytask.si( id ) )
step1 = chord( group( tasks ), chordfinisher.si() )

step2 = ...

workflow = chain( step1, step2 )

    Ursprünglich wollte ich Schritt1 in einer Teilaufgabe, aber aus dem gleichen Grund wie vermutet, wird die Aktion für den Aufruf einer Gruppe endet, die Aufgabe gilt als beendet, und mein workflow bewegt sich auf...

    Wenn jemand etwas besser, ich bin interessiert!

    • Hi,das ist ziemlich viel, was ich am Ende machen. Eine Sache im Auge zu behalten, müssen Sie die dumbfinisher auf das Ergebnis der Gruppe Ausführung. ansonsten, wenn alles in der Gruppe fehl, werden die Kette wird dabei nicht halt bei Schritt 1. (dies kann oder kann nicht sein, was Sie wollen)
    InformationsquelleAutor Jonathan Adami
  2. 17

    So wie es sich herausstellt, in Sellerie-Sie können die Kette in zwei Gruppen zusammen.

    Ich vermute, dass dies ist, weil die Gruppen verkettet mit Aufgaben automatisch zu einem Akkord

    --> Sellerie docs: http://docs.celeryproject.org/en/latest/userguide/canvas.html

    Verkettung eine Gruppe zusammen mit einer anderen Aufgabe automatisch aktualisieren
    es werden ein-Akkord:

    Rückkehr der Gruppen eine übergeordnete Aufgabe. Bei der Verkettung von zwei Gruppen zusammen, vermute ich, dass bei der ersten Gruppe abgeschlossen ist, wird der Akkord beginnt die callback "Aufgabe". Ich vermute, diese "Aufgabe" ist eigentlich die "übergeordnete Aufgabe" der zweiten Gruppe. Ich vermute, dass diese übergeordnete Aufgabe abgeschlossen, sobald es fertig ist startet alle Teilaufgaben innerhalb der Gruppe und als ein Ergebnis das nächste Element nach der 2. Gruppe ausgeführt wird.

    Dies zu demonstrieren, hier einige Beispiel-code. Sie müssen bereits über eine laufende Sellerie Instanz.

    # celery_experiment.py

from celery import task, group, chain, chord
from celery.signals import task_sent, task_postrun, task_prerun

import time
import logging

import random
random.seed()

logging.basicConfig(level=logging.DEBUG)

### HANDLERS ###    
@task_prerun.connect()
def task_starting_handler(sender=None, task_id=None, task=None, args=None, kwargs=None, **kwds):    
    try:
        logging.info('[%s] starting' % kwargs['id'])
    except KeyError:
        pass

@task_postrun.connect()
def task_finished_handler(sender=None, task_id=None, task=None, args=None, kwargs=None, retval=None, state=None, **kwds):
    try:    
        logging.info('[%s] finished' % kwargs['id'])
    except KeyError:
        pass


def random_sleep(id):
    slp = random.randint(1, 3)
    logging.info('[%s] sleep for %ssecs' % (id, slp))
    time.sleep(slp)

@task()
def thing(id):
    logging.info('[%s] begin' % id)
    random_sleep(id)
    logging.info('[%s] end' % id)


def exec_exp():
    st = thing.si(id='st')
    st_arr = [thing.si(id='st_arr1_a'), thing.si(id='st_arr1_b'), thing.si(id='st_arr1_c'),]
    st_arr2 = [thing.si(id='st_arr2_a'), thing.si(id='st_arr2_b'),]
    st2 = thing.si(id='st2')
    st3 = thing.si(id='st3')
    st4 = thing.si(id='st4')

    grp1 = group(st_arr)
    grp2 = group(st_arr2)

    # chn can chain two groups together because they are seperated by a single subtask
    chn = (st | grp1 | st2 | grp2 | st3 | st4)

    # in chn2 you can't chain two groups together. what will happen is st3 will start before grp2 finishes
    #chn2 = (st | st2 | grp1 | grp2 | st3 |  st4)

    r = chn()
    #r2 = chn2()
    • Vielen Dank für diese. Leider für mich, meinen workflow nicht erlauben würde, mich zu verwenden, eine 'relevante' Aufgabe zwischen den Gruppen. So landete ich erstellen eine gefälschte Aufgabe def fake_celery_task(): pass zu laufen, zwischen den Gruppen...
    • in diesem Szenario, wenn sich eine Gruppe findet 2 Aufgaben und gibt die Ergebnisse zurück zu dem Akkord, den ich gehe davon aus, dass der Akkord wird die Ergebnisse in eine willkürliche Reihenfolge
    InformationsquelleAutor w--
Schreibe einen Kommentar