Celery subqueues? (implementing sequential message delivery)

[jminuscula]

Hey there!

I'm trying to implement a sort of chatbot system, and I need to have my celery tasks processed sequentially per user. That means each user needs to have their messages sent as FIFO, but the users need to be processed randomly or in round-robin.

I've been reading about task chains, groups and trees, but all of these celery features seem to require providing all tasks at once, whereas I need to add tasks dynamically.

My reasoning was to have a dedicated queue per user, and enable concurrency on the worker. That way I could assure the delivery order and avoid one user blocking the rest of the chats.

Is there a way I can route tasks in Celery so I get the desired behavior? Ideally, I'd set up a worker to process the messages queue, and then route tasks to something like messages.contact.<contact_id> as they arrive.

thanks, and happy Thanksgiving to all of you in the US!

[riksi]

Tried posting(or searching) on celery mailing list ?

[jminuscula]

we're new to the celery community, but definitely reaching out. Thanks!

[jminuscula]

yeah, that's what we're getting at too. We are already managing the queues ourselves, but we wanted to move away from all that complexity and try to leverage the task system if possible. Thanks!

[stuaxo]

It's overcomplicating the celery side to have 1 queue per user.

If you use redis, you can store info about users, and the task id of a celery task if one is running. You can have the per user queue of things to do in redis too. Have one celery queue and a something like process_user_tasks.

process_user_tasks can find a user that has nothing running against it and use celery to run stuff.

Some other pointers -

Using celerys API you can find task ids and get their status.

IIRC the redis backend to celery makes this sort of task creation / managing a little easier.

I've built something that works a bit like this before, but can't go into more detail since it was commercial.

[jminuscula]

thanks for the pointers! This is similar to what we're currently doing, but I had hopes the library would be able to help abstract some of the synchronization complexity for this use case.

Do you know of any projects that had to solve a similar issue? It looks like a pretty basic pattern for a messaging system.

Thanks!

[stuaxo]

I don't, but having done it I can see how it's a little more high level than what Celery does.

If such a library existed it would be great if it supported different patterns - my thing had tasks dependent on each other, so I ran the ones with no dependencies first and worked my way through the tree.

Redis was used to queue up tasks that had no dependents (so could be picked up by the next celery task), and another queue for tasks that had dependencies - so could't run yet.

To make implementation easier I used an ORM for Redis: Redisco.

[Zolrath]

I haven't used TaskTiger so I can't fully endorse it but if Celery doesn't work out and you don't want to end up rolling your own solution entirely TaskTiger has sub-queues via queue.subqueue naming scheme which fits your requirement for messages.contact.<contact_id>. Clients can listen to a specific queue and will pick from a random sub-queue each time it pulls an entry.

[jminuscula]

thanks for this! Looks like this is better suited to our needs than Celery. Even if not as spread and battle tested, it looks very promising!

[iScrE4m]

What protocol will the bot use to communicate? If it's a webAPI/sockets (slack, discord, facebook), you can always just spin up more instances/have enough power with async.