From 55e28fbc962843e00fc7952d381dfca2acd33bc1 Mon Sep 17 00:00:00 2001 From: ff225 Date: Fri, 17 Jul 2026 15:21:43 +0200 Subject: [PATCH 1/2] fix(server): serialize simulation CSV writes behind a recorder (#10) The simulation CSV handle, its writer and its row counter were class-level attributes on RequestHandler, opened and closed from ASGI request threads while the background I/O thread wrote rows through them. Three races followed: - `close_simulation_csv` cleared `csv_file` while the I/O thread was inside `_write_csv_row`, so a queued row could reach a closed file and raise "I/O operation on closed file"; - `_write_csv_row` tested `csv_writer` and then used `csv_file`, with nothing keeping the two consistent across the gap; - `inference_counter += 1` ran on concurrent request threads, and `set_simulation_csv` reset it underneath them, so ids could collide. InferenceCycleRecorder owns all of it behind one lock: the check and the write happen together, a row that arrives after a close is dropped and reported rather than written, and row ids are handed out atomically. A row that fails to serialize is now logged and swallowed instead of propagating into the background I/O worker. The debug-JSON writing moves here too, since it shares the same background thread and per-server-start lifecycle. `RequestHandler.set_simulation_csv` / `close_simulation_csv` stay as delegating classmethods; `http_server` is unchanged. Second step of #45. --- .../communication/inference_recorder.py | 132 ++++++++++ src/server/communication/request_handler.py | 89 +------ tests/unit/test_inference_recorder.py | 226 ++++++++++++++++++ 3 files changed, 370 insertions(+), 77 deletions(-) create mode 100644 src/server/communication/inference_recorder.py create mode 100644 tests/unit/test_inference_recorder.py diff --git a/src/server/communication/inference_recorder.py b/src/server/communication/inference_recorder.py new file mode 100644 index 0000000..1228fc6 --- /dev/null +++ b/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) diff --git a/src/server/communication/request_handler.py b/src/server/communication/request_handler.py index 7c0c733..de5ec28 100644 --- a/src/server/communication/request_handler.py +++ b/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,18 +715,18 @@ 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, @@ -799,7 +734,7 @@ def handle_device_inference_result(self, body, received_timestamp): "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", ) diff --git a/tests/unit/test_inference_recorder.py b/tests/unit/test_inference_recorder.py new file mode 100644 index 0000000..b2ef7cb --- /dev/null +++ b/tests/unit/test_inference_recorder.py @@ -0,0 +1,226 @@ +"""Unit tests for the simulation-CSV / debug-JSON recorder.""" + +import csv +import json +import threading + +import pytest + +from server.communication.inference_recorder import ( + SIMULATION_CSV_FIELDNAMES, + InferenceCycleRecorder, +) + + +@pytest.fixture +def recorder(tmp_path): + return InferenceCycleRecorder(debug_dir=str(tmp_path / "debug")) + + +def _row(inference_id): + return { + "inference_id": inference_id, + "timestamp": "2026-07-17T10:00:00", + "avg_device_time": 0.5, + "avg_edge_time": 0.25, + "num_device_layers": 2, + "num_edge_layers": 1, + } + + +def test_not_recording_until_a_csv_is_opened(recorder): + assert recorder.is_recording is False + assert recorder.write_row(_row(1)) is False + + +def test_open_writes_header_and_starts_recording(recorder, tmp_path): + path = tmp_path / "sim.csv" + + assert recorder.open_simulation_csv(path) is True + + assert recorder.is_recording is True + header = path.read_text().splitlines()[0] + assert header.split(",") == SIMULATION_CSV_FIELDNAMES + + +def test_write_row_persists_and_flushes(recorder, tmp_path): + path = tmp_path / "sim.csv" + recorder.open_simulation_csv(path) + + assert recorder.write_row(_row(1)) is True + + # Flushed, so readable without closing. + rows = list(csv.DictReader(path.open())) + assert rows[0]["inference_id"] == "1" + assert rows[0]["avg_device_time"] == "0.5" + # Columns the caller does not supply stay empty rather than raising. + assert rows[0]["min_device_time"] == "" + + +def test_next_inference_id_increments_from_one(recorder, tmp_path): + recorder.open_simulation_csv(tmp_path / "sim.csv") + + assert [recorder.next_inference_id() for _ in range(3)] == [1, 2, 3] + + +def test_reopening_resets_the_counter(recorder, tmp_path): + recorder.open_simulation_csv(tmp_path / "a.csv") + recorder.next_inference_id() + recorder.open_simulation_csv(tmp_path / "b.csv") + + assert recorder.next_inference_id() == 1 + + +def test_close_stops_recording_and_resets_counter(recorder, tmp_path): + recorder.open_simulation_csv(tmp_path / "sim.csv") + recorder.next_inference_id() + + recorder.close_simulation_csv() + + assert recorder.is_recording is False + assert recorder.next_inference_id() == 1 + + +def test_close_is_idempotent(recorder, tmp_path): + recorder.open_simulation_csv(tmp_path / "sim.csv") + recorder.close_simulation_csv() + recorder.close_simulation_csv() + + assert recorder.is_recording is False + + +def test_write_after_close_is_dropped_not_raised(recorder, tmp_path): + # Regression: rows still queued on the background I/O thread used to reach a + # closed file and raise "I/O operation on closed file". + path = tmp_path / "sim.csv" + recorder.open_simulation_csv(path) + recorder.close_simulation_csv() + + assert recorder.write_row(_row(1)) is False + + +def test_failed_open_leaves_no_half_open_state(recorder, tmp_path): + assert recorder.open_simulation_csv(tmp_path / "missing-dir" / "sim.csv") is False + + assert recorder.is_recording is False + assert recorder.write_row(_row(1)) is False + + +def test_write_row_survives_a_bad_row(recorder, tmp_path): + recorder.open_simulation_csv(tmp_path / "sim.csv") + + # DictWriter raises on keys outside fieldnames. That must be logged and + # swallowed: a bad telemetry row must not kill the background I/O worker. + assert recorder.write_row({"not_a_column": 1}) is False + # The recorder stays usable afterwards. + assert recorder.write_row(_row(1)) is True + + +def test_concurrent_writes_and_close_never_raise(recorder, tmp_path): + # Regression: `close_simulation_csv` cleared the handle while the background + # thread was inside the write, with no lock between them. + recorder.open_simulation_csv(tmp_path / "sim.csv") + barrier = threading.Barrier(9) + errors = [] + + def writer(): + try: + barrier.wait() + for _ in range(100): + recorder.write_row(_row(recorder.next_inference_id())) + except Exception as exc: + errors.append(exc) + + def closer(): + try: + barrier.wait() + for _ in range(20): + recorder.close_simulation_csv() + recorder.open_simulation_csv(tmp_path / "sim.csv") + except Exception as exc: + errors.append(exc) + + threads = [threading.Thread(target=writer) for _ in range(8)] + threads.append(threading.Thread(target=closer)) + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert errors == [] + + +def test_concurrent_next_inference_id_hands_out_unique_ids(recorder, tmp_path): + recorder.open_simulation_csv(tmp_path / "sim.csv") + ids = [] + lock = threading.Lock() + barrier = threading.Barrier(8) + + def grab(): + barrier.wait() + mine = [recorder.next_inference_id() for _ in range(50)] + with lock: + ids.extend(mine) + + threads = [threading.Thread(target=grab) for _ in range(8)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert sorted(ids) == list(range(1, 401)) + + +def test_close_clears_state_even_if_the_handle_fails_to_close(recorder, tmp_path): + recorder.open_simulation_csv(tmp_path / "sim.csv") + + class ExplodingFile: + def close(self): + raise OSError("disk went away") + + recorder._csv_file = ExplodingFile() + + recorder.close_simulation_csv() + + # A failing close must not strand the recorder in a recording state. + assert recorder.is_recording is False + + +def test_reset_debug_folder_survives_an_unremovable_folder(recorder, monkeypatch): + def boom(path): + raise PermissionError("nope") + + monkeypatch.setattr( + "server.communication.inference_recorder.shutil.rmtree", boom + ) + recorder.reset_debug_folder() + + # Logged and carried on: the folder still exists for the next write. + recorder.reset_debug_folder() + recorder.save_debug_files("dev-1", {}, {}) + + +def test_reset_debug_folder_creates_and_empties(recorder, tmp_path): + debug_dir = tmp_path / "debug" + recorder.reset_debug_folder() + stale = debug_dir / "stale.json" + stale.write_text("{}") + + recorder.reset_debug_folder() + + assert debug_dir.is_dir() + assert not stale.exists() + + +def test_save_debug_files_writes_both_tables(recorder, tmp_path): + recorder.reset_debug_folder() + + recorder.save_debug_files("dev-1", {"layer_0": 1.5}, {"layer_0": 0.25}) + + debug_dir = tmp_path / "debug" + assert json.loads((debug_dir / "dev-1_device_times.json").read_text()) == { + "layer_0": 1.5 + } + assert json.loads((debug_dir / "dev-1_edge_times.json").read_text()) == { + "layer_0": 0.25 + } From fd1642e1b734efe22ee76fa84cbc774943830ee1 Mon Sep 17 00:00:00 2001 From: ff225 Date: Fri, 17 Jul 2026 15:43:15 +0200 Subject: [PATCH 2/2] refactor(server): extract OffloadingService and test the inference hot path (#45) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Third and last step of #45. RequestHandler built the offloading algorithm, ran it, unpicked its candidate list and published the decision telemetry inline. OffloadingService now owns that, and returns an OffloadingDecision carrying everything telemetry needs to explain the choice: the layer, the reason, the candidates, the switch penalty and the estimated cost. Deriving the switch penalty and the cost estimate moves onto the decision object, where they belong — they are properties of the decision, not of the request. The handler is left orchestrating: it asks for a decision and reports it. Building the telemetry *events* stays in the handler on purpose: those need `message_data`, and moving them would couple the service to the transport message shape. The algorithms themselves are untouched: `src/server/offloading_algo/` has no changes. Only the caller moved. Adds tests for handle_device_inference_result, which #45 flagged as the most critical component with no dedicated tests: the EMA updates for device and edge layers, the variance recording, the edge-inference skip conditions, the offloading decision (including forced-local, and that the decision sees the freshly smoothed times), the background I/O scheduling for every output, and that the debug snapshot is immune to later mutation of the live tables. request_handler.py: 1158 -> 933 lines; coverage 48% -> 61%. --- .../communication/offloading_service.py | 184 +++++++++ src/server/communication/request_handler.py | 151 ++----- .../test_handle_device_inference_result.py | 388 ++++++++++++++++++ tests/unit/test_offloading_service.py | 285 +++++++++++++ tests/unit/test_request_handler_helpers.py | 131 ------ 5 files changed, 884 insertions(+), 255 deletions(-) create mode 100644 src/server/communication/offloading_service.py create mode 100644 tests/unit/test_handle_device_inference_result.py create mode 100644 tests/unit/test_offloading_service.py diff --git a/src/server/communication/offloading_service.py b/src/server/communication/offloading_service.py new file mode 100644 index 0000000..50904af --- /dev/null +++ b/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, + ) diff --git a/src/server/communication/request_handler.py b/src/server/communication/request_handler.py index de5ec28..a9ae922 100644 --- a/src/server/communication/request_handler.py +++ b/src/server/communication/request_handler.py @@ -12,19 +12,11 @@ from server.commons import ModelFiles from server.edge.edge_initialization import Edge -from server.offloading_algo.factory import ( - OffloadingContext, - configured_algorithm_class_name, - configured_algorithm_name, - create_offloading_algorithm, -) from server.commons import OffloadingDataFiles from server.commons import EvaluationFiles from server.commons import InputDataFiles from server.telemetry.offloading_decisions import ( - append_offloading_decision, - append_offloading_decision_jsonl, build_offloading_decision_event, ) from server.telemetry.firestore import FirestoreTelemetryPublisher @@ -35,6 +27,7 @@ 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.communication.offloading_service import OffloadingService from server.models.model_input_converter import ModelInputConverter from server.core.delay_simulator import DelaySimulator @@ -391,6 +384,11 @@ def __init__(self): load_evaluation_firestore_config(), default_run_document=EvaluationFiles.server_run_id(), ) + self.offloading_service = OffloadingService( + self.offloading_config, + RequestHandler.device_state, + firestore_publisher=self.firestore_publisher, + ) # Initialize network speed tracking self.last_avg_speed = 0 @@ -485,58 +483,6 @@ def _print_inference_table_row( ) cls.inference_table_rows_printed += 1 - def _create_offloading_algorithm( - self, - *, - device_id: str, - avg_speed: float, - device_inference_times: list, - edge_inference_times: list, - layers_sizes: list, - device_cpu_percent: float, - ): - state = RequestHandler.device_state - return create_offloading_algorithm( - self.offloading_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 _append_and_publish_offloading_decision(self, event, split_config): - """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"], - ) - 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, - ) - def _append_and_publish_inference_cycle(self, event, split_config): """Persist inference-cycle telemetry locally and optionally to Firestore.""" append_evaluation( @@ -760,40 +706,19 @@ def handle_device_inference_result(self, body, received_timestamp): if getattr(message_data, 'avg_speed', 0) > 0: self.last_avg_speed = message_data.avg_speed - offloading_strategy = configured_algorithm_name(self.offloading_config) - offloading_algorithm_class = configured_algorithm_class_name( - self.offloading_config - ) - - offloading_algo = None if self.should_force_local_inference(): - best_offloading_layer = -1 - selection_reason = "forced_local_inference" - decision_candidates = [] + decision = self.offloading_service.forced_local_decision() else: - offloading_algo = self._create_offloading_algorithm( + decision = self.offloading_service.decide( device_id=device_id, avg_speed=self.last_avg_speed, - device_inference_times=list(device_inference_times), - edge_inference_times=list(edge_inference_times), - layers_sizes=list(layers_sizes), + device_inference_times=device_inference_times, + edge_inference_times=edge_inference_times, + layers_sizes=layers_sizes, device_cpu_percent=float(message_data.device_cpu_percent or 0.0), ) - offloading_strategy = offloading_algo.strategy - offloading_algorithm_class = offloading_algo.__class__.__name__ - - # Tentativo di calcolo del livello di offloading ottimale - try: - # Se il modello è conosciuto funzionerà. - best_offloading_layer = offloading_algo.select_offloading_layer() - selection_reason = getattr( - offloading_algo, - "selection_reason", - "lowest_estimated_cost", - ) - decision_candidates = offloading_algo.candidate_evaluations - - # Stampiamo la tabella SOLO se il calcolo è andato a buon fine! + # La tabella si stampa SOLO se il calcolo è andato a buon fine. + if decision.was_computed: RequestHandler._print_inference_table_row( device_id=device_id, offloading_layer_index=message_data.offloading_layer_index, @@ -803,14 +728,8 @@ def handle_device_inference_result(self, body, received_timestamp): network_time_ms=network_time, total_time_ms=total_time, ) - - except IndexError: - # Se mancano i file restituiamo il layer massimo usando la variabile corretta. - best_offloading_layer = -1 - selection_reason = "fallback_missing_model_metadata" - decision_candidates = ( - offloading_algo.candidate_evaluations if offloading_algo else [] - ) + + best_offloading_layer = decision.layer server_start_timestamp = EvaluationFiles.server_start_timestamp() _eval_split = self.evaluation_split_config @@ -818,18 +737,6 @@ def handle_device_inference_result(self, body, received_timestamp): if _eval_outputs["offloading_decisions"]: model_dir = device_state.model_dir(device_id, "unknown") - selected_candidate = next( - ( - candidate - for candidate in decision_candidates - if candidate.get("offloading_layer_index") == best_offloading_layer - and candidate.get("considered_for_selection", True) - ), - None, - ) - switch_penalty = float( - (selected_candidate or {}).get("switch_penalty", 0.0) or 0.0 - ) layer_count = len(layers_sizes) recent_device_layer_times = _sparse_layer_timings( message_data.device_layers_inference_time, @@ -850,21 +757,21 @@ def handle_device_inference_result(self, body, received_timestamp): device_id=device_id, model_dir=model_dir, request_id=str(message_data.message_id), - selected_layer=best_offloading_layer, - selection_reason=selection_reason, + selected_layer=decision.layer, + selection_reason=decision.reason, avg_speed_bytes_per_second=float(self.last_avg_speed), device_cpu_percent=float(message_data.device_cpu_percent or 0.0), edge_cpu_percent=get_edge_cpu_percent(), network_latency_ms=float(network_time), - switch_penalty=switch_penalty, - candidates=decision_candidates, + switch_penalty=decision.switch_penalty, + candidates=decision.candidates, layer_sizes_bytes=[float(size) for size in layers_sizes], device_compute_cost_by_layer=recent_device_layer_times, edge_compute_cost_by_layer=recent_edge_layer_times, system_metrics=get_edge_system_metrics(), - strategy=offloading_strategy, + strategy=decision.strategy, server_start_timestamp=server_start_timestamp, - offloading_algorithm_class=offloading_algorithm_class, + offloading_algorithm_class=decision.algorithm_class, observed={ "acquisition_time_ms": float(acq_time), "device_compute_time_ms": float(device_comp_time), @@ -875,7 +782,7 @@ def handle_device_inference_result(self, body, received_timestamp): ) enqueue_background_io( lambda event=decision_event, split=_eval_split: ( - self._append_and_publish_offloading_decision(event, split) + self.offloading_service.append_and_publish_decision(event, split) ), description="offloading decision CSV write", ) @@ -892,11 +799,7 @@ def handle_device_inference_result(self, body, received_timestamp): device_ema_list, edge_ema_list = device_state.snapshot_times(device_id) ntp_latency = getattr(message_data, 'latency', 0) or 0 avg_payload_size = getattr(message_data, 'payload_size', 0) or 0 - estimated_cost = ( - offloading_algo.lowest_evaluation * 1000 - if offloading_algo is not None and hasattr(offloading_algo, 'lowest_evaluation') - else 0.0 - ) + estimated_cost = decision.estimated_total_cost_ms _layer_sizes = layers_sizes inference_cycle = build_inference_cycle_event( @@ -923,13 +826,13 @@ def handle_device_inference_result(self, body, received_timestamp): edge_ema_ms=[float(t) for t in edge_ema_list], num_device_layers=len(device_values), num_edge_layers=num_edge_layers, - next_offloading_layer=best_offloading_layer, - selection_reason=selection_reason, - num_candidates=len(decision_candidates), + next_offloading_layer=decision.layer, + selection_reason=decision.reason, + num_candidates=len(decision.candidates), estimated_total_cost_ms=estimated_cost, layer_sizes_bytes=[float(s) for s in _layer_sizes], server_start_timestamp=server_start_timestamp, - offloading_algorithm_class=offloading_algorithm_class, + offloading_algorithm_class=decision.algorithm_class, ) enqueue_background_io( lambda event=inference_cycle, split=_eval_split: ( diff --git a/tests/unit/test_handle_device_inference_result.py b/tests/unit/test_handle_device_inference_result.py new file mode 100644 index 0000000..a1f346c --- /dev/null +++ b/tests/unit/test_handle_device_inference_result.py @@ -0,0 +1,388 @@ +"""Unit tests for RequestHandler.handle_device_inference_result. + +This is the hot path — EMA updates, edge inference, the offloading decision and +the background I/O scheduling all meet here — and it had no dedicated tests +(#45). The handler is built with `object.__new__` so none of the heavy __init__ +work (profiler, config loading, Firestore) runs. +""" + +import json +from types import SimpleNamespace + +import pytest + +import server.communication.request_handler as rh +from server.communication.device_state import DeviceStateManager +from server.communication.inference_recorder import InferenceCycleRecorder +from server.communication.offloading_service import OffloadingDecision + +RECEIVED_AT = 1000.5 + + +class NullProfiler: + def start_phase(self, *args, **kwargs): + pass + + def end_phase(self, *args, **kwargs): + pass + + def export_json(self, *args, **kwargs): + pass + + def start_cprofile(self): + pass + + def stop_cprofile(self, *args, **kwargs): + pass + + +class FakeOffloadingService: + """Records how the handler asks for a decision, and what it hands back.""" + + def __init__(self, decision): + self.decision = decision + self.decide_calls = [] + self.forced_local_calls = 0 + self.published = [] + + def decide(self, **kwargs): + self.decide_calls.append(kwargs) + return self.decision + + def forced_local_decision(self): + self.forced_local_calls += 1 + return OffloadingDecision( + layer=-1, + reason="forced_local_inference", + strategy="adaptive_risk", + algorithm_class="AdaptiveRisk", + ) + + def append_and_publish_decision(self, event, split_config): + self.published.append(event) + + +def _message(**overrides): + message = SimpleNamespace( + device_id="dev-1", + timestamp=RECEIVED_AT - 0.1, # 100 ms on the wire + message_id="REQ1", + message_content={"acquisition_time": 0.02}, + device_layers_inference_time=[0.1, 0.2], + offloading_layer_index=1, + layer_output="device-output", + device_cpu_percent=42.0, + avg_speed=5000.0, + latency=0.1, + payload_size=2048, + ) + for key, value in overrides.items(): + setattr(message, key, value) + return message + + +@pytest.fixture +def enqueued(monkeypatch): + """Capture background I/O instead of running it on the worker thread.""" + tasks = [] + monkeypatch.setattr( + rh, + "enqueue_background_io", + lambda task, description="": tasks.append((description, task)) or True, + ) + return tasks + + +@pytest.fixture +def handler(monkeypatch, tmp_path, enqueued): + device_state = DeviceStateManager() + recorder = InferenceCycleRecorder(debug_dir=str(tmp_path / "debug")) + recorder.reset_debug_folder() + monkeypatch.setattr(rh.RequestHandler, "device_state", device_state) + monkeypatch.setattr(rh.RequestHandler, "recorder", recorder) + monkeypatch.setattr(rh.RequestHandler, "num_layers", 3) + + handler = object.__new__(rh.RequestHandler) + handler.profiler = NullProfiler() + handler.network_delay = SimpleNamespace(enabled=False, apply_delay=lambda: 0) + handler.last_avg_speed = 0 + handler.request_count = 0 + handler.debug_mode = False + handler.local_inference_enabled = False + handler.local_inference_probability = 0.0 + handler.evaluation_split_config = {"max_rows": 0, "max_interval_seconds": 0} + handler.evaluation_outputs_config = { + "inference_cycles": False, + "offloading_decisions": False, + } + handler.offloading_service = FakeOffloadingService( + OffloadingDecision( + layer=2, + reason="lowest_estimated_cost", + strategy="adaptive_risk", + algorithm_class="AdaptiveRisk", + candidates=[{"offloading_layer_index": 2, "switch_penalty": 1.5}], + estimated_total_cost_ms=250.0, + ) + ) + + message = _message() + monkeypatch.setattr(rh.RequestHandler, "_from_raw", lambda topic, body: message) + monkeypatch.setattr( + rh.RequestHandler, "_extend_message_data", lambda md, ts, payload: md + ) + monkeypatch.setattr(rh, "load_offloading_ema_alpha_config", lambda: 0.5) + monkeypatch.setattr(rh, "_get_settings", lambda: {"model": {}}) + monkeypatch.setattr( + rh.EvaluationFiles, "server_start_timestamp", staticmethod(lambda: "20260717") + ) + monkeypatch.setattr( + rh.Edge, + "run_inference", + staticmethod(lambda **kwargs: ("edge-prediction", [0.3])), + ) + # `layers_sizes` would otherwise be read from the model's sizes file. + monkeypatch.setattr(DeviceStateManager, "layer_sizes", lambda self, md: [10, 20, 30]) + + handler.message = message + return handler + + +def _call(handler, body=b"payload"): + return handler.handle_device_inference_result(body, RECEIVED_AT) + + +# ── device state ──────────────────────────────────────────────────────── +def test_registers_an_unseen_device_with_the_model_layer_count(handler): + _call(handler) + + profile = rh.RequestHandler.device_state.profiles["dev-1"] + assert profile["model_dir"] == "test_model_96x96" + assert len(profile["edge_inference_times"]) == 3 + + +def test_applies_the_ema_to_device_layer_times(handler): + _call(handler) + + times = rh.RequestHandler.device_state.profiles["dev-1"]["device_inference_times"] + # Seeded at 1, alpha 0.5: 0.5*0.1 + 0.5*1 = 0.55 ; 0.5*0.2 + 0.5*1 = 0.6 + assert times["layer_0"] == pytest.approx(0.55) + assert times["layer_1"] == pytest.approx(0.6) + assert times["layer_2"] == 1 # untouched: the device reported two layers + + +def test_applies_the_ema_to_edge_times_after_the_offloading_layer(handler): + _call(handler) + + times = rh.RequestHandler.device_state.profiles["dev-1"]["edge_inference_times"] + # Edge ran layer 2 (offloading_layer_index=1 → start_layer=2), measured 0.3: + # 0.5*0.3 + 0.5*0.1 = 0.2 + assert times["layer_2"] == pytest.approx(0.2) + assert times["layer_0"] == pytest.approx(0.1) + + +def test_records_measurements_in_the_device_variance_detector(handler): + _call(handler) + + stats = rh.RequestHandler.device_state.variance_stats("dev-1") + assert stats["device"][0]["measurements"] == 1 + assert stats["edge"][2]["measurements"] == 1 + + +# ── edge inference ────────────────────────────────────────────────────── +def test_runs_edge_inference_and_returns_its_prediction(handler): + layer, device_id, prediction = _call(handler) + + assert prediction == "edge-prediction" + assert device_id == "dev-1" + assert layer == 2 + + +def test_skips_edge_inference_when_the_device_ran_everything(handler, monkeypatch): + handler.message.offloading_layer_index = -1 + monkeypatch.setattr( + rh.Edge, + "run_inference", + staticmethod(lambda **kwargs: pytest.fail("edge must not run")), + ) + + _, _, prediction = _call(handler) + + assert prediction == "device-output" + + +def test_skips_edge_inference_past_the_last_offloading_layer(handler, monkeypatch): + rh.RequestHandler.device_state.register("dev-1", num_layers=3) + rh.RequestHandler.device_state.profiles["dev-1"]["ultimo_layer"] = 1 + handler.message.offloading_layer_index = 1 + monkeypatch.setattr( + rh.Edge, + "run_inference", + staticmethod(lambda **kwargs: pytest.fail("edge must not run")), + ) + + _, _, prediction = _call(handler) + + assert prediction == "device-output" + + +# ── offloading decision ───────────────────────────────────────────────── +def test_returns_the_layer_chosen_by_the_service(handler): + layer, _, _ = _call(handler) + + assert layer == 2 + assert handler.offloading_service.forced_local_calls == 0 + + +def test_passes_the_reported_network_speed_to_the_decision(handler): + _call(handler) + + call = handler.offloading_service.decide_calls[0] + assert call["device_id"] == "dev-1" + assert call["avg_speed"] == 5000.0 + assert call["device_cpu_percent"] == 42.0 + assert call["layers_sizes"] == [10, 20, 30] + + +def test_keeps_the_previous_speed_when_the_packet_reports_none(handler): + handler.message.avg_speed = 0 + handler.last_avg_speed = 777.0 + + _call(handler) + + assert handler.offloading_service.decide_calls[0]["avg_speed"] == 777.0 + + +def test_forced_local_inference_bypasses_the_algorithm(handler, monkeypatch): + monkeypatch.setattr(rh.RequestHandler, "should_force_local_inference", lambda self: True) + + layer, _, _ = _call(handler) + + assert layer == -1 + assert handler.offloading_service.forced_local_calls == 1 + assert handler.offloading_service.decide_calls == [] + + +def test_decision_sees_the_freshly_smoothed_times(handler): + _call(handler) + + call = handler.offloading_service.decide_calls[0] + # The EMA update happens before the decision, so it must see 0.55, not 1. + assert call["device_inference_times"][0] == pytest.approx(0.55) + + +# ── background I/O ────────────────────────────────────────────────────── +def test_schedules_the_debug_json_write(handler, enqueued, tmp_path): + _call(handler) + + descriptions = [description for description, _ in enqueued] + assert "debug timing JSON write" in descriptions + + # The task must be runnable off-thread with the snapshot it captured. + task = next(task for desc, task in enqueued if desc == "debug timing JSON write") + task() + assert (tmp_path / "debug" / "dev-1_device_times.json").exists() + + +def test_debug_snapshot_is_taken_before_later_mutations(handler, enqueued, tmp_path): + _call(handler) + task = next(task for desc, task in enqueued if desc == "debug timing JSON write") + + # Mutate the live table after scheduling; the queued task must not see it. + # This is the race the snapshot exists to prevent. + rh.RequestHandler.device_state.update_device_times("dev-1", [99.0], alpha=1.0) + task() + + written = json.loads((tmp_path / "debug" / "dev-1_device_times.json").read_text()) + assert written["layer_0"] == pytest.approx(0.55) + + +def test_does_not_schedule_a_csv_row_when_not_recording(handler, enqueued): + _call(handler) + + descriptions = [description for description, _ in enqueued] + assert "simulation CSV row write" not in descriptions + + +def test_schedules_a_csv_row_while_recording(handler, enqueued, tmp_path): + rh.RequestHandler.recorder.open_simulation_csv(tmp_path / "sim.csv") + + _call(handler) + + task = next(task for desc, task in enqueued if desc == "simulation CSV row write") + assert task() is True + rh.RequestHandler.recorder.close_simulation_csv() + row = (tmp_path / "sim.csv").read_text().splitlines()[1].split(",") + assert row[0] == "1" # inference_id + assert row[8] == "2" # num_device_layers + assert row[9] == "1" # num_edge_layers + + +def test_schedules_the_offloading_decision_write_when_enabled(handler, enqueued): + handler.evaluation_outputs_config["offloading_decisions"] = True + + _call(handler) + + descriptions = [description for description, _ in enqueued] + assert "offloading decision CSV write" in descriptions + + +def test_offloading_decision_event_carries_the_decision(handler, enqueued): + handler.evaluation_outputs_config["offloading_decisions"] = True + + _call(handler) + + task = next( + task for desc, task in enqueued if desc == "offloading decision CSV write" + ) + task() + event = handler.offloading_service.published[0] + assert event["offloading_layer_index"] == 2 + assert event["selection_reason"] == "lowest_estimated_cost" + # The builder reports the penalty in milliseconds. + assert event["switch_penalty"] == pytest.approx(1500.0) + + +def test_schedules_the_inference_cycle_write_when_enabled(handler, enqueued, monkeypatch): + handler.evaluation_outputs_config["inference_cycles"] = True + published = [] + handler.firestore_publisher = SimpleNamespace( + publish_event=lambda kind, event: published.append(event) + ) + monkeypatch.setattr(rh, "append_evaluation", lambda *a, **kw: None) + monkeypatch.setattr( + rh.EvaluationFiles, + "structured_evaluations_file_path", + staticmethod(lambda: "cycles.csv"), + ) + + _call(handler) + + task = next( + task for desc, task in enqueued if desc == "structured evaluation CSV write" + ) + task() + decision = published[0]["offloading_decision"] + assert decision["next_offloading_layer"] == 2 + assert decision["selection_reason"] == "lowest_estimated_cost" + assert decision["estimated_total_cost_ms"] == pytest.approx(250.0) + + +def test_telemetry_is_skipped_entirely_when_both_outputs_are_disabled( + handler, enqueued +): + _call(handler) + + descriptions = [description for description, _ in enqueued] + assert descriptions == ["debug timing JSON write"] + + +# ── profiler export cadence ───────────────────────────────────────────── +def test_exports_profiler_stats_every_50_requests(handler, monkeypatch): + exports = [] + handler.profiler.export_json = lambda name: exports.append(name) + + for _ in range(50): + _call(handler) + + assert exports == ["server_stats.json"] diff --git a/tests/unit/test_offloading_service.py b/tests/unit/test_offloading_service.py new file mode 100644 index 0000000..48e397f --- /dev/null +++ b/tests/unit/test_offloading_service.py @@ -0,0 +1,285 @@ +"""Unit tests for the offloading decision service.""" + +from pathlib import Path + +import pytest + +import server.communication.offloading_service as svc +from server.communication.device_state import DeviceStateManager +from server.communication.offloading_service import ( + OffloadingDecision, + OffloadingService, +) + + +class FakeAlgorithm: + def __init__(self, layer=3, candidates=None, raises=None): + self._layer = layer + self._raises = raises + self.candidate_evaluations = candidates if candidates is not None else [] + # Set on the instance so individual tests can `del` them to model an + # algorithm that does not expose the optional attributes. + self.strategy = "adaptive_risk" + self.selection_reason = "lowest_estimated_cost" + self.lowest_evaluation = 0.25 + + def select_offloading_layer(self): + if self._raises: + raise self._raises + return self._layer + + +@pytest.fixture +def device_state(): + state = DeviceStateManager() + state.register("dev-1", model_dir="FOMO_96_CUT", num_layers=2) + return state + + +@pytest.fixture +def service(device_state): + return OffloadingService({"algorithm": "adaptive_risk"}, device_state) + + +def _decide(service, **overrides): + kwargs = { + "device_id": "dev-1", + "avg_speed": 123.0, + "device_inference_times": [1.0], + "edge_inference_times": [0.5], + "layers_sizes": [10.0], + "device_cpu_percent": 37.5, + } + kwargs.update(overrides) + return service.decide(**kwargs) + + +def test_decide_builds_context_from_device_state(service, device_state, monkeypatch): + calls = [] + + def fake_create(config, context): + calls.append((config, context)) + return FakeAlgorithm() + + monkeypatch.setattr(svc, "create_offloading_algorithm", fake_create) + + decision = _decide(service) + + assert decision.layer == 3 + assert calls[0][0] == {"algorithm": "adaptive_risk"} + context = calls[0][1] + assert context.model_dir == "FOMO_96_CUT" + assert context.avg_speed == 123.0 + assert context.device_cpu_percent == 37.5 + assert context.num_layers == 1 + assert context.adaptive_state is device_state.adaptive_state("dev-1") + + +def test_decide_reports_strategy_and_class_from_the_algorithm(service, monkeypatch): + monkeypatch.setattr( + svc, "create_offloading_algorithm", lambda config, context: FakeAlgorithm() + ) + + decision = _decide(service) + + assert decision.strategy == "adaptive_risk" + assert decision.algorithm_class == "FakeAlgorithm" + assert decision.reason == "lowest_estimated_cost" + assert decision.estimated_total_cost_ms == pytest.approx(250.0) + assert decision.was_computed is True + + +def test_decide_falls_back_to_device_only_on_missing_metadata(service, monkeypatch): + # Regression: a missing layer-sizes file must not pick an arbitrary split. + algorithm = FakeAlgorithm(candidates=[{"offloading_layer_index": 1}]) + algorithm._raises = IndexError("sizes file missing") + monkeypatch.setattr( + svc, "create_offloading_algorithm", lambda config, context: algorithm + ) + + decision = _decide(service) + + assert decision.layer == -1 + assert decision.reason == "fallback_missing_model_metadata" + assert decision.candidates == [{"offloading_layer_index": 1}] + assert decision.estimated_total_cost_ms == 0.0 + assert decision.was_computed is False + + +def test_decide_defaults_the_reason_when_the_algorithm_omits_it(service, monkeypatch): + algorithm = FakeAlgorithm() + del algorithm.selection_reason + monkeypatch.setattr( + svc, "create_offloading_algorithm", lambda config, context: algorithm + ) + + assert _decide(service).reason == "lowest_estimated_cost" + + +def test_decide_tolerates_an_algorithm_without_a_cost_estimate(service, monkeypatch): + algorithm = FakeAlgorithm() + del algorithm.lowest_evaluation + monkeypatch.setattr( + svc, "create_offloading_algorithm", lambda config, context: algorithm + ) + + assert _decide(service).estimated_total_cost_ms == 0.0 + + +def test_forced_local_decision_uses_the_configured_names(service): + decision = service.forced_local_decision() + + assert decision.layer == -1 + assert decision.reason == "forced_local_inference" + assert decision.candidates == [] + assert decision.was_computed is False + # Falls back to the configured names, since no algorithm was built. + assert decision.strategy == "adaptive_risk" + + +def test_switch_penalty_comes_from_the_selected_candidate(): + decision = OffloadingDecision( + layer=2, + reason="lowest_estimated_cost", + strategy="adaptive_risk", + algorithm_class="AdaptiveRisk", + candidates=[ + {"offloading_layer_index": 1, "switch_penalty": 9.0}, + {"offloading_layer_index": 2, "switch_penalty": 4.5}, + ], + ) + + assert decision.selected_candidate["offloading_layer_index"] == 2 + assert decision.switch_penalty == pytest.approx(4.5) + + +def test_switch_penalty_ignores_candidates_excluded_from_selection(): + decision = OffloadingDecision( + layer=2, + reason="lowest_estimated_cost", + strategy="adaptive_risk", + algorithm_class="AdaptiveRisk", + candidates=[ + { + "offloading_layer_index": 2, + "switch_penalty": 4.5, + "considered_for_selection": False, + } + ], + ) + + assert decision.selected_candidate is None + assert decision.switch_penalty == 0.0 + + +@pytest.mark.parametrize("penalty", [None, "", 0]) +def test_switch_penalty_defaults_to_zero_for_missing_values(penalty): + decision = OffloadingDecision( + layer=2, + reason="lowest_estimated_cost", + strategy="adaptive_risk", + algorithm_class="AdaptiveRisk", + candidates=[{"offloading_layer_index": 2, "switch_penalty": penalty}], + ) + + assert decision.switch_penalty == 0.0 + + +def test_switch_penalty_is_zero_when_no_candidate_matches(): + decision = OffloadingDecision( + layer=7, + reason="lowest_estimated_cost", + strategy="adaptive_risk", + algorithm_class="AdaptiveRisk", + candidates=[{"offloading_layer_index": 2, "switch_penalty": 4.5}], + ) + + assert decision.switch_penalty == 0.0 + + +def test_append_and_publish_decision_writes_locally_then_publishes( + device_state, monkeypatch +): + calls = [] + timestamped_jsonl = Path("/tmp/results/offloading_decisions_20260630_105214.jsonl") + + class FakePublisher: + def publish_file( + self, + event_kind, + path, + *, + metadata=None, + document_id=None, + delete_after_publish=True, + ): + calls.append( + ("publish_file", event_kind, path, document_id, delete_after_publish) + ) + return True + + monkeypatch.setattr( + svc, + "append_offloading_decision", + lambda path, event, **kwargs: calls.append( + ("csv", path, event["request_id"], kwargs) + ), + ) + monkeypatch.setattr( + svc, + "append_offloading_decision_jsonl", + lambda path, event, **kwargs: ( + calls.append(("jsonl", path, event["request_id"], kwargs)) + or timestamped_jsonl + ), + ) + monkeypatch.setattr( + svc.EvaluationFiles, "offloading_decisions_file_path", "decisions.csv" + ) + monkeypatch.setattr( + svc.EvaluationFiles, "offloading_decisions_jsonl_base_path", "decisions.jsonl" + ) + service = OffloadingService({}, device_state, firestore_publisher=FakePublisher()) + + service.append_and_publish_decision( + {"request_id": "REQ1", "timestamp": "2026-06-30T10:52:14"}, + {"max_rows": 10, "max_interval_seconds": 60}, + ) + + assert calls == [ + ("csv", "decisions.csv", "REQ1", {"max_rows": 10, "max_interval_seconds": 60}), + ( + "jsonl", + "decisions.jsonl", + "REQ1", + {"max_rows": 10, "max_interval_seconds": 60}, + ), + ( + "publish_file", + "offloading_decisions", + timestamped_jsonl, + timestamped_jsonl.name, + True, + ), + ] + + +def test_append_and_publish_decision_still_writes_locally_without_a_publisher( + device_state, monkeypatch +): + calls = [] + monkeypatch.setattr( + svc, "append_offloading_decision", lambda path, event, **kw: calls.append("csv") + ) + monkeypatch.setattr( + svc, + "append_offloading_decision_jsonl", + lambda path, event, **kw: calls.append("jsonl") or Path("x.jsonl"), + ) + service = OffloadingService({}, device_state, firestore_publisher=None) + + service.append_and_publish_decision( + {"request_id": "REQ1"}, {"max_rows": 0, "max_interval_seconds": 0} + ) + + assert calls == ["csv", "jsonl"] diff --git a/tests/unit/test_request_handler_helpers.py b/tests/unit/test_request_handler_helpers.py index 87e97e0..86c5047 100644 --- a/tests/unit/test_request_handler_helpers.py +++ b/tests/unit/test_request_handler_helpers.py @@ -1,12 +1,9 @@ """Unit tests for pure helpers in request_handler.""" -from pathlib import Path - import numpy as np import pytest import server.communication.request_handler as rh -from server.communication.device_state import DeviceStateManager from server.communication.request_handler import ( _sparse_layer_timings, _to_float_list, @@ -226,43 +223,6 @@ def test_offloading_config_is_loaded(monkeypatch): } -def test_request_handler_creates_algorithm_from_configured_factory(monkeypatch): - calls = [] - - def fake_create(config, context): - calls.append((config, context)) - return "algorithm" - - monkeypatch.setattr(rh, "create_offloading_algorithm", fake_create) - - handler = object.__new__(rh.RequestHandler) - handler.offloading_config = {"algorithm": "adaptive_risk"} - device_state = DeviceStateManager() - monkeypatch.setattr(rh.RequestHandler, "device_state", device_state) - monkeypatch.setitem( - device_state.profiles, - "factory-device", - {"model_dir": "FOMO_96_CUT"}, - ) - - algorithm = handler._create_offloading_algorithm( - device_id="factory-device", - avg_speed=123.0, - device_inference_times=[1.0], - edge_inference_times=[0.5], - layers_sizes=[10.0], - device_cpu_percent=37.5, - ) - - assert algorithm == "algorithm" - assert calls[0][0] == {"algorithm": "adaptive_risk"} - context = calls[0][1] - assert context.model_dir == "FOMO_96_CUT" - assert context.avg_speed == 123.0 - assert context.device_cpu_percent == 37.5 - assert context.adaptive_state is device_state.adaptive_state("factory-device") - - def test_append_and_publish_inference_cycle_keeps_local_write(monkeypatch): calls = [] @@ -300,94 +260,3 @@ def publish_event(self, event_kind, event): ), ("publish", "inference_cycles", "REQ1"), ] - - -def test_append_and_publish_offloading_decision_keeps_local_write(monkeypatch): - calls = [] - timestamped_jsonl = Path( - "/tmp/results/offloading_decisions_20260630_105214.jsonl" - ) - - class FakePublisher: - def publish_file( - self, - event_kind, - path, - *, - metadata=None, - document_id=None, - delete_after_publish=True, - ): - calls.append( - ( - "publish_file", - event_kind, - path, - metadata, - document_id, - delete_after_publish, - ) - ) - return True - - handler = object.__new__(rh.RequestHandler) - handler.firestore_publisher = FakePublisher() - monkeypatch.setattr( - rh, - "append_offloading_decision", - lambda path, event, **kwargs: calls.append( - ("csv", path, event["request_id"], kwargs) - ), - ) - monkeypatch.setattr( - rh, - "append_offloading_decision_jsonl", - lambda path, event, **kwargs: ( - calls.append(("jsonl", path, event["request_id"], kwargs)) - or timestamped_jsonl - ), - ) - monkeypatch.setattr( - rh.EvaluationFiles, - "offloading_decisions_file_path", - "decisions.csv", - ) - monkeypatch.setattr( - rh.EvaluationFiles, - "offloading_decisions_jsonl_base_path", - "offloading_decisions.jsonl", - ) - - handler._append_and_publish_offloading_decision( - {"request_id": "REQ2", "timestamp": "2026-06-30T10:52:14+00:00"}, - {"max_rows": 20, "max_interval_seconds": 120}, - ) - - assert calls == [ - ( - "csv", - "decisions.csv", - "REQ2", - {"max_rows": 20, "max_interval_seconds": 120}, - ), - ( - "jsonl", - "offloading_decisions.jsonl", - "REQ2", - {"max_rows": 20, "max_interval_seconds": 120}, - ), - ( - "publish_file", - "offloading_decisions", - timestamped_jsonl, - { - "offloading_decisions_jsonl_file": ( - "offloading_decisions_20260630_105214.jsonl" - ), - "offloading_decisions_jsonl_path": str(timestamped_jsonl), - "last_event_timestamp": "2026-06-30T10:52:14+00:00", - }, - "offloading_decisions_20260630_105214.jsonl", - True, - ), - ]