From fd1642e1b734efe22ee76fa84cbc774943830ee1 Mon Sep 17 00:00:00 2001 From: ff225 Date: Fri, 17 Jul 2026 15:43:15 +0200 Subject: [PATCH] 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, - ), - ]