Publicado · en mejora
Guía de Limitación de tasa · 5/6
Por ahora, este capítulo solo está disponible en inglés.
Rate limiters show up more often in system design interviews and production code than in classic coding tests. The four problems below are small twists on the algorithms from the previous chapters. Try each one yourself before reading the solution.
You have the request timestamps from an API server's log, as integer seconds in ascending order. The rule is "no window-second span may contain more than limit requests". Return the timestamp of the first request that breaks the rule, or None if the rule always held.
Approach: this is a sliding window log. Drop timestamps that fell out of the window, append the current one, and the first time the count exceeds limit is the answer. Every timestamp enters and leaves once, so the whole pass is O(n).
from collections import deque
def first_violation(times: list[int], limit: int, window: int) -> int | None:
recent: deque[int] = deque()
for t in times:
while recent and recent[0] <= t - window:
recent.popleft()
recent.append(t)
if len(recent) > limit:
return t
return None
print(first_violation([1, 2, 11, 12, 13, 14], limit=3, window=10)) # 14
print(first_violation([1, 5, 11, 15], limit=2, window=10)) # NoneAt time 14 the span (4, 14] holds 11, 12, 13 and 14, four requests, which is the first time the limit of 3 is exceeded.
A token bucket charges 5 tokens for a search and 1 token for a read. Given a list of (time, cost) requests, return None for each accepted request and, for each rejected one, the whole number of seconds to put in Retry-After. The bucket starts full at time 0.
Approach: refill lazily, then compare against the cost. The time to earn the missing tokens is (cost - tokens) / rate; the header takes whole seconds, so round up. Rounding down makes the client retry too early and get rejected again. A request that costs more than the capacity can never pass, so handle it separately.
import math
def simulate(requests: list[tuple[float, float]], rate: float,
capacity: float) -> list[int | None]:
tokens, last = capacity, 0.0
out: list[int | None] = []
for t, cost in requests:
if cost > capacity:
raise ValueError(f"cost {cost} can never fit in capacity {capacity}")
tokens = min(capacity, tokens + (t - last) * rate)
last = t
if tokens >= cost:
tokens -= cost
out.append(None)
else:
out.append(math.ceil((cost - tokens) / rate))
return out
print(simulate([(0, 5), (0, 5), (1, 1), (2, 5)], rate=2, capacity=10))
# [None, None, None, 1]At time 2 only 3 tokens are left, two short. Tokens come back at 2 per second, so the client should retry in 1 second.
Apply both "3 per second" and "5 per minute" to the same user. A request is allowed only if it passes both rules.
Approach: keep one log per rule, but check every rule first and record into every log only when all of them pass. If you record into the first log and then reject on the second rule, rejected requests eat into the first limit, which is a classic bug. The same "check all, then charge all" rule applies when combining several token buckets.
from collections import deque
class MultiLimit:
def __init__(self, rules: list[tuple[int, float]]) -> None:
self.rules = rules # [(limit, window), ...]
self.logs = [deque() for _ in rules]
def allow(self, now: float) -> bool:
for (limit, window), log in zip(self.rules, self.logs):
while log and log[0] <= now - window:
log.popleft()
if len(log) >= limit:
return False # reject without recording
for log in self.logs:
log.append(now) # record only when all pass
return True
limiter = MultiLimit([(3, 1.0), (5, 60.0)])
print([limiter.allow(t) for t in [0, 0.1, 0.2, 0.3, 1.5, 1.6, 1.7, 3.0]])
# [True, True, True, False, True, True, False, False]Compute how long a client should wait after receiving a 429. If the server sent Retry-After, never retry earlier than that; otherwise use exponential backoff with jitter.
Approach: if every client retries after exactly 1, 2 and 4 seconds, they all come back at the same instant. Picking a random delay between 0 and min(cap, base × 2^attempt), known as full jitter, spreads retries out evenly.
import random
def next_delay(attempt: int, retry_after: float | None,
base: float = 0.5, cap: float = 30.0) -> float:
backoff = random.uniform(0, min(cap, base * 2 ** attempt)) # full jitter
if retry_after is not None:
return max(retry_after, backoff)
return backoff
random.seed(7)
for attempt in range(5):
print(attempt, round(next_delay(attempt, None), 2)) # random, capped at 0.5, 1, 2, 4, 8
print(next_delay(0, retry_after=12)) # 12O(n).Retry-After is the time to earn the missing tokens, rounded up.Retry-After and otherwise retry with jittered exponential backoff.
0 comentarios
Iniciar sesión · Inicia sesión para dejar un comentario.
Sé el primero en comentar.