Skip to content

Adaptive Concurrency Limits with AIMD

Every asyncio.Semaphore(N) in a service encodes a guess about a dependency's capacity, made once, by a person, at a moment when the dependency looked a particular way. The guess is wrong in both directions over time: too low during quiet periods, when the limit throttles traffic the dependency could absorb, and too high after the dependency slows down — a deploy, a failing replica, a noisy neighbour — when holding the old limit turns a slowdown into a pile-up. TCP solved the same problem decades ago with AIMD: increase the window gradually while things are fine, cut it sharply at the first sign of congestion. The same loop works for a client's concurrency limit, with latency and errors as the congestion signal. This guide builds an AIMD limiter for asyncio, tunes the signal so it reacts to the dependency rather than to noise, bounds it so adaptation cannot become an outage, and measures it against a fixed limit when a dependency degrades mid-run.

Prerequisites

The control loop around every call 4 stages from wait for room to adjust. The control loop around every call wait for room inflight < limit call measure service time classify fast, slow or failed adjust +1/limit or x0.75 The dependency's own latency is the feedback signal; nobody configures a number.

1. Model a dependency whose capacity changes

A useful test needs an upstream that behaves like a real one: service time grows as concurrency passes its comfortable capacity, and beyond a hard ceiling it rejects. Then change its capacity mid-run.

import asyncio
import time


class Upstream:
    """Service time grows with concurrency; past a hard cap it rejects."""

    def __init__(self, capacity: int, base: float = 0.005, hard: int = 25) -> None:
        self.capacity, self.base, self.hard = capacity, base, hard
        self.inflight = 0
        self.peak = 0

    async def call(self) -> None:
        self.inflight += 1
        self.peak = max(self.peak, self.inflight)
        try:
            if self.inflight > self.hard:
                await asyncio.sleep(self.base)
                raise TimeoutError("overloaded")
            over = max(0, self.inflight - self.capacity)
            await asyncio.sleep(self.base * (1 + over))       # queueing delay
        finally:
            self.inflight -= 1

The shape matters more than the numbers: below capacity, latency is flat and adding concurrency adds throughput; above it, latency grows while throughput does not. That knee is what an adaptive limiter searches for, and it is why latency — not throughput — is the signal.

Verify: calling with 4 concurrent callers against capacity=16 gives the base latency; 30 concurrent callers give longer latencies and eventually TimeoutError.

2. Implement the AIMD loop

Additive increase: after each fast, successful call, raise the limit by 1/limit, so the limit grows by about one per full round of the current limit. Multiplicative decrease: on a failure or a call slower than the target, multiply the limit by a backoff factor.

import asyncio


class AIMDLimiter:
    def __init__(self, initial: int = 4, minimum: int = 1, maximum: int = 64,
                 latency_target: float = 0.02, backoff: float = 0.75) -> None:
        self.limit = float(initial)
        self.min, self.max = minimum, maximum
        self.target, self.backoff = latency_target, backoff
        self.inflight = 0
        self._room = asyncio.Event()
        self._room.set()

    async def acquire(self) -> None:
        while self.inflight >= int(self.limit):
            self._room.clear()
            await self._room.wait()                          # wait for room, not for a permit
        self.inflight += 1

    def release(self) -> None:
        self.inflight -= 1
        self._room.set()

    def record(self, latency: float | None = None, failed: bool = False) -> None:
        if failed or (latency is not None and latency > self.target):
            self.limit = max(self.min, self.limit * self.backoff)      # multiplicative decrease
        else:
            self.limit = min(self.max, self.limit + 1 / max(1.0, self.limit))   # additive increase
        self._room.set()

An asyncio.Event is used instead of a Semaphore because the limit changes: a semaphore's permit count is fixed at construction and can only be adjusted by releasing or absorbing permits, which is error-prone. Waiting for room — a comparison against the current limit — adapts naturally when the limit moves in either direction.

Verify: with initial=4, a series of fast successes grows the limit past 5 within a few rounds, and one failure drops it to 75% of its value.

3. Measure it against a fixed limit when the dependency degrades

Run the same workload twice: once with a static Semaphore(16) and once with AIMD starting from 16, halving the upstream's capacity part-way through.

import asyncio
import time


async def workload(acquire, release, record, up: Upstream, duration: float, per_tick: int):
    ok = fail = 0
    service_times: list[float] = []

    async def one_call() -> None:
        nonlocal ok, fail
        await acquire()
        started = time.perf_counter()
        try:
            await up.call()
            service_times.append(time.perf_counter() - started)
            ok += 1
            if record:
                record(latency=service_times[-1])
        except TimeoutError:
            fail += 1
            if record:
                record(failed=True)
        finally:
            release()

    tasks, stop = [], time.perf_counter() + duration
    while time.perf_counter() < stop:
        for _ in range(per_tick):
            tasks.append(asyncio.create_task(one_call()))
        await asyncio.sleep(0.001)
    await asyncio.gather(*tasks)
    p95 = sorted(service_times)[int(len(service_times) * 0.95) - 1] if service_times else 0.0
    return ok, fail, p95


async def scenario(name: str, adaptive: bool) -> None:
    up = Upstream(capacity=16)

    async def degrade() -> None:
        await asyncio.sleep(0.25)
        up.capacity = 4                                      # the dependency slows down

    asyncio.create_task(degrade())
    if adaptive:
        limiter = AIMDLimiter(initial=16)
        ok, fail, p95 = await workload(limiter.acquire, limiter.release, limiter.record,
                                       up, 0.5, 2)
        extra = f" | final limit {limiter.limit:.1f}"
    else:
        sem = asyncio.Semaphore(16)
        ok, fail, p95 = await workload(sem.acquire, sem.release, None, up, 0.5, 2)
        extra = ""
    print(f"{name:>16}: ok={ok} fail={fail} p95_service={p95 * 1000:.1f}ms{extra}")


async def main() -> None:
    await scenario("fixed limit 16", adaptive=False)
    await scenario("AIMD from 16", adaptive=True)


asyncio.run(main())

The fixed limiter kept sending sixteen concurrent calls into an upstream that could comfortably handle four, and its p95 service time rose to about 66 ms. AIMD found the new capacity — its limit settled near 4.8 — and its p95 stayed around 21 ms, with comparable throughput and a handful of failures during the transition. The adaptive limiter does not make the dependency faster; it stops the client from queueing work inside it, which is where the latency came from.

Verify: p95 service time under AIMD is markedly lower after the degradation, with similar completed-call counts.

p95 service time after the dependency slows down 2 bars comparing fixed limit of 16 with the others. p95 service time after the dependency slows down fixed limit of 16 66.1 ms AIMD, settled near 5 20.8 ms Measured with the step 3 scenario: same arrival rate, same upstream, capacity halved at 0.25 s. Both completed a similar number of calls; only the queueing inside the dependency differs.

4. Choose signals that track the dependency, not the noise

A limiter that reacts to every slow request oscillates. Three adjustments make the signal stable: compare against a rolling baseline rather than a fixed number, treat only meaningful failures as congestion, and apply at most one decrease per round trip.

import asyncio
import collections
import statistics


class TunedAIMD(AIMDLimiter):
    def __init__(self, *args, window: int = 50, tolerance: float = 2.0, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self.samples: collections.deque[float] = collections.deque(maxlen=window)
        self.tolerance = tolerance
        self.last_decrease = 0.0
        self.decreases = 0

    def record(self, latency: float | None = None, failed: bool = False) -> None:
        loop = asyncio.get_event_loop()
        congested = failed
        if latency is not None:
            self.samples.append(latency)
            if len(self.samples) >= 10:
                baseline = statistics.median(self.samples)   # what "normal" looks like now
                congested = congested or latency > baseline * self.tolerance
        if congested:
            if loop.time() - self.last_decrease < self.target * 2:
                return                                       # one cut per round trip, not per call
            self.last_decrease = loop.time()
            self.decreases += 1
            self.limit = max(self.min, self.limit * self.backoff)
        else:
            self.limit = min(self.max, self.limit + 1 / max(1.0, self.limit))
        self._room.set()


def is_congestion(exc: BaseException) -> bool:
    """Only signals that mean 'you are sending too much' should shrink the limit."""
    if isinstance(exc, TimeoutError):
        return True
    status = getattr(exc, "status_code", None)
    return status in (429, 503, 504)                         # not 400, 401, 404: those are ours

Using a rolling median as the baseline lets the limiter work without knowing the dependency's normal latency, and keeps it from shrinking during a period when everything is uniformly slower but still healthy. Excluding client errors matters just as much: a burst of 404s is not congestion, and cutting the limit because of them starves the healthy traffic.

Verify: replaying a trace with occasional slow calls shows far fewer decreases with TunedAIMD than with the basic version, while a sustained slowdown still shrinks the limit.

5. Bound the adaptation and export what it is doing

An adaptive limit is a control loop in production, and control loops need limits of their own: a floor so it cannot collapse to zero, a ceiling so a fast dependency cannot let a client flood it, and metrics so operators can see what it decided and why.

import asyncio


class ObservedAIMD(TunedAIMD):
    def snapshot(self) -> dict:
        return {
            "limit": round(self.limit, 1),
            "inflight": self.inflight,
            "utilisation": round(self.inflight / max(1.0, self.limit), 2),
            "decreases": self.decreases,
            "at_floor": int(self.limit) <= self.min,
            "at_ceiling": int(self.limit) >= self.max,
        }


async def main() -> None:
    limiter = ObservedAIMD(initial=8, minimum=2, maximum=32, latency_target=0.02)
    for _ in range(200):
        limiter.record(latency=0.005)                        # a burst of healthy calls
    print("after healthy traffic:", limiter.snapshot())
    for _ in range(5):
        limiter.record(failed=True)
        await asyncio.sleep(0.05)                            # respect one cut per round trip
    print("after failures:      ", limiter.snapshot())


asyncio.run(main())

Read the three numbers together: sustained at_ceiling means the ceiling, not the dependency, is the constraint; sustained at_floor with failures means the dependency is in trouble and a circuit breaker should take over; and utilisation well below one means the limit is not the bottleneck at all. Alert on the floor and on decrease rate rather than on the limit's absolute value, which moves constantly by design.

Verify: 200 healthy calls raise the limit from 8 to about 21 — additive increase is deliberately slow — and five failures, spaced far enough apart to count separately, walk it back down to about 5.

What each outcome should do to the limit A grid of 5 rows by 1 columns. What each outcome should do to the limit outcome effect on the limit fast success increase by 1/limit latency above baseline multiply by backoff timeout, 429, 503, 504 multiply by backoff 400, 401, 404 no change caller cancelled no change Only signals that mean "you are sending too much" may shrink the limit.

Verification

An adaptive limiter is working when:

  • It tracks capacity: after a dependency degrades, the limit settles near the new capacity within a few round trips.
  • Latency improves over a fixed limit: p95 service time under degradation is lower than with the static limit, at comparable throughput.
  • It is stable when healthy: decrease events are rare during normal operation, and the limit does not oscillate between extremes.
  • Only congestion shrinks it: client errors and cancelled requests do not count as congestion signals.
  • It is bounded and observable: floor, ceiling, current limit, utilisation and decrease rate are exported, with alerts on floor and decrease rate.

Pitfalls & edge cases

  • Adjusting a Semaphore's permits. Semaphores are built for a fixed count; changing the limit by releasing or absorbing permits invites double-release bugs. Compare in-flight against the current limit instead.
  • Counting queue time as service time. Latency measured from submission includes time waiting for the limiter, so the limiter would react to its own queue. Measure from acquiring the slot to the response.
  • Shrinking on client errors. 400, 401 and 404 say nothing about capacity; only timeouts, 429, 503 and 504 should trigger decreases.
  • No floor. Multiplicative decrease on a persistently failing dependency drives the limit to zero, which turns a partial outage into a total one. Keep a floor of one or two.
  • One limiter for several dependencies. Each dependency has its own capacity; a shared limiter mixes their signals. Use one limiter per host or per route, as in bulkhead isolation with per-dependency semaphores.

Frequently Asked Questions

What is AIMD and why use it for concurrency limits?

AIMD stands for additive increase, multiplicative decrease: raise the limit slowly while calls are fast and successful, and cut it sharply when they are slow or fail. It is the control law TCP uses for congestion, and applied to a client's concurrency limit it converges towards the dependency's current capacity without anyone configuring a number.

How do I change an asyncio concurrency limit at runtime?

Do not try to resize a Semaphore. Track in-flight calls yourself and admit a caller only while in-flight is below the current limit, waiting on an asyncio.Event that is set whenever a call finishes or the limit changes. The limit can then move up or down freely between admissions.

Which signals should shrink an adaptive concurrency limit?

Signals that indicate the dependency is overloaded: timeouts, HTTP 429, 503 and 504, and latency significantly above its recent baseline. Client errors such as 400, 401 and 404 say nothing about capacity, and reacting to them starves healthy traffic.

Do adaptive limits replace circuit breakers?

No, they complement each other. An adaptive limit reduces the load sent to a struggling dependency while still using it; a circuit breaker stops calls entirely when the dependency is failing outright. A limiter pinned at its floor with continuing failures is the signal for the breaker to open.