Priority Queues for Mixed GPU/CPU Media Jobs

This page sits beneath Celery Task Routing for Video Jobs inside the broader Pipeline Automation & Batch Processing layer, and it solves one specific failure: a single Celery deployment where a 40-minute GPU transcode holds the whole worker fleet hostage while a two-second ffprobe metadata check waits behind it. The discipline that makes this work — pinning tasks to the hardware they actually need — is the queue-side expression of the media ingestion and format architecture principle that every stage declares its resource profile up front.

Problem Framing

The default Celery topology has one queue named celery and every worker draining it. That is fine when all tasks cost roughly the same. Media pipelines violate that assumption violently: a GPU-bound NVENC transcode or a Whisper inference pass runs for minutes and saturates a scarce accelerator, while a remux (ffmpeg -c copy) or an ffprobe duration check finishes in milliseconds on any CPU core. Mix them in one queue and you get head-of-line blocking — the cheap task is stuck behind the expensive one, not because it needs the same resource, but because the worker that could serve it is busy on the wrong job.

The symptom in production is a latency histogram with a bimodal tail and a flat GPU utilization curve that does not match your backlog:

WARN  queue_latency  queue=celery  p50=1.9s  p99=214.6s   (probe jobs stuck behind transcode)
WARN  worker_util    pool=default  gpu_busy=100%  cpu_idle=71%  (cores idle, jobs waiting)
INFO  backlog        celery=1842   oldest_age=397s          (probe SLA is 5s; breached)

Two things are wrong at once. First, there is no separation of where work runs — light CPU work is competing for the same worker slots as GPU work. Second, even within a single class of work there is no notion of urgency: a re-encode for a live-republish should jump ahead of a bulk back-catalogue re-transcode, but FIFO gives them equal footing. Fixing both requires named queues bound to hardware-specific pools, plus a broker-level priority mechanism inside the queue that carries the urgent work.

Solution Architecture

The design splits the single queue into hardware-affinity lanes. Heavy work routes to a gpu.encode queue drained only by workers that hold a GPU lease and run at concurrency 1 (one accelerator, one in-flight job). Light work routes to a cpu.io queue drained by a wide CPU pool. A producer’s task_routes map decides the lane from the task name, so callers never hard-code a queue. Within gpu.encode, broker priority lets an urgent transcode outrank a bulk one — but priority semantics differ sharply between brokers, and getting them wrong silently disables the feature.

Hardware-affinity Celery routing for mixed GPU and CPU media jobs A task producer applies a task_routes map that sends transcode and inference tasks to a gpu.encode queue and remux and probe tasks to a cpu.io queue. The gpu.encode queue is priority-enabled, holding an urgent priority-9 transcode ahead of a bulk priority-0 transcode; it is drained by a dedicated GPU worker pool running concurrency 1 with prefetch 1 and acks_late enabled. The cpu.io queue is drained by a wide CPU worker pool running concurrency 8 with prefetch 4. Because the pools are separate, a long GPU job never blocks a fast probe job. GPU LANE · SCARCE, LONG-RUNNING CPU LANE · WIDE, SHORT-RUNNING Task producer apply_async() task_routes: transcode.* -> gpu.encode infer.* -> gpu.encode remux.* -> cpu.io probe.* -> cpu.io gpu.encode queue priority-enabled · x-max-priority 10 urgent transcode priority=9 bulk transcode priority=0 cpu.io queue remux · probe · thumbnail · FIFO GPU worker pool -Q gpu.encode --concurrency 1 prefetch 1 · acks_late true one accelerator, one in-flight job CPU worker pool -Q cpu.io --concurrency 8 prefetch 4 · fast turnaround never blocked by GPU work Separate pools = a 40-minute transcode cannot stall a 2-second probe

Because the routing map lives in the app config and not in call sites, you can add a third lane later (say a cpu.package queue for HLS segmentation) without touching a single producer. The affinity contract stays declarative.

Implementation

1. Declare queues and route by hardware affinity

Define the queues explicitly with kombu.Queue so you control priority arguments, and route by task-name prefix. Setting task_default_queue away from the implicit celery name means an unrouted task fails loudly instead of silently landing in a pool that cannot serve it. The Celery routing documentation is the authoritative reference for task_routes and queue declaration.

# celery==5.4.0  kombu==5.4.0  redis==5.0.7  (broker: RabbitMQ 3.13 or Redis 7.2)
from celery import Celery
from kombu import Queue

app = Celery("media", broker="amqp://guest:guest@rabbitmq:5672//")

app.conf.task_queues = (
    # GPU lane: priority-enabled. x-max-priority is a RabbitMQ queue argument.
    Queue("gpu.encode", routing_key="gpu.encode",
          queue_arguments={"x-max-priority": 10}),
    # CPU lane: plain FIFO, wide fan-out.
    Queue("cpu.io", routing_key="cpu.io"),
)

app.conf.task_default_queue = "cpu.io"       # safe default: cheap lane, not GPU
app.conf.task_routes = {
    "transcode.*": {"queue": "gpu.encode"},   # NVENC / libx264 heavy encodes
    "infer.*":     {"queue": "gpu.encode"},   # Whisper / detection inference
    "remux.*":     {"queue": "cpu.io"},       # ffmpeg -c copy, container swaps
    "probe.*":     {"queue": "cpu.io"},       # ffprobe metadata / validation
    "thumbnail.*": {"queue": "cpu.io"},
}

# Producers set urgency per call; the broker orders within gpu.encode by it.
def enqueue_transcode(asset_id: str, urgent: bool = False) -> str:
    sig = app.signature("transcode.nvenc", args=[asset_id])
    return sig.apply_async(priority=9 if urgent else 0).id

The priority keyword only means something if the queue was declared with x-max-priority (RabbitMQ) or the broker is Redis with priority_steps configured — otherwise it is silently ignored. That silent no-op is the single most common reason “priority doesn’t work.”

2. Bind dedicated worker pools to their queues

Never start a single worker that drains both queues — that reintroduces the head-of-line blocking you just designed away. Start two worker processes, each pinned to one queue with a concurrency matched to its resource. The GPU worker runs --concurrency 1 because there is one accelerator; the CPU worker runs wide.

# GPU pool — one in-flight job per accelerator, ack only after the job finishes.
CUDA_VISIBLE_DEVICES=0 celery -A media worker \
    --queues gpu.encode \
    --concurrency 1 \
    --prefetch-multiplier 1 \
    --hostname gpu@%h \
    -Ofair

# CPU pool — wide fan-out for short remux/probe tasks.
celery -A media worker \
    --queues cpu.io \
    --concurrency 8 \
    --prefetch-multiplier 4 \
    --hostname cpu@%h

3. Tune prefetch and acks_late so priority actually holds

Prefetch is where in-queue priority quietly dies. If a worker prefetches ten messages, the broker hands them over in current priority order — but a priority-9 task that arrives after the prefetch is stuck behind those ten already sitting in the worker’s local buffer. For the GPU lane, worker_prefetch_multiplier = 1 with --concurrency 1 means the worker holds exactly one message, so a newly-arrived urgent transcode is next off the broker. Pair it with acks_late so a crash mid-transcode redelivers the job instead of losing it.

app.conf.worker_prefetch_multiplier = 1     # global floor; GPU lane must be 1
app.conf.task_acks_late = True              # ack after completion, not on receipt
app.conf.task_reject_on_worker_lost = True  # redeliver if a worker dies mid-job
app.conf.broker_transport_options = {
    # Redis-only: without priority_steps, Redis priority is a no-op.
    "priority_steps": list(range(10)),
    "queue_order_strategy": "priority",
}

Broker semantics diverge here and the difference is not cosmetic. RabbitMQ implements true per-queue priority via the x-max-priority argument documented in the RabbitMQ priority queue guide — messages are ordered inside one queue. Redis, per the Redis data types reference, has no native priority queue; Celery emulates it by creating one Redis list per priority step (gpu.encode, gpu.encode\x06\x163, …) and draining higher lists first, which is why priority_steps is mandatory and why fewer, coarser steps perform better. If you need strict, low-latency priority under heavy load, RabbitMQ is the correct broker for the GPU lane.

Verification

Prove the two properties that matter: pool isolation and priority ordering.

# 1. Confirm each queue is drained only by its intended pool.
celery -A media inspect active_queues | grep -E "hostname|gpu.encode|cpu.io"
# expected: gpu@host lists ONLY gpu.encode; cpu@host lists ONLY cpu.io

# 2. Confirm the priority queue was actually declared with x-max-priority (RabbitMQ).
rabbitmqctl list_queues name arguments | grep gpu.encode
# expected: gpu.encode  {"x-max-priority":10}

# 3. Head-of-line test: flood a long transcode, then enqueue a probe.
python -c "
from tasks import enqueue_transcode
from media import app
[enqueue_transcode(f'asset{i}') for i in range(20)]   # saturate GPU lane
r = app.signature('probe.duration', args=['x.mp4']).apply_async()
print('probe queued:', r.id)
"
# expected: probe returns in <5s while transcodes are still running (isolation holds)

A healthy deployment shows the GPU worker at ~100% accelerator utilization with a stable one-deep prefetch, the CPU worker turning probe jobs around in single-digit seconds regardless of GPU backlog, and — when you enqueue with priority=9 behind a backlog of priority=0 transcodes — the urgent job starting next rather than at the tail. Export queue_latency per queue, per-pool prefetch_count, and x-max-priority presence to Prometheus so a regression (someone starting a combined worker, or dropping the queue argument) trips an alert instead of silently degrading. This routing layer is the enforcement point for the resource contracts declared upstream in Celery Task Routing for Video Jobs.

Failure Modes & Edge Cases

Edge case Symptom Remediation
priority= set but queue lacks x-max-priority Priority silently ignored, urgent jobs run FIFO Declare the queue with queue_arguments={"x-max-priority": 10}; recreate the queue (args are immutable)
GPU pool prefetch > 1 New urgent transcode stuck behind buffered bulk jobs Set --prefetch-multiplier 1; priority only orders what is still on the broker
One worker drains both queues Long transcode blocks fast probe again Split into two pinned workers, one -Q per queue
Redis broker, no priority_steps Priority is a no-op; all jobs equal Configure priority_steps in broker_transport_options, or move the GPU lane to RabbitMQ
acks_late off, worker OOM-killed mid-encode Transcode lost, no redelivery Enable task_acks_late and task_reject_on_worker_lost so the job is requeued
Too many Redis priority steps List-per-step overhead, latency creep Collapse to 3–4 coarse steps (e.g. 0, 3, 6, 9) rather than 10
x-max-priority set very high (255) Per-message memory overhead in RabbitMQ Keep max priority ≤ 10; the docs advise small ranges

For the broader retry, redelivery, and dead-lettering behaviour that acks_late feeds into, this queue design pairs directly with Retry Logic & Dead Letter Queues, which governs what happens when a redelivered GPU job keeps failing.

FAQ

Why not just use one queue with priority and skip separate pools?

Priority orders work within a queue; it does nothing about which hardware runs it. If a fast probe task carries lower priority than a running transcode, it still waits for a worker slot that is occupied for 40 minutes. Separate pools give you resource isolation — the probe runs on an idle CPU core no matter what the GPU pool is doing. Priority is for ordering urgency inside one lane; pools are for stopping cross-resource blocking. You need both.

Does priority behave the same on RabbitMQ and Redis?

No, and assuming it does is a common outage cause. RabbitMQ has native per-queue priority via the x-max-priority argument and orders messages inside a single queue. Redis has no priority primitive, so Celery fakes it with one list per priority step and drains higher lists first — which only works if you set priority_steps, and gets slower as you add steps. For strict, low-latency priority under load, put the GPU lane on RabbitMQ.

What concurrency should the GPU worker run?

Match it to the number of jobs an accelerator can run without thrashing — usually one per physical GPU for full-frame NVENC transcode or large-model inference. Running --concurrency 4 against a single card does not give you four times the throughput; it gives you four jobs competing for the same VRAM and compute, longer per-job latency, and out-of-memory risk. Scale GPU throughput by adding pinned workers on more cards, not by raising concurrency on one.