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/offloading_service.py

@@ -0,0 +1,184 @@
+"""The offloading decision: picking the split layer and recording why.
+
+`RequestHandler` used to build the algorithm, run it, unpick its candidate list
+and publish the decision telemetry inline. This module owns that, so the handler
+is left orchestrating rather than deciding.
+"""
+
+from dataclasses import dataclass, field
+
+from server.commons import EvaluationFiles
+from server.offloading_algo.factory import (
+    OffloadingContext,
+    configured_algorithm_class_name,
+    configured_algorithm_name,
+    create_offloading_algorithm,
+)
+from server.telemetry.offloading_decisions import (
+    append_offloading_decision,
+    append_offloading_decision_jsonl,
+)
+
+DEVICE_ONLY_LAYER = -1
+
+REASON_FORCED_LOCAL = "forced_local_inference"
+REASON_MISSING_METADATA = "fallback_missing_model_metadata"
+REASON_DEFAULT = "lowest_estimated_cost"
+
+
+@dataclass
+class OffloadingDecision:
+    """The chosen split point, plus everything telemetry needs to explain it."""
+
+    layer: int
+    reason: str
+    strategy: str
+    algorithm_class: str
+    candidates: list = field(default_factory=list)
+    estimated_total_cost_ms: float = 0.0
+
+    @property
+    def selected_candidate(self) -> dict | None:
+        """The candidate the algorithm settled on, if it is still in the list."""
+        return next(
+            (
+                candidate
+                for candidate in self.candidates
+                if candidate.get("offloading_layer_index") == self.layer
+                and candidate.get("considered_for_selection", True)
+            ),
+            None,
+        )
+
+    @property
+    def switch_penalty(self) -> float:
+        return float((self.selected_candidate or {}).get("switch_penalty", 0.0) or 0.0)
+
+    @property
+    def was_computed(self) -> bool:
+        """True when the algorithm actually ran and produced a layer.
+
+        False for a forced-local decision or a metadata fallback, neither of
+        which reflects a real cost comparison.
+        """
+        return self.reason not in (REASON_FORCED_LOCAL, REASON_MISSING_METADATA)
+
+
+class OffloadingService:
+    """Runs the configured offloading algorithm and publishes its decisions."""
+
+    def __init__(self, config: dict, device_state, firestore_publisher=None):
+        self.config = config or {}
+        self.device_state = device_state
+        self.firestore_publisher = firestore_publisher
+
+    def _create_algorithm(
+        self,
+        *,
+        device_id: str,
+        avg_speed: float,
+        device_inference_times: list,
+        edge_inference_times: list,
+        layers_sizes: list,
+        device_cpu_percent: float,
+    ):
+        state = self.device_state
+        return create_offloading_algorithm(
+            self.config,
+            OffloadingContext(
+                avg_speed=avg_speed,
+                num_layers=len(layers_sizes),
+                layers_sizes=list(layers_sizes),
+                inference_time_device=list(device_inference_times),
+                inference_time_edge=list(edge_inference_times),
+                model_dir=state.model_dir(device_id),
+                device_cpu_percent=device_cpu_percent,
+                adaptive_state=state.adaptive_state(device_id),
+                variance_stats=state.variance_stats(device_id),
+            ),
+        )
+
+    def forced_local_decision(self) -> OffloadingDecision:
+        """The decision used when local-only inference is being forced."""
+        return OffloadingDecision(
+            layer=DEVICE_ONLY_LAYER,
+            reason=REASON_FORCED_LOCAL,
+            strategy=configured_algorithm_name(self.config),
+            algorithm_class=configured_algorithm_class_name(self.config),
+        )
+
+    def decide(
+        self,
+        *,
+        device_id: str,
+        avg_speed: float,
+        device_inference_times: list,
+        edge_inference_times: list,
+        layers_sizes: list,
+        device_cpu_percent: float,
+    ) -> OffloadingDecision:
+        """Pick the next offloading layer for a device."""
+        algorithm = self._create_algorithm(
+            device_id=device_id,
+            avg_speed=avg_speed,
+            device_inference_times=device_inference_times,
+            edge_inference_times=edge_inference_times,
+            layers_sizes=layers_sizes,
+            device_cpu_percent=device_cpu_percent,
+        )
+        strategy = algorithm.strategy
+        algorithm_class = algorithm.__class__.__name__
+
+        try:
+            layer = algorithm.select_offloading_layer()
+        except IndexError:
+            # Model metadata (layer sizes) is missing: fall back to device-only
+            # rather than guessing a split point.
+            return OffloadingDecision(
+                layer=DEVICE_ONLY_LAYER,
+                reason=REASON_MISSING_METADATA,
+                strategy=strategy,
+                algorithm_class=algorithm_class,
+                candidates=algorithm.candidate_evaluations,
+            )
+
+        return OffloadingDecision(
+            layer=layer,
+            reason=getattr(algorithm, "selection_reason", REASON_DEFAULT),
+            strategy=strategy,
+            algorithm_class=algorithm_class,
+            candidates=algorithm.candidate_evaluations,
+            estimated_total_cost_ms=(
+                algorithm.lowest_evaluation * 1000
+                if hasattr(algorithm, "lowest_evaluation")
+                else 0.0
+            ),
+        )
+
+    def append_and_publish_decision(self, event: dict, split_config: dict) -> None:
+        """Persist decision telemetry locally and optionally to Firestore."""
+        append_offloading_decision(
+            EvaluationFiles.offloading_decisions_file_path,
+            event,
+            max_rows=split_config["max_rows"],
+            max_interval_seconds=split_config["max_interval_seconds"],
+        )
+        jsonl_path = append_offloading_decision_jsonl(
+            EvaluationFiles.offloading_decisions_jsonl_base_path,
+            event,
+            max_rows=split_config["max_rows"],
+            max_interval_seconds=split_config["max_interval_seconds"],
+        )
+        if self.firestore_publisher is None:
+            return
+        self.firestore_publisher.publish_file(
+            "offloading_decisions",
+            jsonl_path,
+            metadata={
+                "offloading_decisions_jsonl_file": jsonl_path.name,
+                "offloading_decisions_jsonl_path": str(jsonl_path),
+                "last_event_timestamp": event.get("timestamp"),
+            },
+            document_id=jsonl_path.name,
+            delete_after_publish=True,
+        )