AsyncResult.then doesn't work

[slav0nic]

Example from the doc doesn't work

➤ celery report

software -> celery:4.0.0 (latentcall) kombu:4.0.0 py:3.5.2+
            billiard:3.5.0.2 py-amqp:2.1.1
platform -> system:Linux arch:64bit, ELF imp:CPython
loader   -> celery.loaders.default.Loader
settings -> transport:amqp results:disabled

Steps to reproduce

example from the docs http://docs.celeryproject.org/en/stable/whatsnew-4.0.html#asyncresult-then-on-success-on-error (it's also have error in the second line)

import gevent.monkey
gevent.monkey.patch_all()

import time
from celery import Celery

app = Celery(broker='amqp://', backend='rpc')

@app.task
def add(x, y):
    return x + y

def on_result_ready(result):
    print('Received result for id %r: %r' % (result.id, result.result,))

add.delay(2, 2).then(on_result_ready)

time.sleep(3)

Expected behavior

see result from on_result_ready callback

Actual behavior

callback is not firing

also, will be fine have some examples for twisted/asyncio

[ask]

Works fine for me here, but the example is a bit confusing since you cannot actually use it like this.

Since you're patching for gevent in the module, it means you have to use -P gevent in the worker, as the prefork pool doesn't work with the patches enabled.

$ celery worker -A test -P gevent -l info

If you want to use prefork as the worker, but gevent in the client you would have to split these up:

# proj/celery.py
from celery import Celery
app = Celery(broker='amqp://', backend='rpc')

@app.task
def add(x, y):
    return x + y

# proj/client.py  - name of this module irrelevant, can be anywhere just not imported in the worker.
import gevent.monkey
gevent.monkey.patch_all()
import time

from .celery import add

def on_result_ready(result):
    print('Received result for id %r: %r' % (result.id, result.result,))

add.delay(2, 2).then(on_result_ready)

time.sleep(3)  # < run the event loop for a bit.

[slav0nic]

tnx for info
maybe will be fine add some info about requirements/limitations like supported result backends, worker types etc
so, my main usecase for this is use this in twisted client, is this possible (what about basic callback example for asyncio/twisted?) ?

[wintamute]

I agree with @slav0nic, having some more information and examples would be nice.
I'm using send_task() from a different machine via aiohttp/asyncio, so the example with gevent doesn't work for me. However, using the asyncio event loop, the callback is never called.
Both worker and client use amqp as broker with rpc backend.

[slav0nic]

example from ask comment works fine for me (tested with broker='amqp://', backend='rpc' with default worker)
but looks like client side still required gevent/eventlet or something like eventlet reactor for twisted :(

[ask]

It doesn't support asyncio/twisted/tornado yet, for that to work the result backend itself needs to use async I/O (e.g. the redis backend would need to use an asyncio redis client).

You could possibly have asyncio/twisted/tornado use the gevent or eventlet event loops, I think there are adapters that make that possible, but not exactly the optimal solution.

I'm hoping we can accomplish asyncio support for 5.0, that's the main reason for going Python 3.5+ only

[wintamute]

Yes, I noticed that while browsing the sources.

So what is the plan for 5.0? Switch to asyncio only? Or still support tornado/twisted/... with adapters? The wiki is a bit outdated it seems and I couldn't find any other details. Any pointers?

I guess asyncio support requires quite some work besides the backends.
Do you need/want any help?
I added asyncio support for some internal applications of a client of mine, so I'm not completely new to this ;)

[ask]

It's definitely not a small undertaking... :)

We need this for asyncio:

• amqp client
  • publishing
  • operations, e.g. declaring queues, exchanges etc
  • consuming†
• redis client
• kombu publishing messages
• kombu consuming messages†
• prefork pool†
• result backends

† The worker is already using a homegrown event loop, largely using the same API as the core asyncio event loop, and these parts are already using that so should be compatible already.

One of the big benefits with asyncio is that we will get compatibility with twisted/tornado etc for free since they already have, or will get adapters. I have already been able to use tornado libraries with the Celery event loop, that took me like a day to accomplish, so doing the same for asyncio must be possible.

Definitely want help! I guess we need to plan a bit more, I don't even know if there are redis clients already available for example.

[wintamute]

Yes, it's a lot of code and a pretty big codebase, I'll have to get a better understanding of it first.

Regarding clients, a lot a asyncio supported libraries can be found at https://github.com/aio-libs.
Redis is on the list, but I don't have any experience with it yet, but the project is pretty active.

I'm using the Elasticsearch client for my work, which works great and is basically a drop in replacement for the official one, I basically just had to add async/await to existing code.
Everything on this list seems pretty mature and well tested.

I'm also using this asyncio amqp client, so far it works quite fine for my uses cases (mainly notifications via 'topic' exchanges), but it seems it's not as widely used as the ones I mentioned before.

Some planning would definitely help, break it down to some smaller parts, collect ideas etc.

[jxskiss]

For guys want to retrieve result within Tornado/asyncio when using redis backend, I have created a gist to show how to make it happen, thanks to celery's AsyncResult and aioredis, it's totally asyncronous.
Link: https://gist.github.com/jxskiss/aadf53659837b2052145cfc1c37cc8ea

[auvipy]

@jxskiss you could send a pr for adding this into celery