refactor: remove threading from metrics

The server is pure asyncio (single-threaded). The threading.Lock
was never contended. Use a float accumulator in _human_bytes to
avoid the int-to-float type: ignore.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
user
2026-02-15 21:48:47 +01:00
parent 210d3539f1
commit 18789bbc63

View File

@@ -2,19 +2,17 @@
from __future__ import annotations
import threading
import time
class Metrics:
"""Thread-safe connection metrics.
"""Connection metrics.
Counters use a lock for consistency but tolerate minor races
on hot paths (bytes_in/bytes_out) for performance.
All access runs on the single asyncio event-loop thread,
so no locking is needed.
"""
def __init__(self) -> None:
self._lock = threading.Lock()
self.connections: int = 0
self.success: int = 0
self.failed: int = 0
@@ -26,7 +24,6 @@ class Metrics:
def summary(self) -> str:
"""One-line log-friendly summary."""
with self._lock:
uptime = time.monotonic() - self.started
h, rem = divmod(int(uptime), 3600)
m, s = divmod(rem, 60)
@@ -40,10 +37,11 @@ class Metrics:
def _human_bytes(n: int) -> str:
"""Format byte count in human-readable form."""
value = float(n)
for unit in ("B", "K", "M", "G", "T"):
if abs(n) < 1024:
if abs(value) < 1024:
if unit == "B":
return f"{n}{unit}"
return f"{n:.1f}{unit}"
n /= 1024 # type: ignore[assignment]
return f"{n:.1f}P"
return f"{int(value)}{unit}"
return f"{value:.1f}{unit}"
value /= 1024
return f"{value:.1f}P"