[Python] How to set concurrency limit per task (or queue) in Celery?

My example code is:

• run.py

from celery import Celery, Task

app = Celery(
"app",
task_serializer='json',
accept_content=['json'],
result_serializer='json',
timezone='Asia/Tokyo',
enable_utc=True,
backend='redis://127.0.0.1:6369/2',
broker='redis://127.0.0.1:6369/3',
include=['app.tasks']
)

app.conf.task_routes = {'app.tasks.long_run': {'queue': 'long_running_task'}}


class VerboseTask(Task):

def on_failure(self, exc, task_id, args, kwargs, einfo):
print("-------------------------------------------")
print(
f"FAILURE({task_id})："
f"{self.request.task}{kwargs}, Exception={exc}, ErrorInfo={einfo}"
)
print("-------------------------------------------")

def on_success(self, retval, task_id, args, kwargs):
print("-------------------------------------------")
print(f"SUCCESS({task_id})：{self.request.task}{kwargs}")
print("-------------------------------------------")

def on_retry(self, exc, task_id, args, kwargs, einfo):
print("-------------------------------------------")
print(f"RETRY({task_id})：{self.request.task}{kwargs}")
print("-------------------------------------------")

• app/tasks.py

from run import app, VerboseTask
from time import sleep


class TSK:

name = 'tsk'

@app.task(bind=True, base=VerboseTask)
def short_run(self, **kwargs):
for i in range(0, 3):
print(f'{self.request.task} [{self.request.id[0:5]}] → {i*5} ~ {(i+1)*5}')
sleep(5)
print('short_run finished')

@app.task(bind=True, base=VerboseTask)
def long_run(self, **kwargs):
for i in range(0, 10):
print(f'{self.request.task} [{self.request.id[0:5]}] → {i*5} ~ {(i+1)*5}')
sleep(5)
print('long_run finished')



And I run celery with

celery -A run worker -l info --concurrency=4 -Q long_running_task,celery


The problem is when I invoke long_running_task four times and then usual celery task, the long_running_task consume all workers and celery task need to wait for four long running tasks.

I want to keep at least 2 workers for short running tasks. For example if current tasks are two short running tasks and one long running task, then only one worker is available and only short running task able run. Long running task have to wait while three workers will be free at same time to consume one of it.

In other way just limiting of max concurrency for long running task to 2 will be also a possible solution.

I know that I may run two Celery instances with separated concurrency:

celery -A run worker -l info --concurrency=2 -n worker1@%h -Q long_running_task
celery -A run worker -l info --concurrency=2 -n worker2@%h -Q celery


But

1. I want to consume long_running_task workers for other tasks too if that is idle (that is not so important condition and may be ignored)
2. I use docker container and want to keep only one ENTRYPOINT inside it (in other words I don't want to separate workers by running separate commands. I want to execute celery -A run worker ... only once)

Is here any way to customize limits and concurrency for workers that meet my conditions?

Continue reading...