Question about deferred and recurring tasks

[paulzakin]

Hello!

Quick question here:

Let's say I have a recurring task, recurring_task, that ultimately (async) defers a hundred of the task geocode_address, which take 100 seconds. What I've found is the recurring_task is deferred while the worker "works through" all 100 instances of the task geocode_address.

Is that the expected behaviour?

If it is, what is the recommended remedy if you want the recurring task to run as well?

• Does it have something to do with queues?
• Or do I need to run multiple instances of the worker?
• Perhaps is there a system of priority to give the recurring task a higher priority?
• Or should I just have tasks that take less time, like 10 seconds vs. 100

And here is a code snippet:

@app.task(name="geocode_address")
async def geocode_address(pk: int) -> None:
    await asyncio.sleep(100) # Make sure it takes a long time

async def assign_addresses_to_geocode() -> None:
  async for i in range(1, 100):
          await geocode_address.defer_async(pk=i)

@app.periodic(cron="*/5 * * * *")
@app.task(name="recurring_task")
async def recurring_task(timestamp: int) -> None:
    logger.info("Starting: recurring_task")
    await assign_addresses_to_geocode()
    logger.info("Completed: recurring_task")

[ewjoachim]

The general rule is that when a worker decides what's the next job to tackle, it selects, among the available ones, the one with the lowest (incremental) id. This means if you have 100 long tasks deferred and one short task waiting after that, in general it will get executed after the 100 tasks (namely, there's no way to tell if a task will be long or short, and we try to have tasks be executed in a somewhat predictable order).

Now there are still some things worth mentioning. Before diving in the details, there's one point that needs addressing. I might be wrong, but if you create 100 tasks that all take 100 seconds, I see 3 possible outcomes:

• Your workers can process this before the next recurring job gets deferred, all is well
• Your worker can't process this now, but it's a temporary spike, they'll will eventually manage to work through the whole thing.
• Your workers will never be able to work through the queue faster than it's filled. You have a bigger design problem (or not enough workers)

Here are some things that may help situation 3 (and potentially situation 2):

• Queuing locks (the title of the page is aptly named "How to ensure tasks don’t accumulate in the queue"
• Locks are used in the algorithm that decides which job gets executed next
• Queues, indeed may help, but in conjunction of having multiple workers: given the rules stated above, let's say recurring_task are posted to a queue named recurring_task_queue and geocode_address to geocode_address_queue. If you have at least 2 workers, including at least one that listens to recurring_task_queue and not to geocode_address_queue, then it's likely that this worker will reach your recurring_task before the other worker processes all the geocode_address jobs.
• There are existing tickets for other strategies: priorities (and Ability to enqueue a task ahead of already-enqueued tasks? #698). Also we could imagine other things (for example, the concurrency feature doesn't let you define sub-workers linked to different queues. Maybe it should ?)
• Another way that could work : run a recurring task once a day, that defers all your recurring_task jobs at once but with scheduled_at argument. This way, when it reaches the said time, it's automatically the oldest task, and it gets executed first.
  Hope it helps ?

If you think with the current rules, the way it works is wrong, or you think there should be an obvious way to solve your problem, it might be interesting (but I'm not spending a lot of time, if any, developing new features for this lib so any evolution needs to come from other contributors)

[paulzakin]

Wow, great response! Yes, this is more than enough!