Description

This PR is to address issue: #4003

I came across this issue because celery autoscaling wasn't working while running it with airflow. Other people running airflow have had the same issue:
apache/airflow#3989 (comment)

I looked into this and I think the issue surfaces when the prefetch_multiplier=1 and task_acks_late=True. Under those circumstances the prefetch_count is set to the lower bound of the autoscale range:
https://github.com/celery/celery/blob/4.4.2/celery/worker/components.py#L226

In this situation we are caught in a catch-22 with respect to auto-scaling. In order to bump the processes we need autoscale.qty (i.e. reserved requests) to be greater than the number of processes:
https://github.com/celery/celery/blob/4.4.2/celery/worker/autoscale.py#L89

but to increase the prefetch count which would increase the reserved requests, we need to increase the number of processes:
https://github.com/celery/celery/blob/4.4.2/celery/worker/autoscale.py#L140

So we're stuck.

Things work Ok if you have a prefetch_multiplier > 1 because you essentially jump start the prefetch_count. Same goes if you have task_acks_late=False even with prefetch_multiplier=1 because the reserved_requests exceeds the number of processes because we ack before completing each task.

In this PR, I've set the prefetch_count to the upper_bound of the autoscale range on start up rather than the lower bound. Now if work is available reserved_requests will be able to bump up and then the process count will follow suit.

If the user has set an upper bound they should be Ok with pulling that much work (and consequently scaling up workers) if that work is available. With the prefetch count set at the upper bound then there is no need to update in the _grow and _shrink methods.

Below example program to demonstrate failure to scale with prefetch_multiplier=1 and task_acks_late=True.

tasks.py:

from celery import Celery
import time

app = Celery('tasks', broker='redis://localhost')
app.conf['task_acks_late']=True

@app.task
def my_task(item_id):
    print("Processing item {}".format(item_id))
    time.sleep(30)

def add_items(items_count):
    for i in range(0, items_count):
        my_task.apply_async((i,))

add_items(25)

Run:
celery worker -A tasks --autoscale=6,3 --prefetch-multiplier=1

and you'll see that we dont' budge off of 3 processes.