Задачи сельдерея не отменяются

Question

Я запускаю несколько симуляций в виде задач через сельдерей (версия 2.3.2) из ​​django. Моделирование настраивается другой задачей:

В views.py:

result = setup_simulations.delay(parameters)
request.session['sim'] = result.task_id # Store main task id

In tasks.py:

@task(priority=1)
def setup_simulations(parameters):
    task_ids = []
    for i in range(number_of_simulations):
        result = run_simulation.delay(other_parameters)
        task_ids.append(result.task_id)
    return task_ids

После завершения начальной задачи (setup_simulations) я пытаюсь отозвать задачи моделирования следующим образом:

main_task_id = request.session['sim']
main_result = AsyncResult(main_task_id)
# Revoke sub tasks
from celery.task.control import revoke
for sub_task_id in main_result.get():
    sub_result = AsyncResult(sub_task_id); sub_result.revoke() # Does not work
    # revoke(sub_task_id) # Does not work neither

Когда я смотрю вывод "python manage.py celeryd -l info", задачи выполняются так, как будто ничего не произошло. Есть идеи у кого-нибудь, что могло пойти не так?

Answer 1

Как вы упоминаете в комментарии, revoke - это команда удаленного управления, поэтому в настоящее время она поддерживается только транспортами amqp и redis.

Вы можете сделать это самостоятельно, сохранив отозванный флаг в своей базе данныхНапример:

from celery import states
from celery import task
from celery.exceptions import Ignore

from myapp.models import RevokedTasks


@task
def foo():
    if RevokedTasks.objects.filter(task_id=foo.request.id).count():
        if not foo.ignore_result:
            foo.update_state(state=states.REVOKED)
        raise Ignore()

Если ваша задача работает на какой-либо модели, вы даже можете сохранить флаг в ней.