Why a FIFO semaphore starves small jobs

CocoIndex keeps the context an AI agent reads, such as a codebase index, in sync with its sources. One app can watch many sources. Each source has a live component that re-runs only what changed, and every task those components run needs a slot from one shared pool: a slot is permission to run one task. Before CocoIndex 1.1.0, the pool gave a freed slot to the task that had waited longest. So when one source needed a full re-index and another had a few changed items, the small refresh waited behind most of the re-index. In the engine’s test, a refresh of 40 tasks finished only after 283 to 292 of the 400 re-index tasks.

In production the wait is longer. For example, if each re-index task embeds one file and takes 1 s, a re-index of 400,000 files on the default 1,024 slots is about 6.5 minutes of work. Under first come, first served, a one-file fix that arrives behind it waits for most of that, and for that whole time the agent reads the old version of the file.

One CocoIndex app, two sources, one pool of slots

Zoom in to see details.

Close

100%

In CocoIndex a task is one run of a component, the unit of work, such as the processing of one file. max_inflight_components sets the number of slots (1,024 by default); the test and the figures in this post use 8 or 4 slots, so that the effect is easy to see. A job is one update: a full re-index, or a refresh of a few changed items. You do not need to change anything to get the fix: since 1.1.0 the pool shares slots fairly between jobs by default, for App.update() calls and live components alike, and max_inflight_components still sets the pool size (concurrency control docs).

In this post:

• The problem. Why FIFO, the order most semaphores use, lets the job with the most tasks take most of the slots.
• Five rules for picking the job. FIFO, strict priority, a fixed share, round-robin and fewest held first, run on one workload with a short asyncio model, and in an interactive figure you can play.
• What CocoIndex does. How 1.1.0 applies fewest held first across a tree of updates.
• The limits. What the rule does not handle: cost per task, weights and preemption.

Two jobs, one pool: who gets the next slot?

A pool with a free slot has nothing to decide: a task that asks takes the slot. The decision comes when the pool is full, a slot frees up, and several tasks wait: the pool has to pick one of them.

Most semaphores pick the task that has waited longest. Python’s asyncio.Semaphore wakes the first waiter in its queue when a slot frees (locks.py). Tokio’s Semaphore “is fair, which means that permits are given out in the order they were requested”. Java’s Semaphore does the same when you construct it as fair, and its docs recommend that for resource control.

Arrival order treats every task the same, so a job that asks for 400 slots at once takes 400 places in the line, and a job that arrives a moment later stands behind all of them. Think of a shop with 8 cashiers and one line. A customer with 400 items reaches the line first, and each item takes its own place in it. A customer with 40 items arrives next. Each cashier who frees up takes the next item in the line, so the second customer waits until the first customer’s items are through.

The figure runs that shop at a smaller scale with the model from this post: 4 slots, a big job of 20 tasks, and a small job of 4 tasks that arrives 5 ms later. Play it under FIFO, then pick Fewest held first. The other rules and the second workload come up in Five rules for picking the job.

Under FIFO the small job finishes at 60 ms, after all 20 big tasks, although alone it needs 10 ms. Under fewest held first, the rule CocoIndex uses since 1.1.0, it finishes at 30 ms, and the big job finishes at 60 ms instead of 50.

On the real engine, the test test_a_small_update_does_not_wait_for_another_live_components_backlog uses a pool of 8. One live component starts 400 tasks of 10 ms each. Once they fill the pool, a second live component starts 40. When the small job finishes, the test counts how many of the 400 big tasks are already done. Fewer means less of the backlog stood in front of the small job:

Under FIFO the small job waited behind about 71 to 73% of the backlog; under fewest held first, behind about 20%. The engine’s counts differ from the model’s (283 to 292 here, against all 400 in a plain FIFO line) because the engine’s jobs are trees of tasks: even under FIFO, the refresh ran one task at a time on the slot its own run lent it (#2624). The model leaves the tree out; what carries over is how the rules rank. For an agent, this means that under FIFO the time it reads stale content depends on the size of the other job in the pool.

Pick the job, then the task

A pool shared by jobs makes two choices each time a slot frees up: which job gets the slot, and then which of that job’s tasks. FIFO makes only the second choice, over all tasks at once, so the job with the most tasks in the line gets most of the slots. The fix is to make the first choice explicit: keep one line per job, and give the pool a rule that picks the job.

A pool shared by jobs makes two choices for each freed slot

Zoom in to see details.

Close

100%

Inside one job, arrival order is still a good choice: the tasks belong to the same piece of work, and the job finishes when its last task does. The rule for the first choice is where the designs differ.

Five rules for picking the job

We ran five rules on the shape of CocoIndex’s test with the model in this post: 8 slots, a big job of 400 tasks at 0 ms, a small job of 40 tasks at 5 ms, and 10 ms per task. The model runs on a virtual clock, so the numbers repeat exactly. Alone, the small job takes 50 ms and the big job 500 ms. The last column is a second run in which both jobs have 400 tasks, big tasks take 40 ms and small tasks 10 ms; alone, the small job then takes 500 ms. The interactive figure above runs the same rules at a smaller scale.

FIFO. One line for all tasks. It is simple, and no task waits forever, but the small job takes 545 ms for 50 ms of work. Go’s weighted semaphore keeps strict arrival order on purpose: its source notes that serving a later, smaller request first “could cause starvation for large requests”.

Strict priority. Mark the small job as more important, and it finishes in 55 ms, the best time in the table. But someone has to label each job before it runs, and a wrong label undoes the gain: mark the big job first and the small job takes 545 ms, the same as FIFO. A busy high-priority job also starves every job below it. Kubernetes’ API Priority and Fairness uses priority levels, but gives each level its own limit and queues fairly inside a level, “so that, for example, a poorly-behaved controller need not starve others (even at the same priority level)”.

Fixed share. Give each job its own semaphore of 4. The small job is isolated: it takes 100 ms however big the other job is. The cost is idle slots. The big job alone still gets only 4 slots, so it takes 1,000 ms instead of 500. A fixed share also needs the number of jobs in advance, and CocoIndex’s jobs start and end as sources change. Kubernetes’ API Priority and Fairness lets under-used priority levels lend their concurrency to busy ones to win back the idle capacity, at the cost of more settings.

Round-robin. Jobs take turns. In the first run it ties with fewest held first, at 105 ms. But it does not look at what a job already holds, and that shows when tasks differ in length. In the second run, round-robin gives the small job 19% of the slot time in the first second, because each slot the big job wins stays busy four times as long, and the small job finishes at 2,500 ms, as late as under FIFO. Fewest held first gives it 48% of the slot time and finishes it at 1,040 ms.

Fewest held first. Give the freed slot to the waiting job that holds the fewest slots; among equals, to the job that has waited longest. This is max-min fairness on one resource: the next unit goes to whoever has the least. The Dominant Resource Fairness paper extends the same rule to CPUs and memory together, for cluster schedulers. The rule needs a count of slots held per job. It needs no priorities and reserves no slots.

Of the five rules, only fewest held first does well in both runs without priority labels: the small job takes 105 ms in the first and 1,040 ms in the second, and the big job alone still uses all 8 slots. The big job pays for this, but the pool does not lose throughput. Every rule that keeps all 8 slots busy finishes the first run’s whole workload at 550 ms, and they differ only in order. Under fair sharing the small job’s 40 tasks run early, and the big job finishes 50 ms later (10%), which is the small job’s work spread over 8 slots.

How CocoIndex 1.1.0 shares the pool

In engine terms, a task is a component run, and a job is an independent update: a run with no parent run, together with every run below it. That is an App.update(), one cycle of a live component, or one per-key update of a live component. A live component’s updates divide the live component’s share, so the jobs form a tree under the app (admission.rs:26-48). The tree matters when one live component handles many keys: the per-key updates under it share fairly too, so one key’s backlog does not hold back another key.

• Fewest held at each level. A freed slot goes down the tree from the app. At each level it goes to the waiting child whose subtree holds the fewest slots, and among equals to the child that started waiting first. Inside one update, waiting runs keep arrival order. Taking or returning a slot updates the counts on the path to the root.
• No reservation. A run that finds a free slot and nobody waiting takes it at once, so an update that runs alone gets the whole pool. Fairness acts only when a freed slot goes to a waiter, so it adds no latency while slots are free.
• Lending is unchanged. A task that waits for its own child tasks lends its slot to them before they take pool slots. That rule, slot lending, keeps a tree of tasks free of deadlock, whichever waiter the pool serves.

The Rust tests pin the order, starting with a_released_token_goes_to_the_update_holding_the_fewest. Per the PR, both fairness tests fail when the engine ignores the held counts. The PR also notes one gap: fairness between the per-key updates of one live component (mount_each() over a live map), in live mode, has unit tests but no end-to-end test.

What fewest held first does not cover

The rule keeps one count per job. It does not account for three things, and neither does FIFO:

• Cost per task. The rule counts slots held, so two busy jobs get about equal slot time. A job whose tasks take 10 times longer finishes about a tenth as many tasks, and nothing charges for what a slot uses besides time, such as memory or API quota. Sharing by cost needs each task’s cost, known or estimated, before it runs.
• Weights and priorities. You cannot say that a user-facing refresh matters more than a backfill. Strict priorities would let a busy high-priority job starve everyone else, which is the problem this post started from.
• Preemption. A big job that holds every slot keeps them until each of its tasks ends. The small job gets the next freed slots, so with long tasks its first slot can still take a while.

Takeaways

• FIFO across jobs. When the tasks in one pool belong to different jobs, FIFO gives most of the slots to the job with the most tasks in line.
• Two choices. A shared pool decides which job gets a freed slot, then which of that job’s tasks. Arrival order fits the second choice.
• Fewest held first. Serving the job that holds the fewest slots kept the small job at 105 ms in the model, against 545 ms under FIFO, with no priority labels and no reserved slots.
• The cost. Total throughput does not change. The big job finishes later by the small job’s work spread over the pool, 10% in the model.
• In CocoIndex. Fair sharing is on by default from 1.1.0, with no new setting; upgrading is enough.

Reproduce it

Everything in this section was written for this post; none of it is engine code. fair.py is a teaching model: one pool, one line per job, and a rule that picks the job. It has no task tree, no lending and no cancellation. compare.py runs it on a virtual clock, so the outputs do not depend on the machine. The runs used Python 3.9.6.

"""A teaching model of a shared pool that picks which waiting job gets a freed slot.
Not CocoIndex's engine code: no task tree, no lending, no cancellation."""
import asyncio
from collections import defaultdict, deque


class Pool:
    """`size` slots. A task asks for one on behalf of its job; `pick` chooses the job."""

    def __init__(self, size: int, pick) -> None:
        self.free, self.pick = size, pick
        self.held = defaultdict(int)              # job -> slots its tasks hold
        self.waiters: dict[str, deque] = {}       # job -> (arrival, future), oldest first
        self.arrivals = 0

    async def acquire(self, job: str) -> None:
        fut = asyncio.get_running_loop().create_future()
        if not self.request(job, fut):
            await fut                             # release() grants the slot

    def request(self, job: str, waiter) -> bool:
        """Take a slot now (True), or queue `waiter` until release() grants one."""
        if self.free and not any(self.waiters.values()):
            self.free -= 1                        # free and nobody waits: take it now
            self.held[job] += 1
            return True
        self.arrivals += 1
        self.waiters.setdefault(job, deque()).append((self.arrivals, waiter))
        return False

    def release(self, job: str) -> None:
        self.held[job] -= 1
        ready = {j: q for j, q in self.waiters.items() if q}
        if not ready:
            self.free += 1
            return
        chosen = self.pick(self, ready)           # which job ...
        _, waiter = ready[chosen].popleft()       # ... then its oldest task
        self.held[chosen] += 1
        waiter.set_result(None)


def oldest(ready, job):
    return ready[job][0][0]                       # arrival number of the job's first waiter


def fifo(pool, ready):
    """One line for every task: the oldest waiter wins, whatever its job."""
    return min(ready, key=lambda j: oldest(ready, j))


def fewest_held(pool, ready):
    """The job that holds the fewest slots; among equals, the one waiting longest."""
    return min(ready, key=lambda j: (pool.held[j], oldest(ready, j)))


def priority(order):
    """Strict priority: the first job in `order` that has a waiter."""
    return lambda pool, ready: min(ready, key=lambda j: (order.index(j), oldest(ready, j)))


def round_robin():
    """Jobs take turns, whatever they already hold."""
    turn: deque = deque()

    def pick(pool, ready):
        turn.extend(j for j in ready if j not in turn)
        while turn[0] not in ready:
            turn.rotate(-1)
        job = turn[0]
        turn.rotate(-1)
        return job
    return pick

"""Run each rule on one workload, on a virtual clock so the numbers repeat exactly.
A big job of 400 tasks fills a pool of 8 slots; 5 ms later a small job of 40 tasks
arrives. Every task holds its slot for 10 ms. Same shape as CocoIndex's test."""
import heapq
import sys

from fair import Pool, fewest_held, fifo, priority, round_robin


class Clock:
    def __init__(self):
        self.now, self.events, self.n = 0, [], 0

    def at(self, t, fn):
        heapq.heappush(self.events, (t, self.n, fn))
        self.n += 1

    def run(self):
        while self.events:
            self.now, _, fn = heapq.heappop(self.events)
            fn()


def simulate(pool_of, jobs, until=None):
    """jobs: (name, arrives at ms, tasks, ms per task). Returns each task's (job, start, end)."""
    clock, log = Clock(), []

    class Grant:                                   # stands in for an asyncio future
        def __init__(self, job, ms):
            self.job, self.ms = job, ms

        def set_result(self, _):
            start(self.job, self.ms)

    def start(job, ms):
        t0 = clock.now
        def end():
            log.append((job, t0, clock.now))
            pool_of(job).release(job)
        clock.at(t0 + ms, end)

    def arrive(job, n, ms):
        for _ in range(n):
            if pool_of(job).request(job, g := Grant(job, ms)):
                start(job, ms)

    for job, t, n, ms in jobs:
        clock.at(t, lambda job=job, n=n, ms=ms: arrive(job, n, ms))
    clock.run()
    return log


def shared(pick):
    pool = Pool(8, pick)
    return lambda job: pool


def fixed_share():
    pools = {"big": Pool(4, fifo), "small": Pool(4, fifo)}
    return lambda job: pools[job]


RULES = [
    ("FIFO", lambda: shared(fifo)),
    ("priority, small first", lambda: shared(priority(["small", "big"]))),
    ("priority, big first", lambda: shared(priority(["big", "small"]))),
    ("fixed share, 4 + 4", fixed_share),
    ("round-robin", lambda: shared(round_robin())),
    ("fewest held first", lambda: shared(fewest_held)),
]


def finish(log, job):
    return max(end for j, _, end in log if j == job)


if __name__ == "__main__":
    print("big job: 400 tasks at 0 ms; small job: 40 tasks at 5 ms; 8 slots; 10 ms per task")
    print("alone, the small job takes 50 ms and the big job 500 ms")
    print(f"{'rule':<22} {'small job took':>16} {'big done by then':>17} {'big job took':>13} {'big job alone':>14}")
    for name, make in RULES:
        log = simulate(make(), [("big", 0, 400, 10), ("small", 5, 40, 10)])
        small_end = finish(log, "small")
        big_then = sum(1 for j, _, end in log if j == "big" and end <= small_end)
        alone = finish(simulate(make(), [("big", 0, 400, 10)]), "big")
        print(f"{name:<22} {small_end - 5:>13} ms {big_then:>10} of 400 "
              f"{finish(log, 'big'):>10} ms {alone:>11} ms")

    print()
    print("both jobs stay busy: 400 tasks each at 0 ms; big tasks take 40 ms, small tasks 10 ms")
    print("alone, the small job takes 500 ms")
    print(f"{'rule':<22} {'small job took':>16} {'small job share of slot time, first 1 s':>41}")
    for name, make in RULES:
        log = simulate(make(), [("big", 0, 400, 40), ("small", 0, 400, 10)])
        busy = {j: sum(max(0, min(e, 1000) - s) for jj, s, e in log if jj == j) for j in ("big", "small")}
        print(f"{name:<22} {finish(log, 'small'):>13} ms {busy['small'] / (busy['big'] + busy['small']):>40.0%}")

big job: 400 tasks at 0 ms; small job: 40 tasks at 5 ms; 8 slots; 10 ms per task
alone, the small job takes 50 ms and the big job 500 ms
rule                     small job took  big done by then  big job took  big job alone
FIFO                             545 ms        400 of 400        500 ms         500 ms
priority, small first             55 ms          8 of 400        550 ms         500 ms
priority, big first              545 ms        400 of 400        500 ms         500 ms
fixed share, 4 + 4               100 ms         40 of 400       1000 ms        1000 ms
round-robin                      105 ms         48 of 400        550 ms         500 ms
fewest held first                105 ms         48 of 400        550 ms         500 ms

both jobs stay busy: 400 tasks each at 0 ms; big tasks take 40 ms, small tasks 10 ms
alone, the small job takes 500 ms
rule                     small job took   small job share of slot time, first 1 s
FIFO                            2500 ms                                       0%
priority, small first            540 ms                                      50%
priority, big first             2500 ms                                       0%
fixed share, 4 + 4              1000 ms                                      50%
round-robin                     2500 ms                                      19%
fewest held first               1040 ms                                      48%

How CocoIndex applies this

CocoIndex is an open-source incremental engine for the context AI agents read: codebases, documents, meeting notes, PR reviews and Slack threads. You write the pipeline in Python; when a source changes, only the affected components re-run, and the targets the agent queries stay in sync. In live mode, the live components of one app share the engine’s pool, and since 1.1.0 a small refresh of one source does not wait behind another source’s full re-index.

The engine code behind this post is in rust/core/src/engine/admission.rs, and its tests are in test_concurrency_control.py. CocoIndex is Apache 2.0 licensed. To try it, start with the examples and the concurrency control docs. If a workload of yours shares the pool in a way this rule handles badly, open an issue with its shape.

Fair sharing shipped in CocoIndex 1.1.0 (#2624); the 1.0.26–1.1.0 changelog has the release notes.

About the author

George HeCofounder & CTO of CocoIndex · @georgehe0

Maintainer of CocoIndex, Ex-Google Infra Lead. Writes about incremental data infrastructure, Rust internals, and the engineering decisions behind the engine.