diff --git a/.gitignore b/.gitignore index dc89174..3bc74c4 100644 --- a/.gitignore +++ b/.gitignore @@ -116,3 +116,4 @@ data client-sciot.json .python-version uv.lock +sciot-*.json diff --git a/coverage-fast.xml b/coverage-fast.xml new file mode 100644 index 0000000..6bcbf6a --- /dev/null +++ b/coverage-fast.xml @@ -0,0 +1,5407 @@ + + + + + + /Users/admin/Documents/Ricerca/SCIoT_python_client + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/scripts/analysis/variance_analysis.py b/scripts/analysis/variance_analysis.py index 5cebac7..7462bc5 100644 --- a/scripts/analysis/variance_analysis.py +++ b/scripts/analysis/variance_analysis.py @@ -11,14 +11,16 @@ from server.communication.request_handler import RequestHandler -def analyze_current_variance(): +def analyze_current_variance(device_id: str): """ - Analyze current variance state from the RequestHandler's detector. + Analyze current variance state for a device. Provides insights and recommendations. + + Note: variance data lives in the running server's memory, so this only + reports anything when called in-process (e.g. from the server itself). """ - - detector = RequestHandler.variance_detector - stats = detector.get_all_stats() + + stats = RequestHandler.device_state.variance_stats(device_id) print("\n" + "=" * 80) print("CURRENT SYSTEM VARIANCE ANALYSIS") @@ -116,14 +118,13 @@ def analyze_current_variance(): print("\n" + "=" * 80) -def compare_layer_pairs(): +def compare_layer_pairs(device_id: str): """ Compare device vs edge performance for each layer. Helps identify optimal split points. """ - - detector = RequestHandler.variance_detector - stats = detector.get_all_stats() + + stats = RequestHandler.device_state.variance_stats(device_id) print("\n" + "=" * 80) print("DEVICE vs EDGE COMPARISON") @@ -174,16 +175,16 @@ def compare_layer_pairs(): print("\n" + "=" * 80) -def export_variance_data(output_file: str = "variance_stats.json"): +def export_variance_data(device_id: str, output_file: str = "variance_stats.json"): """ Export current variance data to JSON for external analysis. - + Args: + device_id: The device whose variance statistics to export output_file: Path to save the JSON file """ - - detector = RequestHandler.variance_detector - stats = detector.get_all_stats() + + stats = RequestHandler.device_state.variance_stats(device_id) output_path = Path(output_file) @@ -200,23 +201,30 @@ def export_variance_data(output_file: str = "variance_stats.json"): print(""" This utility analyzes inference time variance detected by the system. +Variance statistics are per-device and live in the running server's memory, +so these helpers only report data when called in-process. + Usage: from variance_analysis import ( analyze_current_variance, # Full analysis report compare_layer_pairs, # Device vs Edge comparison export_variance_data # Export to JSON ) - - analyze_current_variance() - compare_layer_pairs() - export_variance_data("my_variance_data.json") + + analyze_current_variance("my-device-id") + compare_layer_pairs("my-device-id") + export_variance_data("my-device-id", "my_variance_data.json") """) - - # Run demonstrations if variance data exists + + # Run demonstrations for every device the handler knows about. print("\nRunning analysis on current system state...") - try: - analyze_current_variance() - compare_layer_pairs() - except Exception as e: - print(f"\n⚠️ Could not analyze system variance: {e}") + known_devices = list(RequestHandler.device_profiles) + if not known_devices: + print("\n⚠️ No devices registered in this process.") print(" (System needs to be running with active inference for data)") + for device_id in known_devices: + try: + analyze_current_variance(device_id) + compare_layer_pairs(device_id) + except Exception as e: + print(f"\n⚠️ Could not analyze variance for {device_id}: {e}") diff --git a/src/server/communication/device_state.py b/src/server/communication/device_state.py new file mode 100644 index 0000000..ece4a31 --- /dev/null +++ b/src/server/communication/device_state.py @@ -0,0 +1,217 @@ +"""Per-device runtime state for the inference/offloading path. + +``RequestHandler`` used to keep every piece of runtime state in class-level +dictionaries mutated from both the ASGI thread and the background I/O thread. +This module owns that state instead, behind a single reentrant lock, and scopes +the variance detector and the adaptive-risk state to one device each. + +The legacy ``device_profiles`` mapping stays the canonical storage for a +device's EMA tables: ``http_server`` and the dashboard read it as a plain dict, +so the manager hands out the live dict rather than a copy. +""" + +import json +import threading + +from server.commons import OffloadingDataFiles +from server.core.variance_detector import VarianceDetector +from server.offloading_algo.adaptive_risk import AdaptiveRiskState + +DEFAULT_MODEL_DIR = "test_model_96x96" +DEFAULT_MODEL_KEY = "fomo_96x96" + +VARIANCE_WINDOW_SIZE = 10 +VARIANCE_THRESHOLD = 0.15 + + +def build_profile( + device_id: str, + *, + model_dir: str = DEFAULT_MODEL_DIR, + model_key: str = DEFAULT_MODEL_KEY, + num_layers: int = 0, + last_offloading_layer: int | None = None, +) -> dict: + """Build a device profile in the legacy ``device_profiles`` shape.""" + profile = { + "id": device_id, + "model_dir": model_dir, + "model_key": model_key, + "device_inference_times": {f"layer_{i}": 1 for i in range(num_layers)}, + "edge_inference_times": {f"layer_{i}": 0.1 for i in range(num_layers)}, + } + if last_offloading_layer is not None: + profile["ultimo_layer"] = last_offloading_layer + return profile + + +class DeviceStateManager: + """Thread-safe registry of per-device runtime state. + + Holds the profile (EMA tables), the variance detector, and the adaptive-risk + state for each device. Every mutation goes through ``_lock``: the EMA updates + run on the ASGI thread while the background I/O worker snapshots the same + dicts. + """ + + def __init__(self): + self._lock = threading.RLock() + # device_id -> legacy profile dict; exposed live via `profiles`. + self._profiles: dict[str, dict] = {} + self._variance_detectors: dict[str, VarianceDetector] = {} + self._adaptive_states: dict[str, AdaptiveRiskState] = {} + self._layer_sizes_cache: dict[str, list] = {} + + @property + def profiles(self) -> dict: + """The live ``device_id -> profile`` mapping. + + Returned by reference for backwards compatibility with direct readers. + """ + return self._profiles + + # ── device lifecycle ──────────────────────────────────────────────── + def register( + self, + device_id: str, + *, + model_dir: str = DEFAULT_MODEL_DIR, + model_key: str = DEFAULT_MODEL_KEY, + num_layers: int = 0, + last_offloading_layer: int | None = None, + ) -> dict: + """Return the device's profile, creating it on first sight.""" + with self._lock: + profile = self._profiles.get(device_id) + if profile is None: + profile = build_profile( + device_id, + model_dir=model_dir, + model_key=model_key, + num_layers=num_layers, + last_offloading_layer=last_offloading_layer, + ) + self._profiles[device_id] = profile + return profile + + def __contains__(self, device_id: str) -> bool: + with self._lock: + return device_id in self._profiles + + def model_dir(self, device_id: str, default: str = DEFAULT_MODEL_DIR) -> str: + with self._lock: + return self._profiles.get(device_id, {}).get("model_dir", default) + + def model_key(self, device_id: str, default: str = DEFAULT_MODEL_KEY) -> str: + with self._lock: + return self._profiles.get(device_id, {}).get("model_key", default) + + def clear(self) -> None: + with self._lock: + self._profiles.clear() + self._variance_detectors.clear() + self._adaptive_states.clear() + self._layer_sizes_cache.clear() + + # ── per-device collaborators ──────────────────────────────────────── + def variance_detector(self, device_id: str) -> VarianceDetector: + """The device's own detector. + + Scoped per device so heterogeneous hardware reporting the same layer + indices cannot pollute each other's variance statistics. + """ + with self._lock: + detector = self._variance_detectors.get(device_id) + if detector is None: + detector = VarianceDetector( + window_size=VARIANCE_WINDOW_SIZE, + variance_threshold=VARIANCE_THRESHOLD, + ) + self._variance_detectors[device_id] = detector + return detector + + def adaptive_state(self, device_id: str) -> AdaptiveRiskState: + with self._lock: + state = self._adaptive_states.get(device_id) + if state is None: + state = AdaptiveRiskState() + self._adaptive_states[device_id] = state + return state + + def variance_stats(self, device_id: str) -> dict: + """Variance statistics for one device, for the offloading context.""" + return self.variance_detector(device_id).get_all_stats() + + # ── EMA updates ───────────────────────────────────────────────────── + def update_device_times(self, device_id: str, layer_times, *, alpha: float) -> None: + """Apply the EMA update for device-measured layer times.""" + profile = self.register(device_id) + detector = self.variance_detector(device_id) + with self._lock: + times = profile["device_inference_times"] + for layer_id, measured in enumerate(layer_times): + self._apply_ema(times, layer_id, float(measured), alpha) + detector.add_device_measurement(layer_id, measured) + + def update_edge_times( + self, device_id: str, layer_times, *, start_layer: int, alpha: float + ) -> None: + """Apply the EMA update for edge-measured layer times.""" + profile = self.register(device_id) + detector = self.variance_detector(device_id) + with self._lock: + times = profile["edge_inference_times"] + for offset, measured in enumerate(layer_times): + layer_id = start_layer + offset + self._apply_ema(times, layer_id, float(measured), alpha) + detector.add_edge_measurement(layer_id, measured) + + @staticmethod + def _apply_ema(times: dict, layer_id: int, measured: float, alpha: float) -> None: + layer_key = f"layer_{layer_id}" + if layer_key in times: + times[layer_key] = alpha * measured + (1 - alpha) * times[layer_key] + else: + times[layer_key] = measured + + # ── reads for the offloading decision ─────────────────────────────── + def snapshot_times(self, device_id: str) -> tuple[list, list]: + """Copy the EMA tables so callers never iterate a dict being mutated.""" + with self._lock: + profile = self._profiles.get(device_id) + if profile is None: + raise ValueError(f"Unknown device_id '{device_id}'") + return ( + list(profile["device_inference_times"].values()), + list(profile["edge_inference_times"].values()), + ) + + def snapshot_debug_times(self, device_id: str) -> tuple[dict, dict]: + """Copy the EMA tables for the background debug-JSON writer.""" + with self._lock: + profile = self._profiles.get(device_id) + if profile is None: + return {}, {} + return ( + dict(profile["device_inference_times"]), + dict(profile["edge_inference_times"]), + ) + + def layer_sizes(self, model_dir: str) -> list: + """Return per-layer output sizes, reading each model's file once.""" + with self._lock: + cached = self._layer_sizes_cache.get(model_dir) + if cached is not None: + return cached + + # Read outside the lock: this touches the disk, and a lookup for another + # model must not block behind it. + with open(OffloadingDataFiles.get_sizes_path(model_dir), "r") as file: + sizes = list(json.load(file).values()) + with self._lock: + return self._layer_sizes_cache.setdefault(model_dir, sizes) + + def load_stats(self, device_id: str) -> tuple[list, list, list]: + """Return (device EMA times, edge EMA times, layer sizes) for a device.""" + device_times, edge_times = self.snapshot_times(device_id) + return device_times, edge_times, self.layer_sizes(self.model_dir(device_id)) 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/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 aaff010..7485534 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 @@ -13,20 +12,11 @@ from server.commons import ModelFiles from server.edge.edge_initialization import Edge -from server.offloading_algo.adaptive_risk import AdaptiveRiskState -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,10 +25,12 @@ 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.communication.offloading_service import OffloadingService from server.models.model_input_converter import ModelInputConverter from server.core.delay_simulator import DelaySimulator -from server.core.variance_detector import VarianceDetector from sciot.config import load_server_config import struct @@ -324,18 +316,17 @@ class RequestHandler: model_registry = {} # hash -> {model_dir, last_offloading_layer, num_layers} device_model_map = {} # device_id -> model_info - # Class-level variance detector (shared across all requests) - variance_detector = VarianceDetector(window_size=10, variance_threshold=0.15) - # 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 - # Dictionary to store device profile data - device_profiles = {} - offloading_states = {} + # Per-device runtime state (profiles, variance detectors, adaptive state). + device_state = DeviceStateManager() + # Live view of `device_state`'s profiles, kept for direct readers such as + # `http_server` and the dashboard. + device_profiles = device_state.profiles num_layers = 0 def __init__(self): @@ -393,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 @@ -418,83 +414,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.""" - profile = RequestHandler.device_profiles[device_id] - self._save_debug_files_data( - device_id, - dict(profile["device_inference_times"]), - dict(profile["edge_inference_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: """ @@ -553,63 +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, - ): - adaptive_state = RequestHandler.offloading_states.setdefault( - device_id, - AdaptiveRiskState(), - ) - 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=RequestHandler.device_profiles.get(device_id, {}).get( - "model_dir", "test_model_96x96" - ), - device_cpu_percent=device_cpu_percent, - adaptive_state=adaptive_state, - variance_stats=RequestHandler.variance_detector.get_all_stats(), - ), - ) - - 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( @@ -630,18 +503,13 @@ def handle_registration(self, device_id: str, model_hash: str = ""): model_info = RequestHandler.model_registry[model_hash] RequestHandler.device_model_map[device_id] = model_info - if device_id not in RequestHandler.device_profiles: - num_layers = model_info["num_layers"] - RequestHandler.device_profiles[device_id] = { - "id": device_id, - "model_dir": model_info["model_dir"], - "model_key": model_info.get("model_key", "fomo_96x96"), - "ultimo_layer": model_info["last_offloading_layer"], - "device_inference_times": { - f"layer_{i}": 1 for i in range(num_layers) - }, - "edge_inference_times": {f"layer_{i}": 0.1 for i in range(num_layers)}, - } + RequestHandler.device_state.register( + device_id, + model_dir=model_info["model_dir"], + model_key=model_info.get("model_key", "fomo_96x96"), + num_layers=model_info["num_layers"], + last_offloading_layer=model_info["last_offloading_layer"], + ) return device_id def handle_device_input(self, rgb565_image, height, width): @@ -744,37 +612,21 @@ def handle_device_inference_result( # ─── STATO E LOGICA DEVICE ─── self.profiler.start_phase("server_state_update", trace_id) - if device_id not in RequestHandler.device_profiles: - RequestHandler.device_profiles[device_id] = { - "id": device_id, - "model_dir": "test_model_96x96", - "model_key": "fomo_96x96", - "device_inference_times": { - f"layer_{i}": 1 for i in range(RequestHandler.num_layers) - }, - "edge_inference_times": { - f"layer_{i}": 0.1 for i in range(RequestHandler.num_layers) - }, - } - - device_inference_times = RequestHandler.device_profiles[device_id]["device_inference_times"] - edge_inference_times = RequestHandler.device_profiles[device_id]["edge_inference_times"] + device_state = RequestHandler.device_state + profile = device_state.register(device_id, num_layers=RequestHandler.num_layers) alpha = load_offloading_ema_alpha_config() - for l_id, inference_time in enumerate(message_data.device_layers_inference_time): - layer_key = f"layer_{l_id}" - if layer_key in device_inference_times: - device_inference_times[layer_key] = (alpha * inference_time + (1 - alpha) * device_inference_times[layer_key]) - else: - device_inference_times[layer_key] = inference_time - RequestHandler.variance_detector.add_device_measurement(l_id, inference_time) - + device_state.update_device_times( + device_id, + message_data.device_layers_inference_time, + alpha=alpha, + ) + self.profiler.end_phase("server_state_update", trace_id) # ─── INFERENZA SUL SERVER (SE NECESSARIA) ─── edge_layer_times = [] num_edge_layers = 0 - profile = RequestHandler.device_profiles[device_id] ultimo_layer = profile.get("ultimo_layer", 58) if message_data.offloading_layer_index == -1 or message_data.offloading_layer_index >= ultimo_layer: @@ -789,7 +641,6 @@ def handle_device_inference_result( prediction, edge_layer_times = Edge.run_inference( offloading_layer_index=message_data.offloading_layer_index, offloading_layer_output=message_data.layer_output, - edge_inference_times=edge_inference_times, model_dir=model_dir, model_key=model_key, models_config=models_config, @@ -800,31 +651,34 @@ def handle_device_inference_result( # Aggiornamento tempi Edge (Post-Inference) self.profiler.start_phase("server_state_update", f"{trace_id}_post") - start_layer = message_data.offloading_layer_index + 1 - for i, measured_time in enumerate(edge_layer_times): - layer_key = f"layer_{start_layer + i}" - if layer_key in edge_inference_times: - edge_inference_times[layer_key] = (alpha * measured_time + (1 - alpha) * edge_inference_times[layer_key]) - else: - edge_inference_times[layer_key] = measured_time - RequestHandler.variance_detector.add_edge_measurement(start_layer + i, measured_time) + device_state.update_edge_times( + device_id, + edge_layer_times, + start_layer=message_data.offloading_layer_index + 1, + alpha=alpha, + ) self.profiler.end_phase("server_state_update", f"{trace_id}_post") # ─── LOGICA ALGORITMO E I/O BACKGROUND ─── self.profiler.start_phase("server_algo_and_io", trace_id) - # Offload I/O tasks - _debug_profile = RequestHandler.device_profiles[device_id] + # Offload I/O tasks. The snapshot is taken here, under the state lock, + # so the background thread never serializes a dict mid-update. + _debug_device_times, _debug_edge_times = device_state.snapshot_debug_times( + device_id + ) + recorder = RequestHandler.recorder enqueue_background_io( - lambda did=device_id, dt=dict(_debug_profile["device_inference_times"]), et=dict(_debug_profile["edge_inference_times"]): self._save_debug_files_data(did, dt, et), + lambda did=device_id, dt=_debug_device_times, et=_debug_edge_times: ( + 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, @@ -832,7 +686,7 @@ def handle_device_inference_result( "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", ) @@ -849,45 +703,28 @@ def handle_device_inference_result( self.network_delay.apply_delay() # Decisione Offloading per la prossima risposta - device_inference_times, edge_inference_times, layers_sizes = RequestHandler._load_stats(device_id) - # Use the server-side receive-rate estimate for the next offloading decision. - if getattr(message_data, "avg_speed", 0) > 0: + ( + device_inference_times, + edge_inference_times, + layers_sizes, + ) = device_state.load_stats(device_id) + # Usa il pacchetto appena ricevuto per calcolare la velocità reale del Wi-Fi! + 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, @@ -897,35 +734,15 @@ def handle_device_inference_result( 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 _eval_outputs = self.evaluation_outputs_config if _eval_outputs["offloading_decisions"]: - model_dir = RequestHandler.device_profiles.get(device_id, {}).get( - "model_dir", "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 - ) + model_dir = device_state.model_dir(device_id, "unknown") layer_count = len(layers_sizes) recent_device_layer_times = _sparse_layer_timings( message_data.device_layers_inference_time, @@ -946,21 +763,21 @@ def handle_device_inference_result( 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), @@ -971,40 +788,25 @@ def handle_device_inference_result( ) 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", ) # ─── STRUTTURATO INFERENCE CYCLE OUTPUT ⟩⟩⟩ # This is placed after the offloading decision so all data is available. - # Skipping it entirely (build + enqueue) when disabled also avoids the - # `_load_stats` disk/cache access below. if _eval_outputs["inference_cycles"]: - model_key = RequestHandler.device_profiles.get(device_id, {}).get( - "model_key", "fomo_96x96" - ) - _model_dir = RequestHandler.device_profiles.get(device_id, {}).get( - "model_dir", "test_model_96x96" - ) + model_key = device_state.model_key(device_id) + _model_dir = device_state.model_dir(device_id) # `device_layers_inference_time` is a numpy array (see inference_protocol), # so the old `array or []` raised "truth value of an array is ambiguous" # for multi-layer arrays, silently aborting the evaluation write. device_values = _to_float_list(message_data.device_layers_inference_time) - device_ema_list = list(RequestHandler.device_profiles[device_id]["device_inference_times"].values()) if device_id in RequestHandler.device_profiles else [] - edge_ema_list = list(RequestHandler.device_profiles[device_id]["edge_inference_times"].values()) if device_id in RequestHandler.device_profiles else [] + 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 - ) - _layer_sizes = ( - list(RequestHandler._load_stats(device_id)[2]) - if hasattr(RequestHandler, '_load_stats') - else [] - ) + estimated_cost = decision.estimated_total_cost_ms + _layer_sizes = layers_sizes inference_cycle = build_inference_cycle_event( device_id=str(device_id), @@ -1030,13 +832,13 @@ def handle_device_inference_result( 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: ( @@ -1068,35 +870,6 @@ def handle_device_inference_result( def handle_offloading_layer(self, best_offloading_layer): return best_offloading_layer - # Class-level layer_sizes cache: model_dir -> list of sizes - # Layer sizes never change at runtime, so we read them once per model. - _layer_sizes_cache = {} - - # modificata - @staticmethod - def _load_stats(device_id): - - # Recupero il profilo dal vocabolario in RAM - profile = RequestHandler.device_profiles.get(device_id) - if profile is None: - raise ValueError(f"Unknown device_id '{device_id}'") - - # Prendo solo i valori convertendoli in liste - device_inference_times = list(profile["device_inference_times"].values()) - edge_inference_times = list(profile["edge_inference_times"].values()) - - # Layer sizes: read from disk once, then serve from cache - model_dir = RequestHandler.device_profiles.get(device_id, {}).get( - "model_dir", "test_model_96x96" - ) - if model_dir not in RequestHandler._layer_sizes_cache: - with open(OffloadingDataFiles.get_sizes_path(model_dir), "r") as file: - sizes_data = json.load(file) - RequestHandler._layer_sizes_cache[model_dir] = list(sizes_data.values()) - layers_sizes = RequestHandler._layer_sizes_cache[model_dir] - - return device_inference_times, edge_inference_times, layers_sizes - @staticmethod def _from_raw(topic: str, payload: bytes): """Parse the raw message payload into a MessageData instance.""" diff --git a/src/server/edge/edge_initialization.py b/src/server/edge/edge_initialization.py index 85fed06..e7264f0 100644 --- a/src/server/edge/edge_initialization.py +++ b/src/server/edge/edge_initialization.py @@ -10,7 +10,6 @@ from server.commons import ModelFiles from server.models.model_manager import ModelManager from server.models.model_input_converter import ModelInputConverter -from server.core.variance_detector import VarianceDetector from sciot.config import load_server_config from pathlib import Path @@ -27,9 +26,6 @@ def load_delay_config(): class Edge: - # Class-level variance detector (shared across edge inference calls) - variance_detector = VarianceDetector(window_size=10, variance_threshold=0.15) - # ===================================================== # OPTIMIZATION: Persistent ModelManager cache. # ModelManager instances (and their TFLite interpreter caches) are @@ -123,9 +119,7 @@ def get_valid_offloading_points(model): return valid_points @staticmethod - def _get_model_manager( - model_dir: str, models_config: dict, edge_inference_times: dict - ) -> ModelManager: + def _get_model_manager(model_dir: str, models_config: dict) -> ModelManager: """ OPTIMIZATION: Get or create a cached ModelManager for the given model_dir. The ModelManager (and its TFLite interpreter cache) persists across calls, @@ -137,19 +131,13 @@ def _get_model_manager( Edge._delay_config = load_delay_config() model_manager = ModelManager( - inference_times=edge_inference_times, computation_delay_config=Edge._delay_config, - variance_detector=Edge.variance_detector, models_config=models_config, ) model_h5_path = ModelFiles.get_model_h5_path(model_dir) model_manager.load_model(model_h5_path) Edge._model_managers[model_dir] = model_manager logger.info(f"Cached ModelManager for model_dir={model_dir}") - else: - # Reuse cached manager, but update inference times reference - # so that the EMA-smoothed times stay in sync with RequestHandler - Edge._model_managers[model_dir].inference_times = edge_inference_times return Edge._model_managers[model_dir] @@ -157,13 +145,16 @@ def _get_model_manager( def run_inference( offloading_layer_index: int, offloading_layer_output: np.array, - edge_inference_times: dict, model_dir: str, model_key: str = None, models_config: dict = None, ): """ Versione dinamica di run_inference. + + Returns the prediction and the raw per-layer timings; smoothing them into + the EMA tables is the caller's job (see `DeviceStateManager`). + :param model_key: La chiave del modello (es. 'fomo_96x96' o 'fomo_144x144') :param models_config: Il dizionario completo dei modelli dal setting.yaml """ @@ -174,9 +165,7 @@ def run_inference( # OPTIMIZATION: Reuse cached ModelManager instead of creating a new one each call. # This preserves the TFLite interpreter cache and avoids re-loading the H5 model. - model_manager = Edge._get_model_manager( - model_dir, models_config, edge_inference_times - ) + model_manager = Edge._get_model_manager(model_dir, models_config) # Preparazione indici dei layer predictions = {} @@ -262,12 +251,14 @@ def run_inference( break # ESECUZIONE INFERENZA DINAMICA - t0 = time.time() + # perf_counter, not time.time: these timings are the authoritative + # per-layer costs feeding the EMA and the offloading decision. + t0 = time.perf_counter() # Passiamo il model_key per usare il TFLite interpreter corretto (96 o 144) prediction = model_manager.predict_single_layer( layer_index, start_layer_offset, prediction_data, model_key=model_key ) - t1 = time.time() + t1 = time.perf_counter() measured_times.append(t1 - t0) predictions[layer.name] = prediction @@ -304,9 +295,8 @@ def initialization( # QUI LA MODIFICA: Passiamo models_config invece di model_dir model_manager = ModelManager( - models_config=models_config, + models_config=models_config, computation_delay_config=delay_config, - variance_detector=Edge.variance_detector, ) model_manager.load_model(ModelFiles.get_model_h5_path(model_dir)) diff --git a/src/server/models/model_manager.py b/src/server/models/model_manager.py index c2052fc..a668db0 100644 --- a/src/server/models/model_manager.py +++ b/src/server/models/model_manager.py @@ -12,12 +12,10 @@ import numpy as np import tensorflow as tf -from server.commons import OffloadingDataFiles from server.commons import ModelFiles from server.logger.log import logger from server.models.model_manager_config import ModelManagerConfig from server.core.delay_simulator import DelaySimulator -from server.core.variance_detector import VarianceDetector class ModelLoadError(RuntimeError): @@ -190,46 +188,27 @@ def load_keras_model_for_inference(model_path: str): def track_inference_time(func): """ - This decorator is used to track the execution time of a function. + Log the execution time of a single-layer prediction. + + This is instrumentation only. The authoritative per-layer timings are the + ones `Edge.run_inference` measures and returns; `DeviceStateManager` owns + smoothing them into the EMA tables with the configured alpha. + :param func: the function to be decorated :return: the decorated function """ @wraps(func) def wrapper(self, layer_id: int, layer_offset: int, *args, **kwargs) -> object: - # Start the timer start_time = time.perf_counter() - # Execute the original function (predict_single_layer) result = func(self, layer_id, layer_offset, *args, **kwargs) - # Calculate the elapsed time elapsed_time = time.perf_counter() - start_time - layer_key = f"layer_{layer_id - layer_offset}" # Use layer_X format to match device times - layer_number = ( - layer_id - layer_offset - ) # Keep the numeric value for variance tracking - # Use exponential moving average to smooth times (alpha=0.2 gives 80% weight to history) - if layer_key in self.inference_times: - alpha = 0.2 # Weight for new measurement - self.inference_times[layer_key] = ( - alpha * elapsed_time + (1 - alpha) * self.inference_times[layer_key] - ) - else: - self.inference_times[layer_key] = elapsed_time - logger.debug( - f"Edge Inference for layer [{layer_number}] took {elapsed_time:.4f} seconds (smoothed: {self.inference_times[layer_key]:.4f}s)" + f"Edge Inference for layer [{layer_id - layer_offset}] " + f"took {elapsed_time:.4f} seconds" ) - # Track variance for edge inference times - if hasattr(self, "variance_detector") and self.variance_detector: - self.variance_detector.add_edge_measurement(layer_number, elapsed_time) - - # OPTIMIZATION: Removed per-layer save_inference_times() call. - # Writing to disk after every single layer prediction (up to 59 times per inference) - # was a major I/O bottleneck. Inference times are tracked in-memory via the - # inference_times dict which is shared with RequestHandler.device_profiles. - return result return wrapper @@ -247,48 +226,22 @@ class ModelManager: model_path: The path to the model. num_layers: The number of layers in the model. model: The model. - inference_times: A dictionary to store the inference times for each layer. """ - # def __init__(self, model_dir: str = ModelManagerConfig.MODEL_DIR_PATH, save_path: str = ModelManagerConfig.SAVE_PATH, - # model_path: str = ModelManagerConfig.MODEL_PATH, - # inference_times: dict = {}, computation_delay_config: dict = None, - # variance_detector: VarianceDetector = None): - # self.model_dir = model_dir # <-- aggiunto - # self.save_path = save_path - - # self.model_path = model_path - # self.num_layers = None - # self.model = None - # # dictionary to store inference times for each layer - # self.inference_times = inference_times - # # cache for TFLite interpreters to avoid recreation overhead - # self._interpreter_cache = {} - # # delay simulator for computation - # self.computation_delay = DelaySimulator(computation_delay_config) - # if self.computation_delay.enabled: - # logger.info(f"Computation delay simulation enabled: {self.computation_delay.get_delay_info()}") - # # variance detector for tracking inference time stability - # self.variance_detector = variance_detector - def __init__( self, models_config: dict = None, save_path: str = ModelManagerConfig.SAVE_PATH, - inference_times: dict = {}, computation_delay_config: dict = None, - variance_detector: VarianceDetector = None, ): # Salviamo l'intera configurazione dei modelli (fomo_96, fomo_144, ecc.) self.models_config = models_config self.save_path = save_path - self.inference_times = inference_times # La cache ora è "vuota" all'inizio e si riempirà per ogni modello self._interpreter_cache = {} self.computation_delay = DelaySimulator(computation_delay_config) - self.variance_detector = variance_detector logger.info("ModelManager dinamico inizializzato.") def get_model_layer(self, layer_id: int) -> tf.keras.layers.Layer: @@ -475,21 +428,3 @@ def predict_single_layer( # internal data buffer, which would cause a RuntimeError on the next invoke(). return np.copy(interpreter.get_tensor(output_details[0]["index"])) - def save_inference_times(self, save_path: str | None = None): - """Save the inference times to a JSON file. - Args: - save_path: The path to save the inference times. - Returns: - None - """ - if save_path is not None: - self.save_path = save_path - self.save_path = ( - self.save_path[:-1] if self.save_path[-1] == "/" else self.save_path - ) - inference_times = self.inference_times - try: - with open(OffloadingDataFiles.data_file_path_edge, "w") as f: - json.dump(inference_times, f, indent=4) - except Exception as e: - logger.error(f"Failed to save inference times: {e}") diff --git a/tests/unit/test_device_state.py b/tests/unit/test_device_state.py new file mode 100644 index 0000000..d1f0f03 --- /dev/null +++ b/tests/unit/test_device_state.py @@ -0,0 +1,226 @@ +"""Unit tests for the per-device runtime state manager.""" + +import json +import threading + +import pytest + +from server.communication.device_state import ( + DEFAULT_MODEL_DIR, + DEFAULT_MODEL_KEY, + DeviceStateManager, + build_profile, +) + + +@pytest.fixture +def manager(): + return DeviceStateManager() + + +def test_register_builds_legacy_profile_shape(manager): + profile = manager.register( + "dev-1", + model_dir="FOMO_96_CUT", + model_key="fomo_96x96", + num_layers=3, + last_offloading_layer=2, + ) + + assert profile == { + "id": "dev-1", + "model_dir": "FOMO_96_CUT", + "model_key": "fomo_96x96", + "ultimo_layer": 2, + "device_inference_times": {"layer_0": 1, "layer_1": 1, "layer_2": 1}, + "edge_inference_times": {"layer_0": 0.1, "layer_1": 0.1, "layer_2": 0.1}, + } + + +def test_register_is_idempotent_and_preserves_state(manager): + first = manager.register("dev-1", num_layers=2) + first["device_inference_times"]["layer_0"] = 0.42 + + second = manager.register("dev-1", num_layers=2) + + assert second is first + assert second["device_inference_times"]["layer_0"] == 0.42 + + +def test_profiles_is_the_live_mapping(manager): + # `http_server` reads `device_profiles` directly, so it must not be a copy. + profile = manager.register("dev-1", num_layers=1) + + assert manager.profiles["dev-1"] is profile + assert "dev-1" in manager + + +def test_build_profile_omits_ultimo_layer_when_unknown(): + assert "ultimo_layer" not in build_profile("dev-1", num_layers=1) + + +def test_update_device_times_applies_ema_to_known_layers(manager): + manager.register("dev-1", num_layers=2) + + manager.update_device_times("dev-1", [3.0, 5.0], alpha=0.5) + + times = manager.profiles["dev-1"]["device_inference_times"] + # Seeded at 1: 0.5*3 + 0.5*1 = 2.0 ; 0.5*5 + 0.5*1 = 3.0 + assert times == {"layer_0": pytest.approx(2.0), "layer_1": pytest.approx(3.0)} + + +def test_update_device_times_seeds_unknown_layers_with_raw_measurement(manager): + manager.register("dev-1", num_layers=1) + + manager.update_device_times("dev-1", [3.0, 7.0], alpha=0.5) + + times = manager.profiles["dev-1"]["device_inference_times"] + assert times["layer_1"] == pytest.approx(7.0) + + +def test_update_edge_times_offsets_by_start_layer(manager): + manager.register("dev-1", num_layers=4) + + manager.update_edge_times("dev-1", [1.0, 2.0], start_layer=2, alpha=1.0) + + times = manager.profiles["dev-1"]["edge_inference_times"] + assert times["layer_0"] == pytest.approx(0.1) + assert times["layer_1"] == pytest.approx(0.1) + assert times["layer_2"] == pytest.approx(1.0) + assert times["layer_3"] == pytest.approx(2.0) + + +def test_update_registers_unseen_device(manager): + manager.update_device_times("brand-new", [1.5], alpha=0.5) + + assert "brand-new" in manager + assert manager.model_dir("brand-new") == DEFAULT_MODEL_DIR + assert manager.model_key("brand-new") == DEFAULT_MODEL_KEY + + +def test_variance_detectors_are_isolated_per_device(manager): + # Heterogeneous devices report the same layer indices; their measurements + # must not land in a shared histogram. + manager.register("fast-device", num_layers=1) + manager.register("slow-device", num_layers=1) + + manager.update_device_times("fast-device", [0.01], alpha=0.5) + manager.update_device_times("slow-device", [5.0], alpha=0.5) + + fast = manager.variance_stats("fast-device")["device"][0] + slow = manager.variance_stats("slow-device")["device"][0] + assert fast["measurements"] == 1 + assert slow["measurements"] == 1 + assert fast["mean"] == pytest.approx(0.01) + assert slow["mean"] == pytest.approx(5.0) + + +def test_variance_detector_and_adaptive_state_are_stable_per_device(manager): + assert manager.variance_detector("dev-1") is manager.variance_detector("dev-1") + assert manager.adaptive_state("dev-1") is manager.adaptive_state("dev-1") + assert manager.adaptive_state("dev-1") is not manager.adaptive_state("dev-2") + + +def test_variance_stats_for_unknown_device_is_empty(manager): + assert manager.variance_stats("ghost")["device"] == {} + + +def test_snapshot_times_returns_copies(manager): + manager.register("dev-1", num_layers=1) + + device_times, edge_times = manager.snapshot_times("dev-1") + device_times.append(999) + + assert manager.snapshot_times("dev-1")[0] == [1] + assert edge_times == [0.1] + + +def test_snapshot_times_rejects_unknown_device(manager): + with pytest.raises(ValueError, match="Unknown device_id 'ghost'"): + manager.snapshot_times("ghost") + + +def test_snapshot_debug_times_returns_copies_and_tolerates_unknown(manager): + manager.register("dev-1", num_layers=1) + + device_times, edge_times = manager.snapshot_debug_times("dev-1") + device_times["layer_0"] = 999 + + assert manager.profiles["dev-1"]["device_inference_times"]["layer_0"] == 1 + assert edge_times == {"layer_0": 0.1} + assert manager.snapshot_debug_times("ghost") == ({}, {}) + + +def test_layer_sizes_reads_each_model_once(manager, tmp_path, monkeypatch): + sizes_file = tmp_path / "sizes.json" + sizes_file.write_text(json.dumps({"layer_0": 10, "layer_1": 20})) + reads = [] + + def fake_path(model_dir): + reads.append(model_dir) + return sizes_file + + monkeypatch.setattr( + "server.communication.device_state.OffloadingDataFiles.get_sizes_path", + fake_path, + ) + + assert manager.layer_sizes("FOMO_96_CUT") == [10, 20] + assert manager.layer_sizes("FOMO_96_CUT") == [10, 20] + assert reads == ["FOMO_96_CUT"] + + +def test_load_stats_uses_the_device_model_dir(manager, tmp_path, monkeypatch): + sizes_file = tmp_path / "sizes.json" + sizes_file.write_text(json.dumps({"layer_0": 10})) + monkeypatch.setattr( + "server.communication.device_state.OffloadingDataFiles.get_sizes_path", + lambda model_dir: sizes_file, + ) + manager.register("dev-1", model_dir="FOMO_96_CUT", num_layers=1) + + device_times, edge_times, layer_sizes = manager.load_stats("dev-1") + + assert device_times == [1] + assert edge_times == [0.1] + assert layer_sizes == [10] + + +def test_clear_drops_all_state(manager): + manager.register("dev-1", num_layers=1) + manager.variance_detector("dev-1") + + manager.clear() + + assert "dev-1" not in manager + assert manager.profiles == {} + + +def test_concurrent_ema_updates_do_not_lose_measurements(manager): + # Regression: `device_profiles` was mutated from the ASGI thread and the + # background I/O thread with no lock. + manager.register("dev-1", num_layers=1) + barrier = threading.Barrier(8) + errors = [] + + def hammer(): + try: + barrier.wait() + for _ in range(200): + manager.update_device_times("dev-1", [2.0], alpha=0.5) + manager.snapshot_debug_times("dev-1") + except Exception as exc: # pragma: no cover - only on a real race + errors.append(exc) + + threads = [threading.Thread(target=hammer) for _ in range(8)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert errors == [] + stats = manager.variance_stats("dev-1")["device"][0] + assert stats["measurements"] == 10 # window_size cap + assert manager.profiles["dev-1"]["device_inference_times"]["layer_0"] == ( + pytest.approx(2.0) + ) diff --git a/tests/unit/test_edge_inference_time_ownership.py b/tests/unit/test_edge_inference_time_ownership.py new file mode 100644 index 0000000..17e22f8 --- /dev/null +++ b/tests/unit/test_edge_inference_time_ownership.py @@ -0,0 +1,85 @@ +"""Regression tests: DeviceStateManager is the only owner of the edge EMA. + +Edge per-layer times used to be smoothed twice — once inside +`track_inference_time` with a hard-coded alpha=0.2, and again by the request +handler with the configured alpha (#72). `ModelManager` also fed a second, +never-consulted `VarianceDetector` (#73). +""" + +import inspect + +import pytest + +from server.communication.device_state import DeviceStateManager +from server.edge.edge_initialization import Edge +from server.models.model_manager import ModelManager, track_inference_time + + +class FakeModelManager: + """Minimal stand-in exposing the attributes the decorator may touch.""" + + def __init__(self): + self.calls = [] + + @track_inference_time + def predict_single_layer(self, layer_id, layer_offset, *args, **kwargs): + self.calls.append((layer_id, layer_offset)) + return "prediction" + + +def test_track_inference_time_is_pure_instrumentation(): + manager = FakeModelManager() + + result = manager.predict_single_layer(3, 1) + + assert result == "prediction" + assert manager.calls == [(3, 1)] + # The decorator must not smuggle timing state onto the instance: the + # authoritative timings are the ones run_inference returns. + assert not hasattr(manager, "inference_times") + assert not hasattr(manager, "variance_detector") + + +def test_model_manager_no_longer_owns_timing_or_variance_state(): + signature = inspect.signature(ModelManager.__init__) + + assert "inference_times" not in signature.parameters + assert "variance_detector" not in signature.parameters + assert not hasattr(ModelManager, "save_inference_times") + + +def test_edge_has_no_module_level_variance_detector(): + # The detector here accumulated measurements that no decision ever read. + assert not hasattr(Edge, "variance_detector") + + +def test_run_inference_does_not_take_an_ema_table(): + signature = inspect.signature(Edge.run_inference) + + assert "edge_inference_times" not in signature.parameters + + +def test_edge_ema_is_applied_exactly_once_with_the_configured_alpha(): + manager = DeviceStateManager() + manager.register("dev-1", num_layers=2) + + # Seeded at 0.1; a single EMA pass with alpha=0.5 against a 0.5s measurement + # gives 0.5*0.5 + 0.5*0.1 = 0.3. Double smoothing would land lower. + manager.update_edge_times("dev-1", [0.5], start_layer=0, alpha=0.5) + + times = manager.profiles["dev-1"]["edge_inference_times"] + assert times["layer_0"] == pytest.approx(0.3) + + +def test_configured_alpha_reaches_the_edge_ema(): + # Regression for the hard-coded alpha=0.2 that ignored settings.yaml. + manager = DeviceStateManager() + manager.register("dev-1", num_layers=1) + + manager.update_edge_times("dev-1", [0.5], start_layer=0, alpha=1.0) + + # alpha=1.0 means "trust the new measurement fully"; a hard-coded 0.2 would + # have produced 0.18 instead. + assert manager.profiles["dev-1"]["edge_inference_times"]["layer_0"] == ( + pytest.approx(0.5) + ) 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_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 + } 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 d97f8a4..168e8ea 100644 --- a/tests/unit/test_request_handler_helpers.py +++ b/tests/unit/test_request_handler_helpers.py @@ -1,7 +1,5 @@ """Unit tests for pure helpers in request_handler.""" -from pathlib import Path - import numpy as np import pytest @@ -253,49 +251,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) - monkeypatch.setattr( - rh.RequestHandler.variance_detector, - "get_all_stats", - lambda: {"needs_retest": False}, - ) - - handler = object.__new__(rh.RequestHandler) - handler.offloading_config = {"algorithm": "adaptive_risk"} - monkeypatch.setitem( - rh.RequestHandler.device_profiles, - "factory-device", - {"model_dir": "FOMO_96_CUT"}, - ) - monkeypatch.setattr(rh.RequestHandler, "offloading_states", {}) - - 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 rh.RequestHandler.offloading_states[ - "factory-device" - ] - - def test_append_and_publish_inference_cycle_keeps_local_write(monkeypatch): calls = [] @@ -333,94 +288,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, - ), - ]