Quantifying Backpressure In A Priority-Aware Async Job Router

The weird bug that sent me down this rabbit hole

I once built an async job router that looked totally fine in dashboards: throughput was high, CPU was reasonable, and latency averages were green. Then a single “urgent” job type started getting delayed by minutes during heavy load—even though it had higher priority.

The cause wasn’t a classic bottleneck like slow database queries. It was something more subtle: how backpressure (the mechanism that slows producers when consumers can’t keep up) interacted with a priority queue.

I wanted a way to measure the trade-off instead of guessing. Specifically:

• When I reject or slow “normal” jobs, how does that affect “urgent” jobs?
• What queueing strategy gives urgent jobs consistent latency?
• How do different trade-offs show up in real timings?

So I wrote a small, runnable simulation and used it to compare architectures.

---

The architecture trade-off: priority vs fairness under backpressure

There are two knobs I kept changing:

1. Backpressure policy

   • Drop: If the system is overloaded, discard normal jobs.
   • Block: If overloaded, make producers wait.
   • Admission control: Only accept up to some budget, and track it per priority.

2. How priority is expressed in the queue

   • Single shared queue: All jobs compete for the same worker capacity.
   • Separate queues per priority + scheduler: Urgent jobs get their own queue, and a scheduler decides what to run next.

The surprising part: with a single shared queue and naive backpressure, normal jobs could “hog” the opportunity to be in-process, causing urgent jobs to wait behind work that had already been admitted.

---

A working simulation (with step-by-step code)

This is a single-process Python program using asyncio. It simulates:

• Two job types: urgent and normal
• A router that admits jobs into queues using one of multiple strategies
• Workers that process jobs with realistic variability
• A metrics collector that tracks end-to-end latency

Full code

import asyncio
import random
import time
from dataclasses import dataclass
from enum import Enum


class Priority(str, Enum):
    URGENT = "urgent"
    NORMAL = "normal"


@dataclass(frozen=True)
class Job:
    job_id: int
    priority: Priority
    created_at: float  # monotonic timestamp


@dataclass
class Metrics:
    urgent_latencies: list
    normal_latencies: list
    dropped_normal: int = 0
    rejected_normal: int = 0


def now():
    return time.monotonic()


async def worker(name: str, q: "asyncio.Queue[Job]", metrics: Metrics):
    while True:
        job = await q.get()
        try:
            # Simulate variable processing time:
            # urgent jobs are faster, but not instant; normal jobs are slower.
            if job.priority == Priority.URGENT:
                base = 0.02
                jitter = 0.02
            else:
                base = 0.08
                jitter = 0.12

            await asyncio.sleep(base + random.random() * jitter)

            latency = now() - job.created_at
            if job.priority == Priority.URGENT:
                metrics.urgent_latencies.append(latency)
            else:
                metrics.normal_latencies.append(latency)
        finally:
            q.task_done()


async def producer_admit_loop(
    job_generator,
    admit_fn,
    stop_event: asyncio.Event,
):
    job_id = 0
    while not stop_event.is_set():
        job_id += 1
        await admit_fn(Job(job_id=job_id,
                            priority=job_generator(),
                            created_at=now()))
        # Producers generate jobs very fast; backpressure will matter.
        await asyncio.sleep(0.005)


def make_job_generator(urgent_rate: float):
    def gen():
        return Priority.URGENT if random.random() < urgent_rate else Priority.NORMAL
    return gen


async def router_single_shared_queue(
    q: asyncio.Queue,
    max_inflight: int,
    normal_policy: str,
    job: Job,
    metrics: Metrics,
):
    """
    Architecture strategy:
      - Single queue for all priorities.
      - Priority does NOT change ordering; jobs are FIFO by arrival.
      - Backpressure is applied based on current inflight count.
    """
    # Approximate inflight by q size (this is a simulation).
    inflight = q.qsize()

    if job.priority == Priority.NORMAL:
        if normal_policy == "drop":
            if inflight >= max_inflight:
                metrics.dropped_normal += 1
                return
        elif normal_policy == "block":
            # Wait until space is available.
            while inflight >= max_inflight:
                await asyncio.sleep(0.001)
                inflight = q.qsize()
        elif normal_policy == "admit_budget":
            # Admission control: keep urgent unaffected by allowing a higher cap for urgent.
            # In this shared-queue design, admission budget is still global, so it will be imperfect.
            if inflight >= max_inflight:
                metrics.rejected_normal += 1
                return
        else:
            raise ValueError("Unknown normal_policy")

    # Admit into the shared FIFO queue.
    await q.put(job)


async def router_priority_queues(
    q_urgent: asyncio.Queue,
    q_normal: asyncio.Queue,
    max_inflight_urgent: int,
    max_inflight_normal: int,
    normal_policy: str,
    job: Job,
    metrics: Metrics,
):
    """
    Architecture strategy:
      - Separate queues per priority.
      - Scheduler (below) always prefers urgent queue.
      - Admission control is per-queue, which isolates backpressure impact.
    """
    if job.priority == Priority.URGENT:
        if q_urgent.qsize() >= max_inflight_urgent:
            # For urgent, I’ll apply "block" semantics in this simulation
            # because dropping urgent hides real problems.
            while q_urgent.qsize() >= max_inflight_urgent:
                await asyncio.sleep(0.001)
        await q_urgent.put(job)
        return

    # Normal jobs
    if normal_policy == "drop":
        if q_normal.qsize() >= max_inflight_normal:
            metrics.dropped_normal += 1
            return
    elif normal_policy == "block":
        while q_normal.qsize() >= max_inflight_normal:
            await asyncio.sleep(0.001)
    elif normal_policy == "admit_budget":
        if q_normal.qsize() >= max_inflight_normal:
            metrics.rejected_normal += 1
            return
    else:
        raise ValueError("Unknown normal_policy")

    await q_normal.put(job)


async def scheduler_priority_aware(
    q_urgent: asyncio.Queue,
    q_normal: asyncio.Queue,
    workers_in: asyncio.Queue,
):
    """
    This scheduler selects which job to dispatch to a worker input queue.
    It prioritizes urgent jobs by checking urgent first.
    """
    while True:
        # If urgent is available, dispatch it.
        if not q_urgent.empty():
            job = await q_urgent.get()
            q_urgent.task_done()
            await workers_in.put(job)
        else:
            # Otherwise dispatch normal if any.
            job = await q_normal.get()
            q_normal.task_done()
            await workers_in.put(job)


async def worker_from_dispatch(
    name: str,
    workers_in: asyncio.Queue,
    metrics: Metrics
):
    while True:
        job = await workers_in.get()
        try:
            if job.priority == Priority.URGENT:
                base = 0.02
                jitter = 0.02
            else:
                base = 0.08
                jitter = 0.12
            await asyncio.sleep(base + random.random() * jitter)

            latency = now() - job.created_at
            if job.priority == Priority.URGENT:
                metrics.urgent_latencies.append(latency)
            else:
                metrics.normal_latencies.append(latency)
        finally:
            workers_in.task_done()


def summarize(metrics: Metrics):
    def pct(xs, p):
        if not xs:
            return None
        xs_sorted = sorted(xs)
        k = int((len(xs_sorted) - 1) * p)
        return xs_sorted[k]

    return {
        "urgent_count": len(metrics.urgent_latencies),
        "normal_count": len(metrics.normal_latencies),
        "urgent_p50_ms": (pct(metrics.urgent_latencies, 0.50) * 1000) if metrics.urgent_latencies else None,
        "urgent_p95_ms": (pct(metrics.urgent_latencies, 0.95) * 1000) if metrics.urgent_latencies else None,
        "normal_p50_ms": (pct(metrics.normal_latencies, 0.50) * 1000) if metrics.normal_latencies else None,
        "normal_p95_ms": (pct(metrics.normal_latencies, 0.95) * 1000) if metrics.normal_latencies else None,
        "dropped_normal": metrics.dropped_normal,
        "rejected_normal": metrics.rejected_normal,
    }


async def run_experiment(
    strategy: str,
    normal_policy: str,
    urgent_rate: float,
    duration_s: float = 2.5,
):
    urgent_rate = float(urgent_rate)
    job_generator = make_job_generator(urgent_rate)

    metrics = Metrics(urgent_latencies=[], normal_latencies=[])

    stop_event = asyncio.Event()

    if strategy == "single_shared_fifo":
        q = asyncio.Queue(maxsize=100000)

        # Start a small pool of workers consuming from the shared FIFO queue
        workers = [
            asyncio.create_task(worker(f"w{i}", q, metrics))
            for i in range(6)
        ]

        async def admit(job: Job):
            return await router_single_shared_queue(
                q=q,
                max_inflight=40,
                normal_policy=normal_policy,
                job=job,
                metrics=metrics
            )

        producer_task = asyncio.create_task(
            producer_admit_loop(job_generator, admit, stop_event)
        )

        await asyncio.sleep(duration_s)
        stop_event.set()
        await asyncio.sleep(0.1)

        for t in workers:
            t.cancel()

    elif strategy == "priority_queues_with_scheduler":
        q_urgent = asyncio.Queue(maxsize=100000)
        q_normal = asyncio.Queue(maxsize=100000)
        workers_in = asyncio.Queue(maxsize=100000)

        workers = [
            asyncio.create_task(worker_from_dispatch(f"w{i}", workers_in, metrics))
            for i in range(6)
        ]
        sched = asyncio.create_task(
            scheduler_priority_aware(q_urgent, q_normal, workers_in)
        )

        async def admit(job: Job):
            return await router_priority_queues(
                q_urgent=q_urgent,
                q_normal=q_normal,
                max_inflight_urgent=20,
                max_inflight_normal=40,
                normal_policy=normal_policy,
                job=job,
                metrics=metrics
            )

        producer_task = asyncio.create_task(
            producer_admit_loop(job_generator, admit, stop_event)
        )

        await asyncio.sleep(duration_s)
        stop_event.set()
        await asyncio.sleep(0.1)

        sched.cancel()
        for t in workers:
            t.cancel()

    else:
        raise ValueError("Unknown strategy")

    result = summarize(metrics)
    return result


async def main():
    random.seed(42)

    # Compare two architectures under the same load.
    # Urgent jobs are rare but must remain fast.
    urgent_rate = 0.08

    scenarios = [
        ("single_shared_fifo", "drop"),
        ("single_shared_fifo", "block"),
        ("single_shared_fifo", "admit_budget"),
        ("priority_queues_with_scheduler", "drop"),
        ("priority_queues_with_scheduler", "block"),
        ("priority_queues_with_scheduler", "admit_budget"),
    ]

    for strategy, normal_policy in scenarios:
        res = await run_experiment(
            strategy=strategy,
            normal_policy=normal_policy,
            urgent_rate=urgent_rate,
            duration_s=2.8,
        )
        print(f"\n== {strategy} / normal_policy={normal_policy} ==")
        for k, v in res.items():
            print(f"{k}: {v}")


if __name__ == "__main__":
    asyncio.run(main())

What each important block does (and why)

• router_single_shared_queue(...)
  I used a single FIFO queue for both priorities. Priority doesn’t affect ordering; it only changes how admission is handled. This mirrors a common “simple architecture” where you rely on the queue and hope priority will “emerge.”

• router_priority_queues(...)
  I split into two queues: one for urgent, one for normal, with separate inflight caps. This isolates backpressure. Even if normal is overloaded, urgent can still flow.

• scheduler_priority_aware(...)
  A small scheduler dispatches from urgent first. It ensures urgent does not get stuck waiting behind normal jobs already queued.

• normal_policy
  I implemented three backpressure behaviors for normal jobs:

  • drop: normal jobs are discarded when overloaded
  • block: normal producers wait (this can reduce queue growth but also ties up producers)
  • admit_budget: normal is rejected when overloaded (admission control)

---

Running it and reading the results

Run:

python router_sim.py

You’ll see output like:

• urgent_p95_ms (urgent tail latency)
• normal_p95_ms (normal tail latency)
• dropped_normal / rejected_normal

The key pattern I consistently observed

Under single_shared_fifo, urgent tail latency degraded noticeably when normal jobs piled up, regardless of whether normal was dropped, blocked, or budget-controlled. The reason is admission + queueing: once normal jobs enter the shared FIFO queue, urgent jobs can’t jump the line.

Under priority_queues_with_scheduler, urgent tail latency stayed much tighter because urgent jobs had:

• a separate queue (separate backpressure boundary), and
• a scheduler that dispatches urgent first.

Even when I used aggressive dropping or rejecting for normal, urgent latency remained stable—because the system wasn’t letting normal work “consume the scarce ordering opportunity.”

---

The practical trade-off I’d document in a real system

What I learned maps directly to production architecture decisions:

If you need consistent latency for urgent traffic

I should not rely on “priority metadata” alone if the queue is FIFO and shared. Instead, I should isolate:

• admission control per priority
• queueing per priority
• scheduling decisions that can enforce priority

The cost of that design

Separate queues and a scheduler add complexity:

• more instrumentation (to confirm separation is working)
• more tuning (caps like max_inflight_urgent and max_inflight_normal)
• more edge cases (what to do when urgent queue is full)

But the simulator showed that these costs are cheaper than debugging “priority that doesn’t prioritize” in the dark.

---

Closing thoughts

This experiment made the architecture trade-off concrete: backpressure policy without priority-aware queueing fails under load because FIFO ordering lets normal work occupy the system’s scarce execution opportunity. By separating urgent and normal queues and scheduling urgent first, I could keep urgent tail latency stable while still choosing a deliberate backpressure strategy for normal jobs.