| """Reserved asynchronous summary logging; disk work stays off the control thread.""" |
| from __future__ import annotations |
| from collections import Counter |
| import csv |
| import json |
| from pathlib import Path |
| import queue |
| import threading |
| import numpy as np |
| from .events import ADJUDICATION_VERSION |
| from .metrics import BallMetrics, REWARD_FIELDS, format_ball |
|
|
|
|
| def json_default(value): |
| if isinstance(value, np.ndarray): |
| return value.tolist() |
| if isinstance(value, np.generic): |
| return value.item() |
| raise TypeError(type(value).__name__) |
|
|
|
|
| class AsyncJournal: |
| def __init__(self, root, capacity): |
| self.root = Path(root) |
| self.queue = queue.Queue() |
| self.slots = threading.BoundedSemaphore(capacity) |
| self.error = None |
| self.closed = False |
| self.dropped_balls = Counter() |
| self.dropped_telemetry = 0 |
| self.submitted_balls = self.written_balls = 0 |
| self.metrics = BallMetrics() |
| self.thread = threading.Thread(target=self._run, name="h2-rl-journal", daemon=True) |
| self.thread.start() |
|
|
| def reserve(self): |
| return not self.closed and self.error is None and self.slots.acquire(blocking=False) |
|
|
| def missed_ball(self, phase): |
| self.dropped_balls[phase] += 1 |
| if sum(self.dropped_balls.values()) == 1: |
| print("[RL ERROR] ball summary unavailable; unlogged attempts are counted in journal_status.json") |
|
|
| def submit_reserved(self, record): |
| if record.get("kind", "ball") == "ball": |
| self.submitted_balls += 1 |
| self.queue.put_nowait(record) |
|
|
| def telemetry(self, record): |
| if self.reserve(): |
| self.submit_reserved(dict(kind="telemetry", summary=record)) |
| else: |
| self.dropped_telemetry += 1 |
|
|
| def snapshot(self): |
| return dict(unlogged_attempts=dict(self.dropped_balls), |
| submitted_balls=self.submitted_balls, written_balls=self.written_balls, |
| unwritten_balls=self.submitted_balls-self.written_balls, |
| dropped_telemetry=self.dropped_telemetry, error=self.error) |
|
|
| def _run(self): |
| try: |
| with (self.root / "balls.jsonl").open("a", encoding="utf-8") as balls, \ |
| (self.root / "control.jsonl").open("a", encoding="utf-8") as control, \ |
| (self.root / "metrics.csv").open("w", newline="", encoding="utf-8") as metrics: |
| template = dict(kind="", phase="", attempt="", **dict.fromkeys(REWARD_FIELDS), |
| **self.metrics.phases["warmup"].snapshot()) |
| writer = csv.DictWriter(metrics, fieldnames=list(template)) |
| writer.writeheader() |
| while True: |
| item = self.queue.get() |
| if item is None: |
| for phase, stats in self.metrics.phases.items(): |
| writer.writerow(dict(kind="final", phase=phase, attempt="", |
| **stats.snapshot(self.dropped_balls[phase]))) |
| return |
| try: |
| summary = dict(item["summary"], adjudication_version=ADJUDICATION_VERSION) |
| is_ball = item.get("kind", "ball") == "ball" |
| stream = balls if is_ball else control |
| stream.write(json.dumps(summary, default=json_default, allow_nan=False) + "\n") |
| stream.flush() |
| if is_ball: |
| self.written_balls += 1 |
| row = self.metrics.add(summary, self.dropped_balls[summary["phase"]]) |
| writer.writerow(row) |
| metrics.flush() |
| print(format_ball(summary, row), flush=True) |
| finally: |
| self.slots.release() |
| except Exception as error: |
| self.error = str(error) |
| print(f"[RL ERROR] journal failed: {error}", flush=True) |
|
|
| def close(self): |
| if self.closed: |
| return |
| self.closed = True |
| self.queue.put_nowait(None) |
| self.thread.join() |
| (self.root / "journal_status.json").write_text( |
| json.dumps(self.snapshot(), indent=2), encoding="utf-8") |
| if self.error: |
| raise RuntimeError(f"RL journal failed: {self.error}") |
|
|
