Long-running tasks with async server

Question

I think everyone knows what to do with long-running tasks in django: use celery and relax. But what if I want to get benefits of the websockets with aiohttp (or tornado)?

Let's say I have very CPU bound task which can take from a couple of seconds till multiple (5-10) minutes. It looks like pretty good idea to handle this task in websocket loop and notify user about the progress. No ajax requests, very fast response for short tasks.

async def websocket_handler(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)

    async for msg in ws:
        if msg.tp == aiohttp.MsgType.text:     
            answer_to_the_ultimate_question_of_life_the_universe_and_everything =\
                long_running_task(msg.data, NotificationHelper(ws))
            ws.send_str(json.dumps({
                'action': 'got-answer',
                'data': answer_to_the_ultimate_question_of_life_the_universe_and_everything,
            }))
    return ws

But on the other hand, CPU-bound task served in such way blocks entire thread as I understand. If I have 10 workers and 11 clients who wants to use application, 11th client won't be served until the 1st client's task is done.

Maybe, I should run tasks which look big in celery and tasks which look small in the main loop?

So, my question: is there any good design pattern for serving long-running tasks with async server?

Thanks!

Answer 1

Just run your long-running CPU-bound task by loop.run_in_executor() and send progress notifications by loop.call_soon_threadsafe().

If your job is not CPU but IO bound (sending emails for example) you may create a new task by loop.create_task() call. It looks like spawning new thread.

If you cannot use fire-and-forget approach you need to use persistent message broker like RabbitMQ (there is https://github.com/benjamin-hodgson/asynqp library for communicating with Rabbit in asyncio way).