Viewed
Coding
Google
Gemini 2.5 Pro
VS
OpenAI
GPT-5.2
Implement a Concurrent Rate Limiter with Sliding Window and Priority Queues
Design and implement a thread-safe rate limiter in Python that supports the following features:
1. **Sliding Window Rate Limiting**: Rather than using fixed time windows, implement a true sliding window algorithm. Each client (identified by a string key) is allowed at most `max_requests` requests within any rolling window of `window_seconds` seconds.
2. **Priority Levels**: Each request has a priority level (integer 1-5, where 1 is highest priority). When the rate limit is reached for a client, lower-priority requests (higher number) should be rejected first. Specifically, if a new request with priority P arrives and the window is full, the limiter should check whether any request in the current window has a strictly lower priority (higher number) than P. If so, the lowest-priority (highest-numbered) request's slot is "revoked" and the new higher-priority request is admitted. The revoked request should be recorded so it can be reported. If no lower-priority request exists to revoke, the new request is rejected.
3. **Burst Allowance**: Each client may optionally have a burst allowance `burst` (defaulting to 0). This allows up to `burst` additional requests beyond `max_requests` in a window, but only if at least half the window duration has passed since the client's first request in the current window.
4. **Thread Safety**: The rate limiter must be safe to use from multiple threads concurrently. Demonstrate this with a test scenario.
5. **Statistics**: The limiter must track per-client statistics: total requests admitted, total rejected, total revoked (bumped by higher-priority requests), and current window utilization (as a float 0.0 to 1.0).
Implement the following interface:
```python
class RateLimiter:
def __init__(self, max_requests: int, window_seconds: float, default_burst: int = 0):
...
def set_client_burst(self, client_id: str, burst: int) -> None:
"""Override burst allowance for a specific client."""
...
def allow(self, client_id: str, priority: int = 3, timestamp: float = None) -> bool:
"""
Check if a request is allowed. If timestamp is None, use current time.
Returns True if the request is admitted, False if rejected.
"""
...
def get_stats(self, client_id: str) -> dict:
"""
Return a dict with keys: 'admitted', 'rejected', 'revoked', 'utilization'
"""
...
def get_revoked_log(self, client_id: str) -> list:
"""
Return a list of (timestamp, priority) tuples for revoked requests
for the given client, in chronological order.
"""
...
```
Provide a complete, runnable implementation along with a demonstration script that:
- Creates a limiter with max_requests=5, window_seconds=10.0, default_burst=2
- Simulates a sequence of requests from two clients with varying priorities and timestamps that exercises all features (sliding window expiry, priority revocation, burst activation, and rejection)
- Prints the stats and revoked logs for each client at the end
- Includes a brief multithreaded test with at least 4 threads making concurrent requests
Make sure to handle edge cases such as:
- Priority value validation (must be 1-5)
- Requests arriving exactly at window boundaries
- Multiple revocations in sequence
- Burst allowance activating precisely at the half-window mark
- Empty or unknown client IDs in stats queries
