Python celery - как подождать все подзадачи в аккорде

Question

Я занимаюсь тестированием сельдерея. У меня есть целые задачи, которые также имеют группы, поэтому приведен аккорд.

Испытание должно выглядеть следующим образом:

• запустить сельдерея задачи (задержки)
• ждут задачи и подзадачи
• Assert

Я попытался следующие:

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) 

Это создает de adlock, chord_unlock повторяется навсегда. Меня не интересуют результаты задачи. Как я могу подождать завершения всех подзадач?

Answer 1

Вы пробовали аккорд + обратный вызов?

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

Хотя это старый вопрос, я просто хотел рассказать, как я избавилась от вопроса тупиковой, только в случае, если это поможет кому-то.

Как и сельдерейные журналы, никогда не используйте get() внутри задачи. Это действительно создаст тупик.

У меня есть аналогичный набор задач сельдерея, который включает в себя цепочку групповых заданий, поэтому делает его аккордом. Я вызываю эти задачи, используя торнадо, делая запрос HTTP. Так что я сделал что-то вроде этого:

@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 

Когда celeryTask() вызывается торнадо, цепь начинает исполняться, & UUID из someTask() будет проходить в job. Это будет выглядеть как

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

Этот UUID возвращается и celeryTask() выходы еще до того, цепочка начинает выполняться (в идеале), следовательно, оставляя место для другой процесс для запуска.

Затем я использовал слой торнадо для проверки состояния задачи. Подробная информация о слое торнадо находится в этом stackoverflow question