🎫 Pattern 11: Rate Limiting & Concurrency Caps

The Problem: Two Different "Too Fast"

There are two distinct knobs, and mixing them up is the #1 mistake here:

• Concurrency limit: "no more than N requests in flight at the same time." A count of simultaneous work. → Semaphore.
• Rate limit: "no more than R requests per second," regardless of how many are in flight. A count over time. → token bucket.

Capping Concurrency with a Semaphore

A threading.Semaphore(N) is a counter that starts at N. acquire() decrements it (blocking at zero); release() increments it. Wrap the critical work in with sem: and at most N threads can be inside at once — everyone else waits at the gate.

import threading, time, random

sem = threading.Semaphore(3)          # at most 3 workers in the critical section
active = 0
active_lock = threading.Lock()

def fetch(url):
    global active
    with sem:                         # blocks here once 3 are already inside
        with active_lock:
            active += 1
            print(f"{url}: in flight (now {active} active)")
        time.sleep(random.uniform(0.2, 0.6))   # simulate the request
        with active_lock:
            active -= 1
    # permit auto-released on exit — even if the body raised

threads = [threading.Thread(target=fetch, args=(f"url-{i}",)) for i in range(8)]
for t in threads: t.start()
for t in threads: t.join()
# You'll never see "now 4 active" — the cap holds.

Why a Semaphore Is Not a Rate Limiter

The trap: "I set Semaphore(5), so I'm doing 5 per second." No. You're doing 5 at a time. Watch the arithmetic:

To bound requests per unit of time, you need something that tracks the clock. Enter the token bucket.

A Thread-Safe Token Bucket

The token bucket is the standard rate-limiting algorithm. A bucket holds up to capacity tokens and refills at rate tokens per second. Each request must acquire() a token; if the bucket is empty, the caller sleeps until enough have dripped back in. The capacity is your burst allowance; the rate is your steady-state ceiling.

import threading, time

class TokenBucket:
    """Rate-limit to `rate` operations/sec, allowing bursts up to `capacity`."""

    def __init__(self, rate: float, capacity: float):
        self.rate = rate                 # tokens added per second
        self.capacity = capacity         # max tokens (burst size)
        self.tokens = capacity           # start full
        self.updated_at = time.monotonic()
        self.lock = threading.Lock()

    def _refill(self):
        now = time.monotonic()
        elapsed = now - self.updated_at
        # add tokens for the time that passed, never exceeding capacity
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.updated_at = now

    def acquire(self, amount: float = 1.0):
        while True:
            with self.lock:
                self._refill()
                if self.tokens >= amount:
                    self.tokens -= amount
                    return
                # not enough yet — compute how long until we have `amount`
                deficit = amount - self.tokens
                wait = deficit / self.rate
            time.sleep(wait)             # sleep OUTSIDE the lock

bucket = TokenBucket(rate=5, capacity=5)   # 5/sec steady, burst of 5

def call_api(i):
    bucket.acquire()                        # blocks to stay under the limit
    print(f"{time.monotonic():.2f}s  request {i}")

threads = [threading.Thread(target=call_api, args=(i,)) for i in range(15)]
for t in threads: t.start()
for t in threads: t.join()
# First 5 fire immediately (the burst), then ~5 per second after.

Combining Both with a ThreadPoolExecutor

In real code you rarely hand-roll threads — you use a pool. The pool's max_workers is your concurrency cap (it plays the semaphore's role), and the token bucket, called at the top of each task, enforces the rate. Two independent knobs, cleanly separated.

from concurrent.futures import ThreadPoolExecutor
import time

bucket = TokenBucket(rate=10, capacity=10)   # ≤ 10 req/s

def fetch(url):
    bucket.acquire()                          # rate gate — may block
    # ... do the actual request here ...
    return f"fetched {url}"

urls = [f"https://api.example.com/item/{i}" for i in range(100)]

# max_workers=5  → never more than 5 requests IN FLIGHT (concurrency cap)
# bucket rate=10 → never more than 10 requests PER SECOND (rate cap)
with ThreadPoolExecutor(max_workers=5) as pool:
    results = list(pool.map(fetch, urls))

print(len(results), "done")

Key Takeaways