This post is also available in Chinese (中文).

Sending HTTP requests at scale is trickier than it looks.

You write a scraper or a batch API caller. It works fine with 10 requests. Then you scale up to 10,000 and everything falls apart: 429 rate limits, event loop starvation, or your thread pool hammering the target service into the ground.

This post walks through concurrency control starting from the basics: semaphores first, then PID controllers, token buckets, and adaptive concurrency adjustment. Every approach includes complete Python code you can run immediately.

The big picture: what concurrency control is really about

Whether you use asyncio or threading, the core question is the same: how do you balance "as fast as possible" with "don't crash the system"?

Here is the general flow:

graph LR
    A[Request Queue] --> B{Concurrency Controller}
    B -->|Acquire token/semaphore| C[Worker Pool]
    C --> D[Target Service]
    D -->|Response| E{Feedback Collector}
    E -->|Success rate / latency| B
    E --> F[Result Aggregator]

    style A fill:#EFF6FF,stroke:#2563EB,stroke-width:2px,color:#1E3A5F
    style B fill:#FFFBEB,stroke:#D97706,stroke-width:2px,color:#92400E
    style C fill:#ECFDF5,stroke:#059669,stroke-width:2px,color:#065F46
    style D fill:#FEF2F2,stroke:#DC2626,stroke-width:2px,color:#991B1B
    style E fill:#FFFBEB,stroke:#D97706,stroke-width:2px,color:#92400E
    style F fill:#EFF6FF,stroke:#2563EB,stroke-width:2px,color:#1E3A5F

The key is that feedback loop: success rates and latency flow back to the controller, which adjusts concurrency in real time. That is the difference between static and dynamic concurrency control.

1. Semaphores: the simplest concurrency limiter

A semaphore is the starting point for concurrency control. The idea is dead simple: set a cap, allow at most N tasks to run concurrently, and make the rest wait.

Async semaphore

In asyncio, use asyncio.Semaphore:

import asyncio
import httpx

async def fetch_url(semaphore: asyncio.Semaphore, client: httpx.AsyncClient, url: str):
    async with semaphore:  # decrements on enter, increments on exit
        response = await client.get(url)
        return response.status_code

async def main():
    urls = [f"https://httpbin.org/delay/1" for _ in range(50)]
    semaphore = asyncio.Semaphore(10)  # max 10 concurrent requests

    async with httpx.AsyncClient(timeout=30) as client:
        tasks = [fetch_url(semaphore, client, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    success = sum(1 for r in results if isinstance(r, int) and r == 200)
    print(f"Done: {success}/{len(urls)} succeeded")

asyncio.run(main())

The critical piece: async with semaphore is a context manager. On entry, the semaphore count decreases by 1. On exit, it increases by 1. When the count hits 0, any new async with suspends the coroutine (it does not block the thread) until another task finishes and releases the semaphore.

Thread semaphore

Same principle in threaded code, slightly different API:

import concurrent.futures
import threading
import time
import random

def fetch_with_semaphore(semaphore: threading.Semaphore, task_id: int):
    with semaphore:
        print(f"[Task {task_id}] Starting")
        time.sleep(random.uniform(0.1, 0.5))  # simulate network request
        print(f"[Task {task_id}] Done")
        return task_id

def main():
    semaphore = threading.Semaphore(5)  # max 5 threads running at once

    with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
        futures = [
            executor.submit(fetch_with_semaphore, semaphore, i)
            for i in range(30)
        ]
        results = [f.result() for f in concurrent.futures.as_completed(futures)]

    print(f"Completed {len(results)} tasks")

main()

The limitation of semaphores

Semaphores are great, but they have an obvious problem: the concurrency cap is hardcoded.

You set it to 10, and it stays at 10 forever. But in reality:

• At 3 AM when the server is idle, 10 is too conservative
• At 3 PM during peak traffic, 10 might be too aggressive
• The target service just deployed a new version, and its capacity dropped temporarily

We need the concurrency limit to adapt.

2. PID controller: dynamic adjustment via feedback loop

The PID controller (Proportional-Integral-Derivative) is a classic algorithm from industrial control theory. The idea: observe the gap between the actual state and the desired state (the error), then adjust the output based on that error.

Applied to concurrency control:

• Target: keep the failure rate below a threshold (say, 5%)
• Output: concurrency level
• Feedback: actual failure rate

The three PID components

Full implementation

import asyncio
import httpx
import time
import random

class PIDController:
    """PID controller for dynamic concurrency adjustment"""

    def __init__(self, kp: float, ki: float, kd: float,
                 target: float, min_output: int, max_output: int):
        self.kp = kp          # proportional gain
        self.ki = ki          # integral gain
        self.kd = kd          # derivative gain
        self.target = target  # target failure rate
        self.min_output = min_output
        self.max_output = max_output

        self._integral = 0.0
        self._prev_error = 0.0

    def compute(self, current_value: float) -> int:
        """Compute new concurrency level from current failure rate"""
        error = self.target - current_value  # positive = room to go faster

        # P: current error
        p_term = self.kp * error

        # I: accumulated error (with anti-windup clamp)
        self._integral += error
        self._integral = max(-50, min(50, self._integral))
        i_term = self.ki * self._integral

        # D: rate of change
        d_term = self.kd * (error - self._prev_error)
        self._prev_error = error

        output = p_term + i_term + d_term
        return int(max(self.min_output, min(self.max_output, output)))


async def fetch_with_pid(url: str, client: httpx.AsyncClient, semaphore: asyncio.Semaphore):
    """Async request with semaphore gate"""
    async with semaphore:
        try:
            resp = await client.get(url, timeout=10)
            # Simulate: random failure to demonstrate PID adjustment
            if random.random() < 0.15:
                return False
            return resp.status_code == 200
        except Exception:
            return False


async def main():
    pid = PIDController(
        kp=20.0, ki=0.5, kd=5.0,
        target=0.05,       # target failure rate: 5%
        min_output=2,       # min 2 concurrent
        max_output=50       # max 50 concurrent
    )

    current_concurrency = 10
    url = "https://httpbin.org/delay/0"

    async with httpx.AsyncClient() as client:
        for batch in range(10):
            semaphore = asyncio.Semaphore(current_concurrency)
            tasks = [fetch_with_pid(url, client, semaphore) for _ in range(20)]
            results = await asyncio.gather(*tasks)

            failure_rate = sum(1 for r in results if not r) / len(results)
            new_concurrency = pid.compute(failure_rate)

            print(f"Batch {batch+1}: "
                  f"concurrency={current_concurrency}, "
                  f"failure_rate={failure_rate:.1%}, "
                  f"adjusted_to={new_concurrency}")

            current_concurrency = new_concurrency

asyncio.run(main())

PID tuning tips

Getting the PID coefficients right takes some iteration. A few rules of thumb:

1. Start with P only: set ki and kd to 0, tune kp until the system roughly tracks the target
2. Add I to eliminate steady-state error: if the failure rate hovers just above the target but never converges, add a small ki
3. Add D to dampen oscillation: if concurrency is swinging wildly between batches, increase kd
4. Always include anti-windup: clamp the integral term to prevent saturation, or the controller will overshoot badly after sustained error

3. Token bucket: controlling request rate

A semaphore controls how many requests are in flight at once. A token bucket controls how many requests you can send per second. These are two different dimensions.

Token bucket vs. semaphore

Example: a semaphore of 10 means at most 10 concurrent requests. But if each request completes in 10ms, you are effectively sending 1,000 requests per second. The semaphore does not care about that.

The token bucket approach:

• Tokens are generated at a fixed rate (say, 100 per second)
• Each request consumes one token
• When the bucket is full, no more tokens are generated (there is a capacity cap)
• No token available? Wait.

Thread-safe token bucket

import threading
import time

class TokenBucket:
    """Thread-safe token bucket"""

    def __init__(self, rate: float, capacity: int):
        self.rate = rate          # tokens generated per second
        self.capacity = capacity  # max tokens in the bucket
        self._tokens = float(capacity)
        self._last_refill = time.monotonic()
        self._lock = threading.Lock()

    def _refill(self):
        """Refill tokens based on elapsed time"""
        now = time.monotonic()
        elapsed = now - self._last_refill
        new_tokens = elapsed * self.rate
        self._tokens = min(self.capacity, self._tokens + new_tokens)
        self._last_refill = now

    def acquire(self, timeout: float = None) -> bool:
        """Acquire one token. Returns True on success, False on timeout."""
        deadline = time.monotonic() + timeout if timeout else None

        while True:
            with self._lock:
                self._refill()
                if self._tokens >= 1:
                    self._tokens -= 1
                    return True

            if deadline and time.monotonic() >= deadline:
                return False
            time.sleep(0.01)  # retry every 10ms


class RateLimitedExecutor:
    """Task executor with rate limiting"""

    def __init__(self, rate: float, capacity: int, max_workers: int):
        self.bucket = TokenBucket(rate, capacity)
        self.max_workers = max_workers

    def execute(self, task_fn, task_args_list):
        """Execute a batch of tasks with rate limiting"""
        import concurrent.futures

        def wrapped(args):
            if self.bucket.acquire(timeout=30):
                return task_fn(*args)
            else:
                raise TimeoutError("Token acquisition timed out")

        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as pool:
            futures = [pool.submit(wrapped, args) for args in task_args_list]
            return [f.result() for f in concurrent.futures.as_completed(futures)]


# Usage
def call_api(task_id: int):
    print(f"[{time.strftime('%H:%M:%S')}] Task {task_id} executing")
    time.sleep(0.1)
    return f"Task {task_id} done"

executor = RateLimitedExecutor(
    rate=10,       # 10 requests per second
    capacity=20,   # burst capacity of 20
    max_workers=5
)

results = executor.execute(call_api, [(i,) for i in range(30)])
print(f"\nCompleted {len(results)} tasks")

Async token bucket

If you are using async code, implement the token bucket with asyncio.sleep instead of busy-waiting:

import asyncio
import time

class AsyncTokenBucket:
    """Async token bucket"""

    def __init__(self, rate: float, capacity: int):
        self.rate = rate
        self.capacity = capacity
        self._tokens = float(capacity)
        self._last_refill = time.monotonic()

    async def acquire(self):
        """Acquire a token (suspends the coroutine, does not block the thread)"""
        while True:
            now = time.monotonic()
            elapsed = now - self._last_refill
            self._tokens = min(self.capacity, self._tokens + elapsed * self.rate)
            self._last_refill = now

            if self._tokens >= 1:
                self._tokens -= 1
                return True

            # Calculate wait time until next token
            wait_time = (1 - self._tokens) / self.rate
            await asyncio.sleep(wait_time)

4. Adaptive concurrency: probe the limit from scratch

All the previous approaches require you to know something in advance — a concurrency cap, a target failure rate, a token rate. But what if you have no idea how much the target service can handle?

Adaptive concurrency control has a simple premise: start low, ramp up, and back off when errors appear. It is essentially TCP congestion control applied to HTTP requests.

Core logic

concurrency = 1
after each batch:
  if failure_rate < threshold → concurrency += step  (probe faster)
  if failure_rate > threshold → concurrency = concurrency / 2  (back off fast)
  concurrency = clamp(concurrency, min, max)

Full implementation

import asyncio
import httpx
import random

class AdaptiveConcurrencyController:
    """Adaptive concurrency controller — automatically probes optimal concurrency"""

    def __init__(self, min_concurrency: int = 1, max_concurrency: int = 100,
                 failure_threshold: float = 0.1, step: int = 2):
        self.min_concurrency = min_concurrency
        self.max_concurrency = max_concurrency
        self.failure_threshold = failure_threshold
        self.step = step

        self.current_concurrency = min_concurrency
        self._history = []

    def adjust(self, failure_rate: float):
        """Adjust concurrency based on failure rate"""
        old = self.current_concurrency

        if failure_rate <= self.failure_threshold:
            # Room to spare — speed up
            self.current_concurrency = min(
                self.max_concurrency,
                self.current_concurrency + self.step
            )
        else:
            # Over threshold — back off fast (halve)
            self.current_concurrency = max(
                self.min_concurrency,
                self.current_concurrency // 2
            )

        self._history.append((old, failure_rate, self.current_concurrency))
        return self.current_concurrency

    def report(self):
        """Print adjustment history"""
        print("\n--- Concurrency Adjustment History ---")
        print(f"{'Round':>5} | {'Concurrency':>11} | {'Failure%':>8} | {'Adjusted':>8}")
        print("-" * 45)
        for i, (old, rate, new) in enumerate(self._history, 1):
            arrow = "^" if new > old else ("v" if new < old else "=")
            print(f"{i:>5} | {old:>11} | {rate:>7.1%} | {new:>7} {arrow}")


async def fetch(client: httpx.AsyncClient, semaphore: asyncio.Semaphore, url: str):
    async with semaphore:
        try:
            resp = await client.get(url, timeout=10)
            # Simulate: higher concurrency = higher failure rate
            simulated_failure = random.random() < (semaphore._value / 100)
            return not simulated_failure and resp.status_code == 200
        except Exception:
            return False


async def main():
    controller = AdaptiveConcurrencyController(
        min_concurrency=1,
        max_concurrency=50,
        failure_threshold=0.15,  # 15% failure rate threshold
        step=3
    )

    url = "https://httpbin.org/delay/0"
    batch_size = 20

    async with httpx.AsyncClient() as client:
        for round_num in range(15):
            concurrency = controller.current_concurrency
            semaphore = asyncio.Semaphore(concurrency)

            tasks = [fetch(client, semaphore, url) for _ in range(batch_size)]
            results = await asyncio.gather(*tasks)

            failure_rate = sum(1 for r in results if not r) / len(results)
            controller.adjust(failure_rate)

    controller.report()

asyncio.run(main())

When you run this, you will see concurrency ramp up steadily, then drop sharply when the failure rate exceeds the threshold, and eventually oscillate around a stable range. That is the adaptive behavior in action.

5. Choosing the right approach

Practical advice

1. Start with a fixed semaphore. In most cases, a conservative concurrency cap is all you need. Do not over-engineer.
2. Instrument first. Regardless of approach, log success rates, response times, and concurrency levels. You cannot tune what you cannot measure.
3. Token buckets and semaphores stack. Semaphores control how many requests are in flight. Token buckets control how many requests are sent per second. They solve different problems and combine naturally.
4. Retries need backoff. Concurrency control solves "how fast to send." Retry strategy solves "what to do on failure." Use exponential backoff with jitter.
5. Test and production differ wildly. Parameters that work locally will behave differently in production. Build in the ability to adjust at runtime.

Conclusion

Concurrency control is not rocket science. At its core, it is one question: how do you balance speed against stability?

• Semaphores are the simplest answer, sufficient for most cases
• Token buckets solve the rate control problem
• PID controllers use feedback loops for fine-grained adjustment
• Adaptive concurrency automatically probes the optimum when you are flying blind

Start simple. Upgrade when you hit real problems. Reaching for the most complex solution first is itself a failure of complexity management.