Approach: Use a token-bucket that tracks current_tokens and last_refill timestamp. On each allow(n) call, compute tokens to add = min(capacity, current + elapsed * refill_rate), then if current_tokens >= n consume and return True else return False. Use monotonic clock and a lock for thread-safety.python
import time
import threading
class TokenBucket:
def __init__(self, capacity: float, refill_rate: float):
"""
capacity: max tokens
refill_rate: tokens per second
"""
self.capacity = float(capacity)
self.refill_rate = float(refill_rate)
self._tokens = float(capacity)
self._last = time.monotonic()
self._lock = threading.Lock()
def _refill(self):
now = time.monotonic()
elapsed = now - self._last
if elapsed <= 0:
return
added = elapsed * self.refill_rate
self._tokens = min(self.capacity, self._tokens + added)
self._last = now
def allow(self, n: float = 1.0) -> bool:
if n <= 0:
return True
with self._lock:
self._refill()
if self._tokens >= n:
self._tokens -= n
return True
return False
