Per-Host Concurrency Limits With Asyncio Semaphores #

Rate limiting controls how often requests leave; concurrency limiting controls how many are in flight at once. A crawler that enforces only the first can still hold two hundred open connections against a struggling host while its request rate looks entirely compliant. This guide implements per-host concurrency bounds in an asyncio crawler, as part of implementing polite rate limiting in the Compliance & Ethical Crawling Foundations section.

Problem Framing #

The failure appears when a host slows down. At ten requests per second with responses arriving in 100 ms, roughly one request is in flight at any moment. When the same host degrades to ten-second responses, the departure rate is unchanged — the limiter is still permitting ten per second — and in-flight requests accumulate to a hundred. The crawler is now consuming a hundred connections, a hundred sockets, and a hundred worker slots on a host that is visibly struggling, and from the host’s perspective the crawler has increased its load by two orders of magnitude at exactly the wrong moment.

Rate alone lets a degrading host attract two orders more concurrency.

A concurrency bound converts that into backpressure. With a semaphore of two, the eleventh request simply waits for a slot, the departure rate falls automatically as responses slow, and the crawler’s footprint on a degraded host shrinks instead of growing.

Step-by-Step Implementation #

1. One semaphore per host, created lazily #

Slot before token: budget is never spent on a request still queueing.

import asyncio
from collections import defaultdict

class HostConcurrency:
    """Per-host semaphores, created on demand and sized from the registry."""

    def __init__(self, default_limit: int = 2):
        self.default_limit = default_limit
        self._limits: dict[str, int] = {}
        self._semaphores: dict[str, asyncio.Semaphore] = {}
        self._lock = asyncio.Lock()

    def configure(self, host: str, limit: int) -> None:
        self._limits[host] = limit

    async def semaphore(self, host: str) -> asyncio.Semaphore:
        sem = self._semaphores.get(host)
        if sem is not None:
            return sem
        async with self._lock:
            # Re-check inside the lock: another task may have created it.
            if host not in self._semaphores:
                limit = self._limits.get(host, self.default_limit)
                self._semaphores[host] = asyncio.Semaphore(limit)
            return self._semaphores[host]

The double check inside the lock matters. Without it, two coroutines racing on first sight of a host create two semaphores, and the second one silently doubles the permitted concurrency for however long it survives.

2. Combine the semaphore with the rate limiter #

Both gates must be satisfied, and the order is deliberate: acquire the concurrency slot first, then the rate token. Reversing them means a task can hold a rate token — which has already been consumed from the budget — while queueing for a slot, so the budget is spent on requests that have not departed.

import time

async def fetch(session, url: str, host: str, gates) -> "Response":
    sem = await gates.concurrency.semaphore(host)
    async with sem:                                   # 1. take an in-flight slot
        wait = gates.limiter.acquire(host)            # 2. then the departure token
        if wait:
            await asyncio.sleep(wait)
        started = time.monotonic()
        try:
            return await session.get(url)
        finally:
            gates.observe(host, time.monotonic() - started)

Holding the semaphore across the rate wait is intentional: a task that is waiting for its departure slot is already committed to this host, and counting it against the concurrency bound keeps the total number of tasks queued per host bounded too.

3. Adapt the limit to observed health #

A static bound of two is a safe default, but a host that is comfortably fast can support more, and one that is degrading should support fewer. Adjusting the limit from observed latency gives the crawler a gentle, automatic response.

class AdaptiveConcurrency(HostConcurrency):
    """Shrink on slowness or refusal, grow slowly when the host is healthy."""

    def __init__(self, default_limit: int = 2, hard_max: int = 4):
        super().__init__(default_limit)
        self.hard_max = hard_max
        self._current: dict[str, int] = {}

    async def resize(self, host: str, new_limit: int) -> None:
        new_limit = max(1, min(self.hard_max, new_limit))
        current = self._current.get(host, self.default_limit)
        if new_limit == current:
            return
        sem = await self.semaphore(host)
        if new_limit > current:
            for _ in range(new_limit - current):
                sem.release()                  # widen: hand out extra permits
        else:
            for _ in range(current - new_limit):
                asyncio.create_task(self._absorb(sem))   # narrow: consume permits
        self._current[host] = new_limit

    @staticmethod
    async def _absorb(sem: asyncio.Semaphore) -> None:
        await sem.acquire()                    # taken and never released: one fewer slot

Narrowing is asynchronous by necessity — a permit currently held by an in-flight request cannot be reclaimed until it returns — so the reduction takes effect as requests complete rather than instantly. That is the desired behaviour: it never cancels work in progress, it simply stops replacing it.

Keep hard_max genuinely small. The point of the adaptive layer is to shrink promptly under strain, not to discover how much concurrency a host will tolerate.

4. Wire it to the health signal #

def on_outcome(self, host: str, outcome: str, latency: float, p95: float) -> None:
    current = self._current.get(host, self.default_limit)
    if outcome in ("throttled", "refused"):
        asyncio.create_task(self.resize(host, 1))            # straight to the floor
    elif latency > p95 * 1.5:
        asyncio.create_task(self.resize(host, current - 1))  # degrading: back off one
    elif latency < p95 * 0.6:
        asyncio.create_task(self.resize(host, current + 1))  # healthy: one more, slowly

The asymmetry mirrors the rate limiter’s: collapse to one slot immediately on a refusal, grow by a single slot at a time when the host looks comfortable. A crawler that halves its concurrency on the first 429 and takes ten minutes to earn it back is behaving the way a considerate client should.

Verification & Testing #

Concurrency bugs are timing bugs, so assert on observed peak concurrency rather than on configuration.

A leaked permit silently and permanently reduces a host’s throughput.

import asyncio
import pytest

@pytest.mark.asyncio
async def test_never_exceeds_the_bound():
    gates = build_gates(concurrency=2, rate=100.0)
    in_flight, peak = 0, 0

    async def one():
        nonlocal in_flight, peak
        sem = await gates.concurrency.semaphore("example.com")
        async with sem:
            in_flight += 1
            peak = max(peak, in_flight)
            await asyncio.sleep(0.05)
            in_flight -= 1

    await asyncio.gather(*(one() for _ in range(50)))
    assert peak <= 2

@pytest.mark.asyncio
async def test_slow_host_reduces_departure_rate():
    gates = build_gates(concurrency=2, rate=100.0)
    started = time.monotonic()
    await asyncio.gather(*(fetch(slow_session, "/x", "example.com", gates)
                           for _ in range(10)))
    # 10 requests, 2 at a time, 1 s each ⇒ at least 5 s despite a 100/s rate limit.
    assert time.monotonic() - started >= 5.0

The second test is the important one: it demonstrates that concurrency, not the rate limit, is what actually bounds the load on a slow host.

Compliance & Operational Guardrails #

The default concurrency bound is small — one or two — and raising it for a host needs a recorded reason.
A refusal collapses concurrency to one immediately, alongside the rate reduction.
The bound is shared across the fleet where workers are distributed, not applied per process.
In-flight count per host is exported as a gauge so a growing footprint is visible.
Concurrency is never widened to compensate for a slow host.

Common Mistakes #

A global semaphore instead of per-host. It bounds your own resource use and does nothing for any individual target, which is the only bound the target can perceive.
Acquiring the rate token before the concurrency slot. Budget is consumed by requests that have not left, so the observed rate falls below the configured one for no benefit.
Treating a per-process bound as a fleet bound. Six workers each permitting two connections present twelve to the host.

Frequently Asked Questions #

How do I enforce this across separate processes? #

Move the permit count into the shared store the rate limiter already uses: a counter incremented on acquire and decremented on release, with a TTL so a crashed worker’s permits are reclaimed. The local semaphore then bounds the process and the shared counter bounds the fleet, and a request needs both.

Is two really the right default? #

It is a defensible starting point for an unfamiliar host, and it is what most published crawler guidance implies. Raise it only where the host has told you it is fine — an agreement, a documented API limit, or a Crawl-delay so short that it plainly anticipates parallel access — and record the reason in the registry entry rather than in a constant.

What about HTTP/2 multiplexing? #

Multiplexed streams over one connection still represent concurrent work for the origin, so count streams rather than connections. The transport being efficient does not make the server’s rendering or database load any lighter, and it is the origin’s cost that the bound exists to limit.

What happens to the semaphore when a request times out? #

Nothing special, provided the acquisition uses a context manager: the slot is released when the block exits, whether it exits normally or by exception. The failure to avoid is acquiring and releasing manually across a code path that can raise between them, which leaks a permit permanently and silently reduces the host’s concurrency by one. After a few such leaks the crawler stops fetching that host entirely, and the symptom — a host that mysteriously stalls after a period of instability — is genuinely hard to diagnose from the outside.

Should concurrency be reduced for hosts that are simply slow? #

Yes, and the adaptive layer above does exactly that. A host answering in fifteen seconds is either overloaded or rate-limiting you by tarpitting, and in both cases holding several connections open makes it worse. Reducing to one slot costs you very little throughput — the host was slow anyway — and removes most of the resource pressure your crawl was contributing. Growing back gradually once latency recovers keeps the adjustment from oscillating.

Implementing Polite Rate Limiting — the departure-rate layer this complements.
Token Bucket vs Leaky Bucket for Crawlers — choosing the algorithm that governs departures.
Connection Pool Tuning for httpx and requests — matching the transport’s pool to the concurrency you intend.