Serialize simulation CSV writes behind a recorder (#10)

src/server/communication/inference_recorder.py

@@ -0,0 +1,132 @@
+"""Recording of inference-cycle side outputs (simulation CSV, debug JSON).
+
+These writes happen on the background I/O thread while the ASGI threads open,
+close and reset the same handles, so every access to the CSV state goes through
+one lock. Rows queued before a close are dropped rather than written to a closed
+file: telemetry must never take down the inference path.
+"""
+
+import csv
+import json
+import os
+import shutil
+import threading
+
+from server.logger.log import logger
+
+SIMULATION_CSV_FIELDNAMES = [
+    "inference_id",
+    "timestamp",
+    "avg_device_time",
+    "min_device_time",
+    "max_device_time",
+    "avg_edge_time",
+    "min_edge_time",
+    "max_edge_time",
+    "num_device_layers",
+    "num_edge_layers",
+]
+
+DEBUG_DIR = "data/models/debug"
+
+
+class InferenceCycleRecorder:
+    """Owns the simulation CSV handle, its row counter, and debug JSON output."""
+
+    def __init__(self, debug_dir: str = DEBUG_DIR):
+        self._lock = threading.RLock()
+        self._csv_file = None
+        self._csv_writer = None
+        self._inference_counter = 0
+        self.debug_dir = debug_dir
+
+    # ── simulation CSV lifecycle ────────────────────────────────────────
+    @property
+    def is_recording(self) -> bool:
+        with self._lock:
+            return self._csv_writer is not None
+
+    def open_simulation_csv(self, csv_file_path) -> bool:
+        """Start recording simulation results, replacing any open file."""
+        with self._lock:
+            try:
+                self._close_locked()
+                self._csv_file = open(csv_file_path, "w", newline="")
+                self._csv_writer = csv.DictWriter(
+                    self._csv_file, fieldnames=SIMULATION_CSV_FIELDNAMES
+                )
+                self._csv_writer.writeheader()
+                self._csv_file.flush()
+                self._inference_counter = 0
+                logger.info(f"Simulation CSV recording enabled: {csv_file_path}")
+                return True
+            except Exception as e:
+                logger.error(f"Failed to setup simulation CSV: {e}")
+                # Never leave a half-open handle behind on failure.
+                self._csv_file = None
+                self._csv_writer = None
+                return False
+
+    def close_simulation_csv(self) -> None:
+        with self._lock:
+            self._close_locked()
+            self._inference_counter = 0
+
+    def _close_locked(self) -> None:
+        if self._csv_file is not None:
+            try:
+                self._csv_file.close()
+            except Exception as e:
+                logger.warning(f"Failed to close simulation CSV: {e}")
+        self._csv_file = None
+        self._csv_writer = None
+
+    # ── row recording ───────────────────────────────────────────────────
+    def next_inference_id(self) -> int:
+        """Reserve the next row id. Concurrent requests must not collide."""
+        with self._lock:
+            self._inference_counter += 1
+            return self._inference_counter
+
+    def write_row(self, row: dict) -> bool:
+        """Write one row and flush.
+
+        Called from the background I/O thread; the check and the write are done
+        under the lock so a concurrent close cannot leave us writing to a file
+        that has just been closed.
+        """
+        with self._lock:
+            if self._csv_writer is None:
+                return False
+            try:
+                self._csv_writer.writerow(row)
+                self._csv_file.flush()
+                return True
+            except Exception as e:
+                logger.warning(f"Failed to write simulation CSV row: {e}")
+                return False
+
+    # ── debug JSON output ───────────────────────────────────────────────
+    def reset_debug_folder(self) -> None:
+        """Empty the debug folder; called once per server start."""
+        if os.path.exists(self.debug_dir):
+            try:
+                shutil.rmtree(self.debug_dir)
+            except Exception as e:
+                logger.warning(f"Unable to clear debug folder: {e}")
+        os.makedirs(self.debug_dir, exist_ok=True)
+
+    def save_debug_files(self, device_id, device_times: dict, edge_times: dict) -> None:
+        """Write the per-device timing JSON files.
+
+        Safe to call from the background I/O thread: the caller passes snapshots,
+        never the live EMA tables.
+        """
+        device_file = f"{self.debug_dir}/{device_id}_device_times.json"
+        edge_file = f"{self.debug_dir}/{device_id}_edge_times.json"
+
+        with open(device_file, "w") as f:
+            json.dump(device_times, f, indent=4)
+
+        with open(edge_file, "w") as f:
+            json.dump(edge_times, f, indent=4)

src/server/communication/request_handler.py

@@ -4,7 +4,6 @@
 import time
 import queue
 import threading
-import shutil
 import math
 from datetime import datetime
 from pathlib import Path
@@ -35,6 +34,7 @@
 from server.communication.message_data import MessageData
 from server.communication.inference_protocol import decode_inference_payload
 from server.communication.device_state import DeviceStateManager
+from server.communication.inference_recorder import InferenceCycleRecorder
 
 from server.models.model_input_converter import ModelInputConverter
 from server.core.delay_simulator import DelaySimulator
@@ -323,10 +323,9 @@ class RequestHandler:
 
     model_registry = {}  # hash -> {model_dir, last_offloading_layer, num_layers}
     device_model_map = {}  # device_id -> model_info
-    # Class-level CSV file tracking for simulation results
-    csv_file = None
-    csv_writer = None
-    inference_counter = 0
+    # Simulation CSV + debug JSON output, thread-safe across the ASGI and
+    # background I/O threads.
+    recorder = InferenceCycleRecorder()
     header_printed = False
     inference_table_rows_printed = 0
 
@@ -417,81 +416,17 @@ def __init__(self):
 
 
     def _cleanup_debug_folder(self):
-        debug_dir = "data/models/debug"
-        if os.path.exists(debug_dir):
-            try:
-                shutil.rmtree(debug_dir)
-            except Exception as e:
-                logger.warning(f"Unable to clear debug folder: {e}")
-        os.makedirs(debug_dir, exist_ok=True)
-
-    # Debug file saving
-    def _save_debug_files(self, device_id):
-        """Legacy sync method – prefer _save_debug_files_data for bg thread."""
-        device_times, edge_times = RequestHandler.device_state.snapshot_debug_times(
-            device_id
-        )
-        self._save_debug_files_data(device_id, device_times, edge_times)
-
-    @staticmethod
-    def _save_debug_files_data(device_id, device_times, edge_times):
-        """Write debug JSON files (safe to call from the background I/O thread)."""
-        debug_dir = "data/models/debug"
-        device_file = f"{debug_dir}/{device_id}_device_times.json"
-        edge_file = f"{debug_dir}/{device_id}_edge_times.json"
-
-        with open(device_file, "w") as f:
-            json.dump(device_times, f, indent=4)
-
-        with open(edge_file, "w") as f:
-            json.dump(edge_times, f, indent=4)
-
-    @staticmethod
-    def _write_csv_row(row):
-        """Write a single CSV row + flush (safe to call from the background I/O thread)."""
-        if RequestHandler.csv_writer:
-            RequestHandler.csv_writer.writerow(row)
-            RequestHandler.csv_file.flush()
+        RequestHandler.recorder.reset_debug_folder()
 
     @classmethod
     def set_simulation_csv(cls, csv_file_path):
         """Set the CSV file for recording simulation results"""
-        import csv
-
-        try:
-            if cls.csv_file:
-                cls.csv_file.close()
-            cls.csv_file = open(csv_file_path, "w", newline="")
-            cls.csv_writer = csv.DictWriter(
-                cls.csv_file,
-                fieldnames=[
-                    "inference_id",
-                    "timestamp",
-                    "avg_device_time",
-                    "min_device_time",
-                    "max_device_time",
-                    "avg_edge_time",
-                    "min_edge_time",
-                    "max_edge_time",
-                    "num_device_layers",
-                    "num_edge_layers",
-                ],
-            )
-            cls.csv_writer.writeheader()
-            cls.csv_file.flush()
-            cls.inference_counter = 0
-            logger.info(f"Simulation CSV recording enabled: {csv_file_path}")
-        except Exception as e:
-            logger.error(f"Failed to setup simulation CSV: {e}")
+        cls.recorder.open_simulation_csv(csv_file_path)
 
     @classmethod
     def close_simulation_csv(cls):
         """Close the CSV file"""
-        if cls.csv_file:
-            cls.csv_file.close()
-            cls.csv_file = None
-            cls.csv_writer = None
-            cls.inference_counter = 0
+        cls.recorder.close_simulation_csv()
 
     def should_force_local_inference(self) -> bool:
         """
@@ -780,26 +715,26 @@ def handle_device_inference_result(self, body, received_timestamp):
         _debug_device_times, _debug_edge_times = device_state.snapshot_debug_times(
             device_id
         )
+        recorder = RequestHandler.recorder
         enqueue_background_io(
             lambda did=device_id, dt=_debug_device_times, et=_debug_edge_times: (
-                self._save_debug_files_data(did, dt, et)
+                recorder.save_debug_files(did, dt, et)
             ),
             description="debug timing JSON write",
         )
 
-        if RequestHandler.csv_writer:
-            RequestHandler.inference_counter += 1
+        if recorder.is_recording:
             device_values = message_data.device_layers_inference_time
             row = {
-                "inference_id": RequestHandler.inference_counter,
+                "inference_id": recorder.next_inference_id(),
                 "timestamp": datetime.now().isoformat(),
                 "avg_device_time": sum(device_values) / len(device_values) if device_values else 0,
                 "avg_edge_time": sum(edge_layer_times) / len(edge_layer_times) if edge_layer_times else 0,
                 "num_device_layers": len(device_values),
                 "num_edge_layers": num_edge_layers,
             }
             enqueue_background_io(
-                lambda r=row: self._write_csv_row(r),
+                lambda r=row: recorder.write_row(r),
                 description="simulation CSV row write",
             )