Skip to content

Fair Scheduling Across Tenants in a Worker Pool

A document-conversion service serves hundreds of customers from one worker pool. Most submit a few files an hour; one runs a nightly migration and submits fifty thousand. With a single FIFO queue, every small customer's job sits behind that backlog, and support tickets arrive from people whose two-page PDF took eleven minutes. Nothing is broken — the pool is working exactly as designed, at full throughput, in submission order. FIFO is simply the wrong policy for shared capacity: it gives each job an equal chance, when what the product promises is that each tenant gets a fair share. This guide measures the noisy-neighbour effect, replaces the single queue with per-tenant queues and round-robin dispatch, adds weights so paid plans get a larger share, bounds each tenant's backlog so fairness does not become unbounded memory, and exports the metrics that show whether the policy is working.

Prerequisites

One queue versus per-tenant rotation 2 lanes over time. One queue versus per-tenant rotation FIFO queue bulk bulk bulk acme waits round robin bulk acme bulk globex time → Throughput is identical; only who waits changes.

1. Measure the noisy neighbour

Simulate the workload: 90% of jobs from one bulk tenant, the rest from two small tenants, four workers, one queue. Record how long each job waited before it completed, grouped by tenant.

import asyncio
import collections
import random
import statistics
import time

Job = tuple[str, int, float]                              # tenant, id, processing time


def make_jobs(seed: int = 5, n: int = 300) -> list[Job]:
    rng = random.Random(seed)
    jobs = []
    for i in range(n):
        tenant = "bulk" if rng.random() < 0.9 else rng.choice(["acme", "globex"])
        jobs.append((tenant, i, 0.004))
    return jobs


async def measure(strategy, jobs: list[Job], workers: int = 4) -> dict[str, tuple[float, float, int]]:
    latencies: dict[str, list[float]] = collections.defaultdict(list)
    started = time.perf_counter()

    async def run(job: Job) -> None:
        await asyncio.sleep(job[2])
        latencies[job[0]].append(time.perf_counter() - started)

    await strategy(jobs, workers, run)
    return {t: (round(statistics.median(v), 3), round(max(v), 2), len(v))
            for t, v in sorted(latencies.items())}


async def fifo(jobs: list[Job], workers: int, run) -> None:
    queue: asyncio.Queue[Job] = asyncio.Queue()
    for job in jobs:
        queue.put_nowait(job)

    async def worker() -> None:
        while not queue.empty():
            await run(queue.get_nowait())

    await asyncio.gather(*(worker() for _ in range(workers)))


print("fifo (median, worst, count):", asyncio.run(measure(fifo, make_jobs())))

With one queue, the small tenants' medians land around 0.11 and 0.17 seconds and their worst case at about 0.31 seconds — the same as the bulk tenant's, because they are interleaved throughout its backlog. Their experience is entirely determined by someone else's volume.

Verify: the small tenants' worst-case latency is close to the bulk tenant's, and roughly equal to the total time to drain the queue.

2. Dispatch round-robin across per-tenant queues

Give each tenant its own queue and have workers take the next job from the next tenant in rotation. Small tenants then wait behind at most one job per other active tenant, not behind the whole backlog.

import asyncio
import collections


async def round_robin(jobs: list[Job], workers: int, run) -> None:
    queues: dict[str, collections.deque[Job]] = {}
    for job in jobs:
        queues.setdefault(job[0], collections.deque()).append(job)
    rotation = collections.deque(queues)
    lock = asyncio.Lock()                                  # dispatch decisions are serialised

    async def next_job() -> Job | None:
        async with lock:
            for _ in range(len(rotation)):
                tenant = rotation[0]
                rotation.rotate(-1)
                if queues[tenant]:
                    return queues[tenant].popleft()
            return None

    async def worker() -> None:
        while (job := await next_job()) is not None:
            await run(job)

    await asyncio.gather(*(worker() for _ in range(workers)))


print("round robin (median, worst, count):", asyncio.run(measure(round_robin, make_jobs())))

The small tenants' medians dropped to about 0.03 seconds and their worst case to about 0.06, while the bulk tenant's median rose only slightly — it still receives every worker-second the small tenants do not use. Total throughput is unchanged; only the order differs.

Verify: small-tenant medians fall several-fold versus FIFO, the bulk tenant still completes all its jobs, and the total run time is roughly the same.

Small-tenant latency, FIFO versus round robin 4 bars comparing FIFO, small tenant worst with the others. Small-tenant latency, FIFO versus round robin FIFO, small tenant worst 0.31 s FIFO, small tenant median 0.107 s round robin, small tenant worst 0.06 s round robin, small tenant median 0.033 s Measured with the step 1 and 2 code: 300 jobs, 90% from one bulk tenant, 4 workers. The bulk tenant still finishes in about the same total time.

3. Weight the rotation by plan or cost

Plain round robin gives every tenant an equal share regardless of plan, and treats a one-second job the same as a one-millisecond one. Deficit round robin fixes both: each tenant accumulates credit proportional to its weight and spends credit equal to each job's cost.

import asyncio
import collections


class FairQueue:
    """Deficit round robin over per-tenant queues, with per-tenant backlog limits."""

    def __init__(self, weights: dict[str, float], max_backlog: int = 1000,
                 default_weight: float = 1.0) -> None:
        self.weights = weights
        self.default_weight = default_weight
        self.max_backlog = max_backlog
        self.queues: dict[str, collections.deque] = {}
        self.deficit: dict[str, float] = collections.defaultdict(float)
        self.rotation: collections.deque[str] = collections.deque()
        self.rejected: collections.Counter[str] = collections.Counter()
        self._items = asyncio.Semaphore(0)
        self._lock = asyncio.Lock()

    def submit(self, tenant: str, job, cost: float = 1.0) -> bool:
        queue = self.queues.get(tenant)
        if queue is None:
            queue = self.queues[tenant] = collections.deque()
            self.rotation.append(tenant)
        if len(queue) >= self.max_backlog:
            self.rejected[tenant] += 1                      # step 4: shed instead of buffering
            return False
        queue.append((job, cost))
        self._items.release()
        return True

    async def get(self) -> tuple[str, object]:
        await self._items.acquire()                         # blocks until some job exists
        async with self._lock:
            for _ in range(len(self.rotation) * 2):
                tenant = self.rotation[0]
                queue = self.queues[tenant]
                if not queue:
                    self.rotation.rotate(-1)
                    continue
                job, cost = queue[0]
                if self.deficit[tenant] >= cost:            # enough credit: serve it
                    self.deficit[tenant] -= cost
                    queue.popleft()
                    return tenant, job
                self.deficit[tenant] += self.weights.get(tenant, self.default_weight)
                self.rotation.rotate(-1)                    # earn credit, try the next tenant
            raise RuntimeError("no job available")


async def main() -> None:
    fair = FairQueue(weights={"enterprise": 4, "free": 1}, max_backlog=200)
    rng = random.Random(5)
    for i in range(300):
        tenant = "free" if rng.random() < 0.8 else "enterprise"
        fair.submit(tenant, i)

    served: collections.Counter[str] = collections.Counter()
    latencies: dict[str, list[float]] = collections.defaultdict(list)
    started = time.perf_counter()

    async def worker() -> None:
        while any(fair.queues[t] for t in fair.queues):
            tenant, _job = await fair.get()
            await asyncio.sleep(0.003)
            served[tenant] += 1
            latencies[tenant].append(time.perf_counter() - started)

    await asyncio.gather(*(worker() for _ in range(4)))
    print("served:", dict(served),
          "| medians:", {t: round(statistics.median(v), 3) for t, v in latencies.items()},
          "| rejected:", dict(fair.rejected))


asyncio.run(main())

With weights 4:1, the enterprise tenant's median latency was about 0.03 seconds against 0.13 for the free tier, although free submitted four times as many jobs. Costs make the accounting honest when job sizes differ: charge estimated pages, bytes or seconds rather than one credit per job, so a tenant submitting huge jobs cannot consume more than its share.

Verify: the weighted tenant's median latency is several times lower than the unweighted one's, and both tenants make progress throughout.

4. Bound each tenant's backlog

Fairness in order does not bound memory. A tenant that submits faster than the pool drains will grow its queue forever unless the queue says no. Per-tenant limits keep one tenant's backlog from consuming the heap, and turn "eventually" into an immediate, actionable error.

import asyncio


async def main() -> None:
    fair = FairQueue(weights={"bulk": 1, "acme": 1}, max_backlog=50)
    accepted = sum(fair.submit("bulk", i) for i in range(200))        # 200 offered
    print("bulk accepted:", accepted, "rejected:", fair.rejected["bulk"])
    print("acme still accepted:", fair.submit("acme", "small-job"))   # unaffected


asyncio.run(main())

The bulk tenant filled its 50 slots and had the rest rejected, while the small tenant's submission was accepted immediately. Rejections should reach the client as a retryable signal — HTTP 429 with Retry-After, or a queue-full error the SDK backs off on — as covered in handling 429 and Retry-After responses in async clients. A per-tenant limit is also the natural place to enforce plan quotas.

Verify: exactly max_backlog jobs are accepted for the flooding tenant, the rejection counter holds the rest, and other tenants are unaffected.

5. Export per-tenant fairness metrics

Fairness is a property you must watch, because the workload changes. Three series per tenant tell the story: wait time before a job starts, share of completed work, and rejected submissions. Aggregate by plan, not by tenant ID, to keep label cardinality bounded.

import collections
import statistics


class FairnessMetrics:
    def __init__(self) -> None:
        self.waits: dict[str, list[float]] = collections.defaultdict(list)
        self.completed: collections.Counter[str] = collections.Counter()

    def record(self, plan: str, wait_seconds: float) -> None:
        self.waits[plan].append(wait_seconds)
        self.completed[plan] += 1

    def report(self) -> dict[str, dict[str, float]]:
        total = sum(self.completed.values()) or 1
        return {
            plan: {
                "p50_wait_s": round(statistics.median(waits), 3),
                "p95_wait_s": round(sorted(waits)[int(len(waits) * 0.95) - 1], 3),
                "share_of_completions": round(self.completed[plan] / total, 3),
            }
            for plan, waits in self.waits.items()
        }


metrics = FairnessMetrics()
for i in range(100):
    metrics.record("free", 0.10 + i * 0.001)
for i in range(25):
    metrics.record("enterprise", 0.02 + i * 0.001)
print(metrics.report())

Alert when a plan's p95 wait exceeds its service-level target, and watch share_of_completions against the configured weights: a share far below the weight means the tenant is not submitting enough to use it, while a share far above means the weights or the cost function need revisiting. Sudden rejections concentrated on one tenant usually mean a client loop gone wrong, not a capacity problem.

Verify: the report shows a lower p50 and p95 wait for the weighted plan and shares that roughly track the configured weights under sustained load from both.

What each policy controls A grid of 5 rows by 2 columns. What each policy controls policy controls does not control FIFO queue arrival order fairness, memory round robin equal share of order plan weights, memory deficit round robin weighted share by cost memory per-tenant backlog limit memory and admission order rate limiting at the edge volume per tenant order A multi-tenant pool needs one row from each half of this table.

Verification

Fair scheduling is working when:

  • Small tenants are insulated: their p95 wait is close to their own processing time rather than to the largest tenant's backlog.
  • Throughput is unchanged: total completions per second match the FIFO baseline; fairness reorders work, it does not reduce capacity.
  • Weights are honoured: completion shares track configured weights when all tenants have work queued.
  • Backlogs are bounded per tenant: one tenant cannot grow the queue without limit, and rejections are reported per tenant.
  • Metrics exist per plan: wait percentiles, completion share and rejections are exported with bounded cardinality.

Pitfalls & edge cases

  • Fairness without limits. Round robin alone still lets a flooding tenant occupy unbounded memory; always pair it with per-tenant backlog caps.
  • Equal credit for unequal jobs. Charging one credit per job lets a tenant submitting huge jobs take far more capacity than its share. Charge by an estimate of cost, and correct the estimate from measured durations.
  • Starving idle tenants on wake-up. A tenant that was idle should not accumulate unlimited credit and then burst; cap accumulated deficit at a small multiple of the weight.
  • Per-process fairness only. With several worker processes, each schedules its own share; global fairness needs a shared broker or a coordinator, and per-tenant rate limits at the edge.
  • Long jobs block the slot. Round robin is non-preemptive: a ten-minute job holds its worker. Split long jobs into chunks, or run them in a separate pool.

Frequently Asked Questions

How do I stop one tenant from monopolising an async worker pool?

Replace the single FIFO queue with one queue per tenant and dispatch jobs round-robin across tenants, so a tenant's job waits behind at most one job from each other active tenant. Add per-tenant backlog limits so a flooding tenant is rejected rather than buffered.

What is deficit round robin and why use it for worker pools?

Deficit round robin gives each tenant credit proportional to its weight on every pass and charges each job a cost. A tenant may run a job only when its accumulated credit covers the cost. This lets different plans receive different shares and accounts for jobs of different sizes, unlike plain round robin.

Does fair scheduling reduce total throughput?

No. It changes the order in which queued work runs, not how fast workers process it. Total completions per second stay the same as with FIFO, while waiting time shifts from small tenants to the tenant with the large backlog, which still receives all the capacity nobody else is using.

How should a multi-tenant pool handle a tenant that floods it?

Bound that tenant's queue and reject further submissions with a retryable error such as HTTP 429 with Retry-After, rather than buffering indefinitely. Combine the backlog limit with per-tenant rate limiting at the edge, and alert when rejections concentrate on one tenant.