gistfile1.txt

# jobs.py

import time
import random
import math
import multiprocessing
from celery import Task, shared_task, group

from api.utils.logger import logger

@shared_task(bind=True)
def low_priority_task(self: Task, id: int):
    worker = multiprocessing.current_process().name
    logger.info(f"starting low priority task {id=} on {worker=}")
    sleep_time = math.ceil(random.uniform(10, 30))
    time.sleep(sleep_time)
    logger.info(f"finishing low priority task {id=} and {sleep_time=}")
    return sleep_time

@shared_task(bind=True)
def spawn_many_low_priority_tasks(self: Task, num_tasks: int):
    worker = multiprocessing.current_process().name
    logger.info(f"spawning {num_tasks} low priority tasks: {self.name=} on {worker=}")
    return self.replace(group(low_priority_task.s(i) for i in range(num_tasks)))

@shared_task(bind=True, priority=0)
def high_priority_task(self: Task, id: int):
    worker = multiprocessing.current_process().name
    logger.info(f"*** STARTING HIGH PRIORITY *** task {id=} on {worker=}")
    sleep_time = math.ceil(random.uniform(5, 15))
    time.sleep(sleep_time)
    logger.info(f"*** FINISHING HIGH PRIORITY *** task {id=} and {sleep_time=}")
    return sleep_time
    
# celery.py
from celery import Celery
from celery.schedules import crontab
from celery.signals import worker_ready

from api.constants import CELERY_BACKEND_URL, CELERY_BROKER_URL

celery_app = Celery(
    main="erop_testing_tasks",
    broker=CELERY_BROKER_URL,
    backend=CELERY_BACKEND_URL,
    broker_connection_retry_on_startup=True,
    timezone="EST",
    include=[
        "api.tasks.test.jobs",
    ],
    beat_schedule={
        "make low priority tasks": {
            "task": "api.tasks.test.jobs.spawn_many_low_priority_tasks",
            "schedule": crontab(),
            'args': (3,)
        },
    },
)

celery_app.conf.update(
    task_default_priority=5,
          broker_transport_options={
          "queue_order_strategy": "priority",
          "priority_steps": list(range(10)),  # 0 highest .. 9 lowest
          "sep": ":",
      },
)

@worker_ready.connect
def on_worker_ready(**kwargs):
    from api.tasks.test import jobs

    # spawn many low priority tasks
    jobs.spawn_many_low_priority_tasks.apply_async(args=[10])

# README
Steps

# 1. worker
Before you start
`redis-cli -n 1 FLUSHDB`

start up Celery
`poetry run celery -A api.tasks.test.celery worker -l INFO -B --concurrency=2`

# 2. confirm backlog in db 1 (do this fast, while lows wait)
redis-cli -n 1 llen celery:5      # expect ~38

# 3. drop high task (baked priority=0)
poetry run python -c "from api.tasks.test.celery import celery_app; from api.tasks.test.jobs import high_priority_task; high_priority_task.delay(999)"

# 4. confirm it sits in priority-0 bucket before a slot frees
redis-cli -n 1 llen celery        # expect 1

# 5. watch worker log → 'starting high priority task ... id=999' jumps ahead of queued lows