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, - ), - ]