Control media pipeline concurrency with bounded queues

Control a media pipeline's load at the stage that can become saturated. Limit active downstream work, bound the waiting queue, and slow admission when that queue fills. A successful acquisition job should not trigger unlimited downloads, transcription requests or analysis tasks.

This guide is about application-side backpressure: making a producer wait when its consumer has no capacity. It complements the batch recovery guide, which covers individual outcomes. Here the question is how much work to admit before those outcomes exist.

Separate the limits you actually need

A concurrency limit controls work in progress. A rate limit controls how frequently requests start. A queue limit controls waiting work. None substitutes for the other two.

For example, a transcription tool might finish small files quickly while holding larger files for much longer. The same number of active requests can therefore produce very different request rates. Likewise, a short API submission request can create a remote job that continues after the request returns. Limiting concurrent submission calls does not necessarily limit remote jobs.

Give each stage its own ownership boundary:

Use the existing job ID and status endpoint to reconcile acquisition. Your application's downstream state remains separate from the media job's state. Do not label an entire workflow complete solely because the acquisition stage succeeded.

Queue references before downloading bytes

Queue a stable record identifier, not the full media body. Fetch the file only when the consumer is ready to own the work. An item-count limit cannot bound memory if each queued item can contain a large video.

Keep access credentials outside the work identifier. A waiting item should let the worker retrieve current authorized access when it starts. The storage guide explains why a temporary URL is not a durable media reference.

Decide what happens when the queue is full. An internal producer can wait. An interactive request might instead receive an application-defined capacity response. A webhook receiver should normally acknowledge only after durable acceptance, then let a separate worker schedule execution. Waiting for a long-running analysis inside the webhook request can entangle delivery acknowledgement with downstream availability.

These are application design choices, not claims about Tornado's account quotas or a universal queue size. Choose limits from your consumer's contract and measurements.

Demonstrate a bounded local worker pool

The following Python 3.11+ example accepts an iterable of small references. It starts a fixed number of workers, and the producer awaits space in the queue instead of creating one task per item.

Python's queue documentation specifies that a positive maxsize makes put() wait when full. The TaskGroup contract provides task lifetime management and failure propagation. The example deliberately stops the group when a handler raises an ordinary exception; it does not silently discard that item.

import asyncio


async def consume_bounded(refs, handle, *, workers=2, waiting=3):
    if type(workers) is not int or type(waiting) is not int:
        raise ValueError("Use positive integer limits")
    if workers < 1 or waiting < 1:
        raise ValueError("Use positive integer limits")
    queue = asyncio.Queue(maxsize=waiting)
    stop = object()

    async def worker():
        while True:
            ref = await queue.get()
            try:
                if ref is stop:
                    return
                await handle(ref)
            finally:
                queue.task_done()

    async with asyncio.TaskGroup() as group:
        for _ in range(workers):
            group.create_task(worker())
        for ref in refs:
            await queue.put(ref)
        for _ in range(workers):
            await queue.put(stop)


async def demo():
    active = 0
    peak = 0
    completed = []

    async def simulated_handler(ref):
        nonlocal active, peak
        active += 1
        peak = max(peak, active)
        try:
            await asyncio.sleep(0)  # Yield; no network request.
            completed.append(ref)
        finally:
            active -= 1

    await consume_bounded(range(8), simulated_handler)
    assert sorted(completed) == list(range(8))
    assert 1 <= peak <= 2
    assert active == 0


asyncio.run(demo())

The exact demonstration was executed locally, alongside checks for empty input, invalid limits, handler failure, cancellation and a blocked consumer. These are synthetic scheduling checks, not media throughput measurements or live API tests.

The queue holds at most waiting entries; workers can each hold one additional item, and the producer can hold one item while awaiting insertion. The input iterable, handler and result storage can have their own memory costs. In particular, the demonstration's completed list is only a small test fixture. A production consumer should persist outcomes without accumulating an unbounded result list.

This is a single-process, in-memory pattern. It does not survive process loss, coordinate replicas, enforce requests per second or cancel an already accepted remote job. Handlers must cooperate with cancellation; avoid blocking the event loop or hiding cancellation exceptions. If a handler submits asynchronous remote work, returning immediately releases its local worker slot before that remote work finishes.

Put durability and shared capacity around the pattern

For a service that must recover after interruption, store accepted work and outcomes durably. Claim work using your queue or database's supported ownership mechanism. Bound how much each worker prefetches, and record an outcome before acknowledging completed work. The duplicate-event guide covers accepting notifications without scheduling the same logical action twice.

A limit of two workers in one process becomes up to eight active handlers across four equivalent processes. If the consumer has a shared limit, coordinate admission across those processes. Do not assume an in-memory semaphore protects an organization-wide quota.

Also distinguish cancellation from remote completion. If an HTTP wait times out after a downstream service accepted the request, the local slot can become free while remote work still runs. Retain its identifier or unresolved state and reconcile it before admitting replacements under a remote-work limit. The restart recovery pattern explains the same uncertainty at the acquisition boundary.

Measure whether the limit helps the user

Track queue age, active work, completed outcomes and failed or unresolved work separately. A short queue can still contain one old item that never advances. A drained queue can reflect discarded work rather than successful processing.

Start with a conservative consumer limit and a representative workload. Slow the consumer deliberately in a local simulation and confirm that the producer stops advancing rather than allocating tasks indefinitely. Then test failure and shutdown while work is pending. In an authorized integration test, observe the complete handoff, not only the submission response.

Use the time-to-usable-media guide to assess the queue delay alongside processing and delivery. Increasing concurrency is useful only if the resulting workflow remains within its resource limits and produces acceptable outcomes.

Build your first Tornado media workflow, then add a bounded handoff to the next tool. Open the dashboard to inspect acquisition results while your application records downstream progress independently.