INNER CODE UNIT · Python

_TrafficState

lij768423-svg/grok-register-panel · batch_traffic.py:252

class _TrafficState:
    def __init__(self, path: Path, batch_id: str):
        self.path = path
        self.batch_id = batch_id
        self.lock = threading.RLock()
        self.stop_event = threading.Event()
        existing = read_metrics(path)
        if existing.get("batch_id") != batch_id:
            existing = initialize_batch(path, batch_id, target=1, workers=1)
        self.metrics = existing
        self.metrics["running"] = True
        self.metrics["active_connections"] = 0
        self.writer = threading.Thread(target=self._writer_loop, daemon=True)
        self.writer.start()

    def _writer_loop(self) -> None:
        while not self.stop_event.wait(0.5):
            self.flush()

View source record →