2

I am using 2 celery instances.

The configuration of the first instance is:

app = Celery('tasks', broker='amqp://guest@localhost//')

app.conf.update(
    CELERY_RESULT_BACKEND='amqp',
    CELERY_TASK_RESULT_EXPIRES=18000,
    CELERY_ACCEPT_CONTENT=['json'],
    CELERY_TASK_SERIALIZER='json',

    CELERY_ROUTES={
        'task_polling': {
            'queue': 'task_polling_queue'
        },
        'save_shell_task': {
            'queue': 'save_shell_task_queue'
        },
        'save_validation_chain_task': {
            'queue': 'save_validation_chain_task_queue'
        },
        'do_work': {
            'queue': 'do_work_queue'
        },
        'send_mail': {
            'queue': 'send_mail_queue'
        }
    },
)

@shared_task(name='do_work', ignore_result=True)
def do_work(_serialized_task):
    for bla in blala:
        do_something()  
        is_canceled = send_task('save_validation_chain_task', [], 
                                {'_params': my_params}).get() == True

Launched with the following command:

celery -A tasks worker --loglevel=info -Q do_work_queue,send_mail_queue

and the second one:

app = Celery()

app.conf.update(
    CELERY_RESULT_BACKEND='amqp',
    CELERY_TASK_RESULT_EXPIRES=18000,
    CELERY_ACCEPT_CONTENT=['json'],
    CELERY_TASK_SERIALIZER ='json',
    CELERYBEAT_SCHEDULE={
        'periodic_task': {
            'task': 'task_polling',
            'schedule': timedelta(seconds=1),
        },
    },
    CELERY_ROUTES={
        'task_polling': {
            'queue': 'task_polling_queue'
        },
        'save_shell_task': {
            'queue': 'save_shell_task_queue'
        },
        'save_validation_chain_task': {
            'queue': 'save_validation_chain_task_queue'
        },
        'do_work': {
            'queue': 'do_work_queue'
        },
        'send_mail': {
            'queue': 'send_mail_queue'
        }
    },
)


@shared_task(name='save_shell_task', ignore_result=True)
def save_shell_task(_result):
    ShellUpdate(_json_result=_result).to_db()


@shared_task(name='save_validation_chain_task', ignore_result=False)
def save_validation_chain_task(_result):
    return ValidationChainUpdate(_json_result=_result).to_db()

This one is launched with:

celery -A my_prog worker -B --concurrency=1 -P processes -Q task_polling_queue,save_shell_task_queue,save_validation_chain_task_queue

The issue is that the send_task(...).get() is not receiving the result. The program is waiting in a loop.

It seems that celery doesn't received the queue result or doesn't waiting the right queue result. The issue is certainly due to the -Q parameter. Do you have any idea where could be the issue in the configuration?

thank you

EDIT: The global idea is to have two celery instances with different source codes. That why I decided to enumerate the queues to remove the dependency. I really think this is why the result is not consumed as I can't specify the queue result in the command as this one has a variable name for each result (queue created for each result dynamically by celery). Any solution to keep two different source codes for the celery instance is good for me. I would like to avoid to use another result backend as the volumetric is very low.

2
  • Just out of curiosity if you did 'my_prog.save_validation_chain_task_queue' instead of just 'save_validation_chain_task_queue' Nov 14, 2014 at 19:06
  • I corrected the call. It was save_validation_chain_task. The worker is well called but the result is not received. If I had the namespace my_prog, the worker is not called at all.
    – Julio
    Nov 16, 2014 at 21:34

1 Answer 1

4
+100

You have correct setup and config. The only problem is you have set ignore_result for do_work task.

@shared_task(name='do_work', ignore_result=True)

When set this, even though your task is completed by the worker the state of the task will be always PENDING. Thats why when you are doing a .get() on that task, it never completes the execution of the that statement.

Since you are accepting only json

CELERY_ACCEPT_CONTENT=['json'],
CELERY_TASK_SERIALIZER ='json',

you also need to set

CELERY_RESULT_SERIALIZER = 'json',

in both of your configs.

Note:

In your case, your are doing a .get() on a task INSIDE another task. This should be avoided. For now it will work fine. From celery 3.2 it will raise and error instead of warining.

You can use chain to prevent launching synchronous subtasks if they fit your need. When chain is called, it returns async_result object which has parent property. For example

task1 = add.s(1,2)
task2 = add.s(5)
task3 = add.s(10)

result = chain(task1 | task2 | task3)()

result.revoke(terminate=True)                 # revokes task3
result.parent.revoke(terminate=True)          # revokes task2
result.parent.parent.revoke(terminate=True)   # revokes task1

If they don't fit, you can use signals to call some other tasks/functions. Here is a simple example(I have not tested this code).

from celery.signals import task_success

@app.task
def small_task():
      print('small task completed')


@app.task
@task_success.connect(sender=small_task)
def big_task(**kwargs):
      print('called by small_task. LOL {0}'.format(kwargs))
8
  • The CELERY_RESULT_SERIALIZER setting fixed the issue. Thank you. Regarding the .get() inside a task, the chaining is good if you have a small context to exchange between each worker. In my case I have a worker with a big context. The get is only required to know if the worker should continue to work on this big task or not. This is a workaround as it is not possible to cancel a ongoing task with celery. Do you have any better suggestion to be compliant with 3.2?
    – Julio
    Nov 17, 2014 at 15:42
  • I mostly use chain of tasks with immutable signature. You can use signals for your case. See updated method. Nov 17, 2014 at 17:42
  • Thank you but I really don't know how to handle a stop command to your big_task. My understanding from your celery link is that I should save the context in memcached and cut the big task in small one chained. In my case, cutting the big task is very difficult to setup. Do you have another suggestion to setup a simple stop command?
    – Julio
    Nov 17, 2014 at 19:16
  • If you are just worried about stopping tasks, you can use chain. Added a simple example to stop chain. Nov 18, 2014 at 6:51
  • 1
    It should work with the python dictionary also. @sattva_venu Sep 11, 2020 at 4:03

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.