Combining Celery Gevent Pool with asyncio (asgiref)

Question

I have lots of async code which is used for websocket connectons. Idea is simple: fetch data from websocket, calculate some math and do lots of io.

 data = await websocket.recv()
 # Do some math
 await obj.save()
 await redis.publish(data)
 await api_client.send(data)
 # And so on

This works perfect in an independent container.

But now i received another task:

Make Celery worker which will "re-check" data from third party api once in 10 minutes and do all the same logic. So i have to put all this stuff into celery task. Currently i do something like this:

@app.task
def sync_with_api():
     for obj in objects_to_sync:
          asgiref.sync.async_to_sync(websocket_logic)(obj)

This works, but i do all of it in synchronous way, but we expect thousand of objects to sync. So i am wondering about performance.

I thought about using gevent pool and did this:

@app.task
def sync_objects():
     for obj in objects_to_sync:
          app.send_task("core.sync_task", args=[obj])

@app.task
def sync_task(obj):
     asgiref.sync.async_to_sync(websocket_logic)(obj)

From my perspective, this should works better. But i am not sure is it a good idea of combining gevent with asyncio because this is two different concurrency models. Does anyone had experience with problems like that?

Or i should just refactor original code and made two versions:

Async to work with websockets
Sync to work with gevent

Vladislav · Accepted Answer · 2023-12-15 16:06:13Z

0

I ended up with the solution:

async def sync_object(obj):
    await api_client.call()
    # some another logic

@app.task
def task_to_sync():
     objects = Objects.objects.all()
     loop = asyncio.get_event_loop()
     group = asyncio.gather(*[sync_objects(obj) for obj in objects])
     loop.run_until_complete(group)
     loop.close()

I have two celery containers. One for i/o background tasks and gevent as concurrency model. And second with prefork for common tasks and cpu bound. This task should be called once in 10 minutes, so this solution is good enough to handle it. So i call from prefork celery and it work good enough.

answered Dec 15, 2023 at 16:06

Vladislav

671 silver badge10 bronze badges

Sign up to request clarification or add additional context in comments.

Collectives™ on Stack Overflow

Combining Celery Gevent Pool with asyncio (asgiref)

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related