Understanding celery task prefetching

Question

I just found out about the configuration option CELERYD_PREFETCH_MULTIPLIER (docs). The default is 4, but (I believe) I want the prefetching off or as low as possible. I set it to 1 now, which is close enough to what I'm looking for, but there's still some things I don't understand:

1. Why is this prefetching a good idea? I don't really see a reason for it, unless there's a lot of latency between the message queue and the workers (in my case, they are currently running on the same host and at worst might eventually run on different hosts in the same data center). The documentation only mentions the disadvantages, but fails to explain what the advantages are.

2. Many people seem to set this to 0, expecting to be able to turn off prefetching that way (a reasonable assumption in my opinion). However, 0 means unlimited prefetching. Why would anyone ever want unlimited prefetching, doesn't that entirely eliminate the concurrency/asynchronicity you introduced a task queue for in the first place?

3. Why can prefetching not be turned off? It might not be a good idea for performance to turn it off in most cases, but is there a technical reason for this not to be possible? Or is it just not implemented?

4. Sometimes, this option is connected to CELERY_ACKS_LATE. For example. Roger Hu writes «[…] often what [users] really want is to have a worker only reserve as many tasks as there are child processes. But this is not possible without enabling late acknowledgements […]» I don't understand how these two options are connected and why one is not possible without the other. Another mention of the connection can be found here. Can someone explain why the two options are connected?

Solution 1

1. Prefetching can improve the performance. Workers don't need to wait for the next message from a broker to process. Communicating with a broker once and processing a lot of messages gives a performance gain. Getting a message from a broker (even from a local one) is expensive compared to the local memory access. Workers are also allowed to acknowledge messages in batches

2. Prefetching set to zero means "no specific limit" rather than unlimited

3. Setting prefetching to 1 is documented to be equivalent to turning it off, but this may not always be the case (see https://stackoverflow.com/a/33357180/71522)

4. Prefetching allows to ack messages in batches. CELERY_ACKS_LATE=True prevents acknowledging messages when they reach to a worker

Solution 2

Old question, but still adding my answer in case it helps someone. My understanding from some initial testing was same as that in David Wolever's answer. I just tested this more in celery 3.1.19 and -Ofair does work. Just that it is not meant to disable prefetch at the worker node level. That will continue to happen. Using -Ofair has a different effect which is at the pool worker level. In summary, to disable prefetch completely, do this:

1. Set CELERYD_PREFETCH_MULTIPLIER = 1
2. Set CELERY_ACKS_LATE = True at a global level or task level
3. Use -Ofair while starting the workers
4. If you set concurrency to 1, then step 3 is not needed. If you want a higher concurrency, then step 3 is essential to avoid tasks getting backed up in a node that could be run long running tasks.

Adding some more details:

I found that the worker node will always prefetch by default. You can only control how many tasks it prefetches by using CELERYD_PREFETCH_MULTIPLIER. If set to 1, it will only prefetch as many tasks as the number of pool workers (concurrency) in the node. So if you had concurrency = n, the max tasks prefetched by the node will be n.

Without the -Ofair option, what happened for me was that if one of the pool worker processes was executing a long running task, the other workers in the node would also stop processing the tasks already prefetched by the node. By using -Ofair, that changed. Even though one of the workers in the node was executing a long running tasks, others would not stop processing and would continue to process the tasks prefetched by the node. So I see two levels of prefetching. One at the worker node level. The other at the individual worker level. Using -Ofair for me seemed to disable it at the worker level.

How is ACKS_LATE related? ACKS_LATE = True means that the task will be acknowledged only when the task succeeds. If not, I suppose it would happen when it is received by a worker. In case of prefetch, the task is first received by the worker (confirmed from logs) but will be executed later. I just realized that prefetched messages show up under "unacknowledged messages" in rabbitmq. So I'm not sure if setting it to True is absolutely needed. We anyway had our tasks set that way (late ack) for other reasons.

Solution 3

Just a warning: as of my testing with the redis broker + Celery 3.1.15, all of the advice I've read pertaining to CELERYD_PREFETCH_MULTIPLIER = 1 disabling prefetching is demonstrably false.

To demonstrate this:

1. Set CELERYD_PREFETCH_MULTIPLIER = 1
2. Queue up 5 tasks that will each take a few seconds (ex, time.sleep(5))

3. Start watching the length of the task queue in Redis: watch redis-cli -c llen default

4. Start celery worker -c 1

5. Notice that the queue length in Redis will immediately drop from 5 to 3

CELERYD_PREFETCH_MULTIPLIER = 1 does not prevent prefetching, it simply limits the prefetching to 1 task per queue.

-Ofair, despite what the documentation says, also does not prevent prefetching.

Short of modifying the source code, I haven't found any method for entirely disabling prefetching.

Solution 4

I cannot comment on David Wolever's answers, since my stackcred isn't high enough. So, I've framed my comment as an answer since I'd like to share my experience with Celery 3.1.18 and a Mongodb broker. I managed to stop prefetching with the following:

1. add CELERYD_PREFETCH_MULTIPLIER = 1 to the celery config
2. add CELERY_ACKS_LATE = True to the celery config
3. Start celery worker with options: --concurrency=1 -Ofair

Leaving CELERY_ACKS_LATE to the default, the worker still prefetches. Just like the OP I don't fully grasp the link between prefetching and late acks. I understand what David says "CELERY_ACKS_LATE=True prevents acknowledging messages when they reach to a worker", but I fail to understand why late acks would be incompatible with prefetch. In theory a prefetch would still allow to ack late right - even if not coded as such in celery ?

Solution 5

I experienced something a little bit different with SQS as broker.

The setup was:

CELERYD_PREFETCH_MULTIPLIER = 1
ACKS_ON_FAILURE_OR_TIMEOUT=False
CELERY_ACKS_LATE = True
CONCURRENCY=1

After task fail (exception raised), the worker became unavailable since the message was not acked, both local and remote queue.

The solution that made the workers continue consuming work was setting

CELERYD_PREFETCH_MULTIPLIER = 0

I can only speculate that acks_late was not taken in consideration when writing the SQS transport

View more solutions

Comments

[Jonathan Geisler]

Thanks. 2) Okay, but why would anyone want 'no specific limit'? 3) I'm pretty sure I still see "Got task from broker" messages before the current task has finished.

[Ron Klein]

AFAIK, setting prefetching to 1 is not equivalent to turning it off. It's the lowest possible value (while not hacking the source code) for prefetch, which, in turn, is the number of CPU/cores in the current machine.

[nitwit]

@RonKlein It's not the number of CPUs/cores, it's the number of Celery workers you've defined (which can in some cases be the same, but usually not). Also, if each worker is prefetching a single task, then executing it, then prefetching another, this is equivalent to turning prefetching off, so IMHO your statement is incorrect. (The system must always prefetch as many tasks as there are workers if it wants to keep all workers fed with tasks.)

[Ron Klein]

@nitwit, my statement is based on my own experience with Celery + Redis as its backend. To elaborate, consider the following scenario: Using a single machine for workers, I start a Celery worker with CELERYD_PREFETCH_MULTIPLIER set to 1. A task is handled. The first thing being executed is to unsubscribe from the current queue. Once the task is finished, I either get another task, or get an error claiming no handler found for current task. One way or another, this means that prefetching is not turned off.

[tigeronk2]

I think CELERYD_PREFETCH_MULTIPLIER set to 1 basically means 'fetching'. Pre-fetching is implied when the value is set to >1. So, in effect tasks beyond 1 are being 'pre-fetched'.

[David Wolever]

I've run some experiments, and (at least with the Redis broker) settings CELERYD_PREFETCH_MULTIPLIER = 1 does not disable prefetching. It simply — as the name suggests — only prefetches one task at a time.

[Jonathan Geisler]

Thanks for still contributing to this question! Could you add a bit more detail? For example, you're writing -Ofair has a "different effect", but not how the effect is different. Also, you're bringing up CELERY_ACKS_LATE, as others have before, but so far nobody has managed to explain to me what that attribute has to do with disabling prefetching.

[JiminyCricket]

I had this same issue, running with a redis backend. I had 4 concurrent tasks running and when one started hanging the others would wait for that one to finish (it wouldn't) - Killing that worker would then allow others to resume. I already had prefetch=1, celery_acks=True and when I added -Ofair it fixed the issue where they were waiting for the hung worker. Unfortunately the hung worker issue still isn't fixed for me and so all the workers eventually hang, but at least they no longer do it at the exact same time.

[Manish]

please take a look at this link

[jodag]

As other answers have mentioned if you also set CELERY_ACKS_LATE = 1 then you will effectively disable prefetching.

[gCoh]

I created a PR handling this case, it will be available in celery==4.4.0 github.com/celery/celery/pull/5843

[kapad]

@DavidWolever do you still have the code that you used to run this experiment? would you please share it here if you do. Thanks.

[Waqas Ali]

Not working for me. I am using celery 5.2, is there any settings which can impact?

[Waqas Ali]

This works for me with celery 5.2, adding here, may be this is helpful for someone. stackoverflow.com/a/64869186/1900645