-
Notifications
You must be signed in to change notification settings - Fork 436
Open
Labels
performancePerformance related itemsPerformance related items
Milestone
Description
Summary
The PerformanceTracker class in mcpgateway/services/performance_tracker.py uses list.pop(0) for buffer management, which is O(n) complexity. This should be replaced with collections.deque for O(1) operations.
Impact
- CPU Overhead: O(n) operation on every timing record after buffer is full
- Latency Spikes: With 1000 samples (default max), each
pop(0)shifts all elements - Scalability: Under high load with many operations being tracked, this becomes a measurable bottleneck
Affected Code
File: mcpgateway/services/performance_tracker.py
| Location | Issue |
|---|---|
| Line 37 | self.operation_timings: Dict[str, List[float]] = defaultdict(list) - uses list requiring manual O(n) eviction |
| Lines 91-93 | track_operation context manager uses pop(0) for buffer eviction |
| Lines 125-127 | record_timing method uses pop(0) for buffer eviction |
| Lines 271-272 | check_performance_degradation uses list slicing (needs adaptation for deque) |
Proposed Solution
Replace list with collections.deque using maxlen parameter for automatic O(1) eviction:
from collections import defaultdict, deque
class PerformanceTracker:
def __init__(self):
# Must be set before creating factory (lambda captures self.max_samples)
self.max_samples = getattr(settings, "perf_max_samples_per_operation", 1000)
# Use deque with maxlen for O(1) eviction
self.operation_timings: Dict[str, deque[float]] = defaultdict(
lambda: deque(maxlen=self.max_samples)
)This eliminates the manual buffer size checking and pop(0) calls entirely.
Performance Analysis
| Operation | Current (list) | Proposed (deque) |
|---|---|---|
| Append | O(1) amortized | O(1) |
| Evict oldest | O(n) | O(1) |
| Get length | O(1) | O(1) |
With max_samples=1000 at 1000 req/s with 10 tracked operation types:
- Current: ~10,000,000 element shifts/second (worst case)
- Proposed: 0 element shifts/second
Implementation Tasks
- Import
dequefrom collections (line 14) - Update type hint
Dict[str, List[float]]→Dict[str, deque[float]](line 37) - Move
max_samplesinitialization beforeoperation_timings - Update defaultdict factory to use
lambda: deque(maxlen=self.max_samples) - Remove manual size checking in
track_operation(delete lines 91-93) - Remove manual size checking in
record_timing(delete lines 125-127) - Fix
check_performance_degradationslicing (lines 271-272) - convert to list first - Remove unused
Listfrom typing imports (line 19) - Add unit tests for buffer eviction behavior
- Verify thread safety (deque append is atomic in CPython GIL)
Acceptance Criteria
-
list.pop(0)calls removed (2 locations) -
dequewithmaxlenused for automatic eviction -
check_performance_degradationworks correctly with deque - No functional change to statistics calculations
- All existing tests pass
- New tests verify buffer size limits
- Code passes
make verify
References
📄 Full analysis: todo/performance/optimize-performance-tracker-buffer.md
Metadata
Metadata
Assignees
Labels
performancePerformance related itemsPerformance related items