From d362acc5fee0a87c0de7651404951fa831a2e6b9 Mon Sep 17 00:00:00 2001 From: "bedogni@unimore.it" Date: Fri, 10 Jul 2026 14:16:08 +0200 Subject: [PATCH 1/4] Add config-driven adaptive offloading --- docs/CONFIGURATION.md | 23 ++ docs/OFFLOADING_DECISION_EVENTS.md | 6 +- docs/config/server.full.yaml | 10 + src/sciot/config.py | 91 ++++++ src/server/communication/request_handler.py | 80 +++-- src/server/offloading_algo/adaptive_risk.py | 298 ++++++++++++++++++ src/server/offloading_algo/factory.py | 94 ++++++ src/server/offloading_algo/offloading_algo.py | 6 + src/server/settings.yaml | 10 + tests/unit/test_adaptive_offloading.py | 115 +++++++ tests/unit/test_config_validation.py | 51 +++ tests/unit/test_request_handler_helpers.py | 65 ++++ 12 files changed, 829 insertions(+), 20 deletions(-) create mode 100644 src/server/offloading_algo/adaptive_risk.py create mode 100644 src/server/offloading_algo/factory.py create mode 100644 tests/unit/test_adaptive_offloading.py diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 4b64e64..02c6e6e 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -38,12 +38,35 @@ Server validation covers: - HTTP/WebSocket/MQTT hostnames, ports, endpoints/topics, NTP server names, and model references; - model dimensions, split-layer limits, and model directories; - local-inference probability in the inclusive range `0.0..1.0`; +- offloading algorithm selection under `offloading.algorithm` (`static` or `adaptive_risk`) and adaptive tuning values; - delay sections and delay distributions; - boolean flags such as `verbose`, `debug_cprofiler`, compression, and input saving; - `evaluation.split.max_rows` and `evaluation.split.interval_minutes` (non-negative); - `evaluation.outputs.inference_cycles` and `evaluation.outputs.offloading_decisions` (booleans); - optional Firestore telemetry publishing under `evaluation.firestore`. +## Offloading algorithm + +The server chooses the next split point through the configured offloading +algorithm. If the section is absent, the runtime keeps the legacy `static` +behavior. To enable the adaptive risk-aware policy: + +```yaml +offloading: + algorithm: adaptive_risk # static | adaptive_risk + adaptive_risk: + network_ewma_alpha: 0.35 + network_window_size: 10 + uncertainty_weight: 0.50 + hysteresis_margin_ms: 5 + min_speed_bytes_per_second: 1 + probe_probability: 0.05 +``` + +`adaptive_risk` reuses the same candidate cost components as `static`, then +adds penalties for unstable device/edge measurements, noisy network speed, and +small layer switches. + ## Evaluation outputs The server writes inference-cycle CSV output under diff --git a/docs/OFFLOADING_DECISION_EVENTS.md b/docs/OFFLOADING_DECISION_EVENTS.md index 1bca133..fbbdad9 100644 --- a/docs/OFFLOADING_DECISION_EVENTS.md +++ b/docs/OFFLOADING_DECISION_EVENTS.md @@ -23,7 +23,7 @@ structure. | `device_id` | string | Client/device identifier used for registration and split requests. | | `model_dir` | string | Model artifact directory/profile used for the decision. | | `request_id` | string | Correlation ID linking the client request, prediction, and decision event. | -| `strategy` | string | Algorithm family, currently `static`. | +| `strategy` | string | Algorithm family, such as `static` or `adaptive_risk`. | | `selected_layer` | integer | Selected offloading layer; `-1` means local-only fallback. | | `selection_reason` | string | Why the layer was selected, such as `lowest_estimated_cost`, `forced_local_inference`, or `fallback_missing_model_metadata`. | | `avg_speed_bytes_per_second` | number | Network speed estimate used by the algorithm. | @@ -43,6 +43,10 @@ Each candidate includes: | `considered_for_selection` | boolean | Whether the candidate participates in the final choice. | | `selected` | boolean | Whether this candidate matches the selected layer. | +Adaptive candidates may also include `adaptive_score`, `risk_penalty`, +`switch_penalty`, `network_speed_ewma`, `network_uncertainty`, +`device_uncertainty`, and `edge_uncertainty`. + ## Example ```csv diff --git a/docs/config/server.full.yaml b/docs/config/server.full.yaml index 604f536..be9ddbf 100644 --- a/docs/config/server.full.yaml +++ b/docs/config/server.full.yaml @@ -70,6 +70,16 @@ local_inference_mode: offloading_algo: ema_alpha: 0.5 +offloading: + algorithm: adaptive_risk # static | adaptive_risk + adaptive_risk: + network_ewma_alpha: 0.35 + network_window_size: 10 + uncertainty_weight: 0.50 + hysteresis_margin_ms: 5 + min_speed_bytes_per_second: 1 + probe_probability: 0.05 + delay_simulation: computation: enabled: false diff --git a/src/sciot/config.py b/src/sciot/config.py index e347990..3ad6750 100644 --- a/src/sciot/config.py +++ b/src/sciot/config.py @@ -20,6 +20,7 @@ VALID_TRANSPORTS = {"http", "websocket", "mqtt"} VALID_DELAY_TYPES = {"none", "static", "gaussian", "uniform", "exponential"} +VALID_OFFLOADING_ALGORITHMS = {"static", "adaptive_risk"} SERVER_REQUIRED_ENDPOINTS = { "registration", @@ -118,6 +119,7 @@ def validate_server_config(config: Mapping[str, Any]) -> dict[str, Any]: "local_inference_mode", errors, ) + _validate_offloading_config(normalized.get("offloading", {}), "offloading", errors) _optional_bool(normalized, "verbose", errors) _optional_bool(normalized, "debug_cprofiler", errors) _validate_evaluation_config(normalized.get("evaluation", {}), "evaluation", errors) @@ -712,6 +714,95 @@ def _validate_probability_block(value: Any, path: str, errors: list[str]): errors.append(f"{path}.probability: must be between 0.0 and 1.0") +def _validate_offloading_config(value: Any, path: str, errors: list[str]): + if value in (None, {}): + return + if not isinstance(value, dict): + errors.append(f"{path}: must be a mapping") + return + + allowed = {"algorithm", "adaptive_risk"} + unexpected = set(value) - allowed + if unexpected: + errors.append( + f"{path}: unsupported keys {sorted(unexpected)}; expected {sorted(allowed)}" + ) + + algorithm = value.get("algorithm", "static") + if not isinstance(algorithm, str) or algorithm not in VALID_OFFLOADING_ALGORITHMS: + errors.append( + f"{path}.algorithm: must be one of {sorted(VALID_OFFLOADING_ALGORITHMS)}" + ) + + adaptive = value.get("adaptive_risk") + if adaptive is None: + return + if not isinstance(adaptive, dict): + errors.append(f"{path}.adaptive_risk: must be a mapping") + return + + adaptive_allowed = { + "network_ewma_alpha", + "network_window_size", + "uncertainty_weight", + "hysteresis_margin_ms", + "min_speed_bytes_per_second", + "probe_probability", + } + adaptive_unexpected = set(adaptive) - adaptive_allowed + if adaptive_unexpected: + errors.append( + f"{path}.adaptive_risk: unsupported keys {sorted(adaptive_unexpected)}; " + f"expected {sorted(adaptive_allowed)}" + ) + + alpha = _optional_number( + adaptive, + "network_ewma_alpha", + errors, + path=f"{path}.adaptive_risk.network_ewma_alpha", + ) + if alpha is not None and not 0 < alpha <= 1: + errors.append( + f"{path}.adaptive_risk.network_ewma_alpha: must be between 0.0 and 1.0" + ) + + _optional_positive_int( + adaptive, + "network_window_size", + errors, + path=f"{path}.adaptive_risk.network_window_size", + ) + _optional_non_negative_number( + adaptive, + "uncertainty_weight", + errors, + path=f"{path}.adaptive_risk.uncertainty_weight", + ) + _optional_non_negative_number( + adaptive, + "hysteresis_margin_ms", + errors, + path=f"{path}.adaptive_risk.hysteresis_margin_ms", + ) + _optional_positive_number( + adaptive, + "min_speed_bytes_per_second", + errors, + path=f"{path}.adaptive_risk.min_speed_bytes_per_second", + ) + probe_probability = _optional_number( + adaptive, + "probe_probability", + errors, + path=f"{path}.adaptive_risk.probe_probability", + ) + if probe_probability is not None and not 0 <= probe_probability <= 1: + errors.append( + f"{path}.adaptive_risk.probe_probability: must be between 0.0 and 1.0" + ) + + def _model_reference( config: Mapping[str, Any], key: str, diff --git a/src/server/communication/request_handler.py b/src/server/communication/request_handler.py index 37e3439..a3f83da 100644 --- a/src/server/communication/request_handler.py +++ b/src/server/communication/request_handler.py @@ -8,13 +8,18 @@ import math from datetime import datetime from pathlib import Path -import numpy as np from PIL import Image import hashlib from server.commons import ModelFiles from server.edge.edge_initialization import Edge -from server.offloading_algo.offloading_algo import OffloadingAlgo +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 @@ -127,6 +132,11 @@ def load_offloading_ema_alpha_config() -> float: return float(_get_settings().get("offloading_algo", {}).get("ema_alpha", 0.5)) +def load_offloading_config(): + """Load the configured offloading algorithm and tuning parameters.""" + return _get_settings().get("offloading", {}) or {} + + # ── Background I/O writer ─────────────────────────────────────────────────── # A single daemon thread drains a queue of callables, so that debug-JSON, # simulation-CSV, and evaluation-CSV writes never block the inference path. @@ -192,6 +202,7 @@ class RequestHandler: # Dictionary to store device profile data device_profiles = {} + offloading_states = {} num_layers = 0 def __init__(self): @@ -248,6 +259,7 @@ def __init__(self): self.evaluation_split_config = load_evaluation_split_config() # Load which evaluation CSV outputs are enabled self.evaluation_outputs_config = load_evaluation_outputs_config() + self.offloading_config = load_offloading_config() self.firestore_publisher = FirestoreTelemetryPublisher.from_config( load_evaluation_firestore_config(), default_run_document=EvaluationFiles.server_run_id(), @@ -288,7 +300,6 @@ def _cleanup_debug_folder(self): # Debug file saving def _save_debug_files(self, device_id): """Legacy sync method – prefer _save_debug_files_data for bg thread.""" - debug_dir = "data/models/debug" profile = RequestHandler.device_profiles[device_id] self._save_debug_files_data( device_id, @@ -375,6 +386,35 @@ def should_force_local_inference(self) -> bool: return should_force + def _create_offloading_algorithm( + self, + *, + device_id: str, + avg_speed: float, + device_inference_times: list, + edge_inference_times: list, + layers_sizes: list, + ): + 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" + ), + 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( @@ -639,6 +679,11 @@ def handle_device_inference_result(self, body, received_timestamp): # 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(): @@ -646,22 +691,25 @@ def handle_device_inference_result(self, body, received_timestamp): selection_reason = "forced_local_inference" decision_candidates = [] else: - offloading_algo = OffloadingAlgo( + offloading_algo = self._create_offloading_algorithm( + device_id=device_id, avg_speed=self.last_avg_speed, - num_layers=len(layers_sizes), + device_inference_times=list(device_inference_times), + edge_inference_times=list(edge_inference_times), 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" - ), ) + 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.static_offloading() - selection_reason = "lowest_estimated_cost" + 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! @@ -676,12 +724,6 @@ def handle_device_inference_result(self, body, received_timestamp): decision_candidates = ( offloading_algo.candidate_evaluations if offloading_algo else [] ) - - offloading_algorithm_class = ( - offloading_algo.__class__.__name__ - if offloading_algo is not None - else OffloadingAlgo.__name__ - ) server_start_timestamp = EvaluationFiles.server_start_timestamp() _eval_split = self.evaluation_split_config @@ -699,6 +741,7 @@ def handle_device_inference_result(self, body, received_timestamp): selection_reason=selection_reason, avg_speed_bytes_per_second=float(self.last_avg_speed), candidates=decision_candidates, + strategy=offloading_strategy, server_start_timestamp=server_start_timestamp, offloading_algorithm_class=offloading_algorithm_class, observed={ @@ -879,7 +922,6 @@ def _extend_message_data( def build_model_registry(cls, models_config: dict): for model_name, model_config in models_config.items(): model_dir = model_config["model_dir"] - h5_path = ModelFiles.get_model_h5_path(model_dir) try: hasher = hashlib.md5() tflite_dir = ( diff --git a/src/server/offloading_algo/adaptive_risk.py b/src/server/offloading_algo/adaptive_risk.py new file mode 100644 index 0000000..a60c833 --- /dev/null +++ b/src/server/offloading_algo/adaptive_risk.py @@ -0,0 +1,298 @@ +from __future__ import annotations + +import itertools +import random +import statistics +from dataclasses import dataclass, field +from typing import Any + +from server.offloading_algo.offloading_algo import OffloadingAlgo, _DEBUG_ENABLED + + +@dataclass +class AdaptiveRiskConfig: + network_ewma_alpha: float = 0.35 + network_window_size: int = 10 + uncertainty_weight: float = 0.50 + hysteresis_margin_ms: float = 5.0 + min_speed_bytes_per_second: float = 1.0 + probe_probability: float = 0.05 + + @classmethod + def from_mapping(cls, value: dict[str, Any] | None) -> "AdaptiveRiskConfig": + value = value or {} + return cls( + network_ewma_alpha=float(value.get("network_ewma_alpha", 0.35)), + network_window_size=int(value.get("network_window_size", 10)), + uncertainty_weight=float(value.get("uncertainty_weight", 0.50)), + hysteresis_margin_ms=float(value.get("hysteresis_margin_ms", 5.0)), + min_speed_bytes_per_second=float(value.get("min_speed_bytes_per_second", 1.0)), + probe_probability=float(value.get("probe_probability", 0.05)), + ) + + +@dataclass +class AdaptiveRiskState: + previous_selected_layer: int | None = None + network_speed_ewma: float | None = None + network_speed_samples: list[float] = field(default_factory=list) + + def observe_network_speed(self, raw_speed: float, config: AdaptiveRiskConfig) -> float: + speed = float(raw_speed or 0.0) + if speed <= 0: + speed = self.network_speed_ewma or config.min_speed_bytes_per_second + speed = max(config.min_speed_bytes_per_second, speed) + + if self.network_speed_ewma is None: + self.network_speed_ewma = speed + else: + alpha = config.network_ewma_alpha + self.network_speed_ewma = ( + alpha * speed + (1.0 - alpha) * self.network_speed_ewma + ) + + self.network_speed_samples.append(speed) + max_samples = max(1, config.network_window_size) + if len(self.network_speed_samples) > max_samples: + del self.network_speed_samples[: len(self.network_speed_samples) - max_samples] + + return self.network_speed_ewma + + def network_cv(self) -> float: + samples = self.network_speed_samples + if len(samples) < 2: + return 0.0 + mean = statistics.mean(samples) + if mean <= 0: + return 0.0 + return statistics.stdev(samples) / mean + + +class AdaptiveRiskOffloadingAlgo(OffloadingAlgo): + strategy = "adaptive_risk" + + def __init__( + self, + *, + config: AdaptiveRiskConfig, + state: AdaptiveRiskState, + variance_stats: dict[str, Any] | None = None, + **kwargs, + ) -> None: + super().__init__(**kwargs) + self.config = config + self.state = state + self.variance_stats = variance_stats or {} + self.network_speed_ewma = self.state.observe_network_speed( + self.avg_speed, + self.config, + ) + self.network_uncertainty = self.state.network_cv() + self.selection_reason = "adaptive_lowest_risk_score" + + def select_offloading_layer(self) -> int: + return self.adaptive_offloading() + + def adaptive_offloading(self) -> int: + if _DEBUG_ENABLED: + from server.logger.log import logger + + logger.debug(f"Performing Adaptive Risk Offloading: {self.num_layers} layers") + + self._record_edge_only_candidate() + self._record_split_candidates() + self._record_device_only_candidate() + + selected = self._selected_considered_candidate() + if selected is None: + self.best_offloading_layer = -1 + self.lowest_evaluation = 0.0 + self.selection_reason = "fallback_no_adaptive_candidates" + else: + self.best_offloading_layer = int(selected["offloading_layer_index"]) + self.lowest_evaluation = float(selected["adaptive_score"]) + self.selection_reason = "adaptive_lowest_risk_score" + + if self._should_probe_local(): + self.best_offloading_layer = -1 + self.selection_reason = "adaptive_probe_local_inference" + + self.state.previous_selected_layer = self.best_offloading_layer + return self.best_offloading_layer + + def _record_edge_only_candidate(self) -> None: + edge_compute_cost = sum(self.inference_time_edge[: self.num_layers]) + risk_penalty = self._risk_penalty( + base_latency=edge_compute_cost, + device_range=range(0, 0), + edge_range=range(0, self.num_layers), + ) + self._record_adaptive_candidate( + strategy="edge_only", + layer=-1, + device_compute_cost=0.0, + network_cost=0.0, + edge_compute_cost=edge_compute_cost, + base_latency=edge_compute_cost, + risk_penalty=risk_penalty, + considered_for_selection=False, + ) + + def _record_split_candidates(self) -> None: + layers_to_check = range(0, self.num_layers - 1) + if self.valid_offloading_points: + layers_to_check = [ + layer + for layer in layers_to_check + if layer in self.valid_offloading_points + ] + + device_prefix = list(itertools.accumulate(self.inference_time_device)) + edge_prefix = list(itertools.accumulate(self.inference_time_edge)) + total_edge = edge_prefix[self.num_layers - 1] if self.num_layers > 0 else 0.0 + + for layer in layers_to_check: + device_compute_cost = device_prefix[layer] + edge_compute_cost = ( + total_edge - edge_prefix[layer] if layer < self.num_layers - 1 else 0.0 + ) + network_cost = self.layers_sizes[layer] / self.network_speed_ewma + base_latency = device_compute_cost + network_cost + edge_compute_cost + risk_penalty = self._risk_penalty( + base_latency=base_latency, + device_range=range(0, layer + 1), + edge_range=range(layer + 1, self.num_layers), + ) + self._record_adaptive_candidate( + strategy="split", + layer=layer, + device_compute_cost=device_compute_cost, + network_cost=network_cost, + edge_compute_cost=edge_compute_cost, + base_latency=base_latency, + risk_penalty=risk_penalty, + considered_for_selection=True, + ) + + def _record_device_only_candidate(self) -> None: + device_compute_cost = sum(self.inference_time_device[: self.num_layers]) + network_cost = ( + self.layers_sizes[self.num_layers - 1] / self.network_speed_ewma + if self.num_layers > 0 + else 0.0 + ) + base_latency = device_compute_cost + network_cost + risk_penalty = self._risk_penalty( + base_latency=base_latency, + device_range=range(0, self.num_layers), + edge_range=range(0, 0), + ) + self._record_adaptive_candidate( + strategy="device_only", + layer=self.num_layers - 1, + device_compute_cost=device_compute_cost, + network_cost=network_cost, + edge_compute_cost=0.0, + base_latency=base_latency, + risk_penalty=risk_penalty, + considered_for_selection=True, + ) + + def _record_adaptive_candidate( + self, + *, + strategy: str, + layer: int, + device_compute_cost: float, + network_cost: float, + edge_compute_cost: float, + base_latency: float, + risk_penalty: float, + considered_for_selection: bool, + ) -> None: + switch_penalty = self._switch_penalty(layer) + adaptive_score = base_latency + risk_penalty + switch_penalty + self._record_candidate( + strategy=strategy, + layer=layer, + device_compute_cost=device_compute_cost, + network_cost=network_cost, + edge_compute_cost=edge_compute_cost, + estimated_total_cost=base_latency, + considered_for_selection=considered_for_selection, + ) + candidate = self.candidate_evaluations[-1] + candidate.update( + { + "adaptive_score": float(adaptive_score), + "risk_penalty": float(risk_penalty), + "switch_penalty": float(switch_penalty), + "network_speed_ewma": float(self.network_speed_ewma), + "network_uncertainty": float(self.network_uncertainty), + "device_uncertainty": float( + self._max_cv("device", self._device_layers_for(strategy, layer)) + ), + "edge_uncertainty": float( + self._max_cv("edge", self._edge_layers_for(strategy, layer)) + ), + } + ) + + def _selected_considered_candidate(self) -> dict[str, Any] | None: + considered = [ + candidate + for candidate in self.candidate_evaluations + if candidate.get("considered_for_selection") + ] + if not considered: + return None + return min(considered, key=lambda candidate: candidate["adaptive_score"]) + + def _risk_penalty( + self, + *, + base_latency: float, + device_range: range, + edge_range: range, + ) -> float: + uncertainty = max( + self.network_uncertainty, + self._max_cv("device", device_range), + self._max_cv("edge", edge_range), + ) + return base_latency * self.config.uncertainty_weight * uncertainty + + def _max_cv(self, section: str, layers: range) -> float: + stats_by_layer = self.variance_stats.get(section, {}) + max_cv = 0.0 + for layer in layers: + stats = stats_by_layer.get(layer) or stats_by_layer.get(str(layer)) or {} + max_cv = max(max_cv, float(stats.get("cv", 0.0) or 0.0)) + return max_cv + + def _switch_penalty(self, layer: int) -> float: + previous = self.state.previous_selected_layer + if previous is None or previous == layer: + return 0.0 + return max(0.0, self.config.hysteresis_margin_ms) / 1000.0 + + def _should_probe_local(self) -> bool: + if self.config.probe_probability <= 0: + return False + if not self.variance_stats.get("needs_retest"): + return False + return random.random() < self.config.probe_probability + + def _device_layers_for(self, strategy: str, layer: int) -> range: + if strategy == "edge_only": + return range(0, 0) + if strategy == "device_only": + return range(0, self.num_layers) + return range(0, layer + 1) + + def _edge_layers_for(self, strategy: str, layer: int) -> range: + if strategy == "edge_only": + return range(0, self.num_layers) + if strategy == "device_only": + return range(0, 0) + return range(layer + 1, self.num_layers) diff --git a/src/server/offloading_algo/factory.py b/src/server/offloading_algo/factory.py new file mode 100644 index 0000000..0c13b67 --- /dev/null +++ b/src/server/offloading_algo/factory.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from server.offloading_algo.adaptive_risk import ( + AdaptiveRiskConfig, + AdaptiveRiskOffloadingAlgo, + AdaptiveRiskState, +) +from server.offloading_algo.offloading_algo import OffloadingAlgo + + +DEFAULT_OFFLOADING_CONFIG = { + "algorithm": "static", + "adaptive_risk": { + "network_ewma_alpha": 0.35, + "network_window_size": 10, + "uncertainty_weight": 0.50, + "hysteresis_margin_ms": 5, + "min_speed_bytes_per_second": 1, + "probe_probability": 0.05, + }, +} + + +ALGORITHM_CLASSES = { + "static": OffloadingAlgo, + "adaptive_risk": AdaptiveRiskOffloadingAlgo, +} + + +@dataclass +class OffloadingContext: + avg_speed: float + num_layers: int + layers_sizes: list[float] + inference_time_device: list[float] + inference_time_edge: list[float] + model_dir: str + adaptive_state: AdaptiveRiskState | None = None + variance_stats: dict[str, Any] | None = None + + +def normalized_offloading_config(config: dict[str, Any] | None) -> dict[str, Any]: + config = config or {} + adaptive = { + **DEFAULT_OFFLOADING_CONFIG["adaptive_risk"], + **(config.get("adaptive_risk") or {}), + } + return { + "algorithm": config.get("algorithm", DEFAULT_OFFLOADING_CONFIG["algorithm"]), + "adaptive_risk": adaptive, + } + + +def configured_algorithm_name(config: dict[str, Any] | None) -> str: + return normalized_offloading_config(config)["algorithm"] + + +def configured_algorithm_class_name(config: dict[str, Any] | None) -> str: + algorithm_name = configured_algorithm_name(config) + return ALGORITHM_CLASSES[algorithm_name].__name__ + + +def create_offloading_algorithm( + config: dict[str, Any] | None, + context: OffloadingContext, +): + offloading_config = normalized_offloading_config(config) + algorithm_name = offloading_config["algorithm"] + + common_kwargs = { + "avg_speed": context.avg_speed, + "num_layers": context.num_layers, + "layers_sizes": context.layers_sizes, + "inference_time_device": context.inference_time_device, + "inference_time_edge": context.inference_time_edge, + "model_dir": context.model_dir, + } + + if algorithm_name == "static": + return OffloadingAlgo(**common_kwargs) + + if algorithm_name == "adaptive_risk": + state = context.adaptive_state or AdaptiveRiskState() + return AdaptiveRiskOffloadingAlgo( + **common_kwargs, + config=AdaptiveRiskConfig.from_mapping(offloading_config["adaptive_risk"]), + state=state, + variance_stats=context.variance_stats, + ) + + raise ValueError(f"Unsupported offloading algorithm: {algorithm_name}") diff --git a/src/server/offloading_algo/offloading_algo.py b/src/server/offloading_algo/offloading_algo.py index d2ebcfc..dccb76f 100644 --- a/src/server/offloading_algo/offloading_algo.py +++ b/src/server/offloading_algo/offloading_algo.py @@ -12,6 +12,8 @@ class OffloadingAlgo: + strategy = "static" + # ── Class-level cache for valid offloading points ──────────────────── # The JSON file never changes at runtime, so we read it once per model. _valid_points_cache = {} # model_dir -> Optional[set] @@ -235,6 +237,10 @@ def static_offloading(self) -> int: ) return self.best_offloading_layer + def select_offloading_layer(self) -> int: + """Return the selected layer for the configured algorithm interface.""" + return self.static_offloading() + def _record_candidate( self, *, diff --git a/src/server/settings.yaml b/src/server/settings.yaml index b39c168..c08d759 100644 --- a/src/server/settings.yaml +++ b/src/server/settings.yaml @@ -54,6 +54,16 @@ local_inference_mode: enabled: false probability: 0.2 +offloading: + algorithm: adaptive_risk # static | adaptive_risk + adaptive_risk: + network_ewma_alpha: 0.35 + network_window_size: 10 + uncertainty_weight: 0.50 + hysteresis_margin_ms: 5 + min_speed_bytes_per_second: 1 + probe_probability: 0.05 + # Splitting degli output di valutazione (CSV, JSONL e inference_cycles_*.csv). # I CSV di valutazione vengono spezzati in piu' segmenti: se ne apre uno nuovo # al primo criterio che scatta. I mirror JSONL delle decisioni usano un file per diff --git a/tests/unit/test_adaptive_offloading.py b/tests/unit/test_adaptive_offloading.py new file mode 100644 index 0000000..f7e2302 --- /dev/null +++ b/tests/unit/test_adaptive_offloading.py @@ -0,0 +1,115 @@ +import pytest + +from server.offloading_algo.adaptive_risk import ( + AdaptiveRiskConfig, + AdaptiveRiskOffloadingAlgo, + AdaptiveRiskState, +) +from server.offloading_algo.factory import ( + OffloadingContext, + configured_algorithm_name, + create_offloading_algorithm, + normalized_offloading_config, +) +from server.offloading_algo.offloading_algo import OffloadingAlgo + + +def _context(**overrides): + values = { + "avg_speed": 1_000_000.0, + "num_layers": 3, + "layers_sizes": [0.0, 0.0, 0.0], + "inference_time_device": [0.01, 0.5, 0.5], + "inference_time_edge": [0.0, 0.01, 0.1], + "model_dir": "FOMO_96_CUT", + "adaptive_state": AdaptiveRiskState(), + "variance_stats": {}, + } + values.update(overrides) + return OffloadingContext(**values) + + +@pytest.fixture(autouse=True) +def _disable_valid_points_loading(monkeypatch): + monkeypatch.setattr(OffloadingAlgo, "_load_valid_offloading_points", lambda *_: None) + + +def test_factory_defaults_to_static_algorithm(): + config = normalized_offloading_config({}) + + assert config["algorithm"] == "static" + assert configured_algorithm_name({}) == "static" + assert isinstance(create_offloading_algorithm({}, _context()), OffloadingAlgo) + + +def test_factory_creates_adaptive_risk_algorithm(): + algorithm = create_offloading_algorithm( + {"algorithm": "adaptive_risk"}, + _context(), + ) + + assert isinstance(algorithm, AdaptiveRiskOffloadingAlgo) + assert algorithm.strategy == "adaptive_risk" + + +def test_adaptive_risk_penalty_can_avoid_raw_fast_unstable_candidate(): + algorithm = create_offloading_algorithm( + { + "algorithm": "adaptive_risk", + "adaptive_risk": { + "uncertainty_weight": 10.0, + "probe_probability": 0.0, + }, + }, + _context(variance_stats={"edge": {1: {"cv": 1.0}}}), + ) + + selected_layer = algorithm.select_offloading_layer() + + assert selected_layer == 1 + split_0 = next( + candidate + for candidate in algorithm.candidate_evaluations + if candidate["strategy"] == "split" + and candidate["offloading_layer_index"] == 0 + ) + assert split_0["risk_penalty"] > 0 + assert split_0["adaptive_score"] > split_0["estimated_total_cost"] + + +def test_network_speed_state_uses_ewma_smoothing(): + state = AdaptiveRiskState() + config = AdaptiveRiskConfig(network_ewma_alpha=0.5) + + assert state.observe_network_speed(100.0, config) == 100.0 + assert state.observe_network_speed(10.0, config) == 55.0 + assert state.network_speed_samples == [100.0, 10.0] + + +def test_hysteresis_keeps_previous_layer_for_small_improvement(): + state = AdaptiveRiskState(previous_selected_layer=1) + algorithm = create_offloading_algorithm( + { + "algorithm": "adaptive_risk", + "adaptive_risk": { + "hysteresis_margin_ms": 5.0, + "probe_probability": 0.0, + }, + }, + _context( + adaptive_state=state, + inference_time_device=[0.05, 0.012, 1.0], + inference_time_edge=[0.0, 0.010, 0.040], + ), + ) + + selected_layer = algorithm.select_offloading_layer() + + assert selected_layer == 1 + switched_candidate = next( + candidate + for candidate in algorithm.candidate_evaluations + if candidate["strategy"] == "split" + and candidate["offloading_layer_index"] == 0 + ) + assert switched_candidate["switch_penalty"] == pytest.approx(0.005) diff --git a/tests/unit/test_config_validation.py b/tests/unit/test_config_validation.py index bbd9651..172d39b 100644 --- a/tests/unit/test_config_validation.py +++ b/tests/unit/test_config_validation.py @@ -162,6 +162,30 @@ def test_documented_example_configs_are_valid(path, loader): ), "offloading_algo: unsupported keys", ), + ( + lambda cfg: cfg.__setitem__( + "offloading", {"algorithm": "moonshot"} + ), + "offloading.algorithm: must be one of", + ), + ( + lambda cfg: cfg.__setitem__( + "offloading", {"adaptive_risk": {"network_ewma_alpha": 0}} + ), + "offloading.adaptive_risk.network_ewma_alpha: must be between 0.0 and 1.0", + ), + ( + lambda cfg: cfg.__setitem__( + "offloading", {"adaptive_risk": {"network_window_size": 0}} + ), + "offloading.adaptive_risk.network_window_size: must be greater than 0", + ), + ( + lambda cfg: cfg.__setitem__( + "offloading", {"adaptive_risk": {"probe_probability": 1.5}} + ), + "offloading.adaptive_risk.probe_probability: must be between 0.0 and 1.0", + ), ], ) def test_invalid_server_config_reports_field_errors(mutate, expected_error): @@ -172,6 +196,33 @@ def test_invalid_server_config_reports_field_errors(mutate, expected_error): validate_server_config(config) +def test_offloading_config_accepts_static_and_adaptive_risk(): + config = _server_config() + + config["offloading"] = {"algorithm": "static"} + assert validate_server_config(config)["offloading"]["algorithm"] == "static" + + config["offloading"] = { + "algorithm": "adaptive_risk", + "adaptive_risk": { + "network_ewma_alpha": 0.35, + "network_window_size": 10, + "uncertainty_weight": 0.5, + "hysteresis_margin_ms": 5, + "min_speed_bytes_per_second": 1, + "probe_probability": 0.05, + }, + } + assert validate_server_config(config)["offloading"]["algorithm"] == "adaptive_risk" + + +def test_offloading_config_can_be_absent_for_static_compatibility(): + config = _server_config() + config.pop("offloading", None) + + assert validate_server_config(config) + + @pytest.mark.parametrize( ("mutate", "expected_error"), [ diff --git a/tests/unit/test_request_handler_helpers.py b/tests/unit/test_request_handler_helpers.py index 5d49e14..5914a11 100644 --- a/tests/unit/test_request_handler_helpers.py +++ b/tests/unit/test_request_handler_helpers.py @@ -11,6 +11,7 @@ load_evaluation_firestore_config, load_evaluation_outputs_config, load_offloading_ema_alpha_config, + load_offloading_config, ) @@ -100,6 +101,70 @@ def test_offloading_ema_alpha_is_loaded(monkeypatch): assert load_offloading_ema_alpha_config() == 0.2 +def test_offloading_config_defaults_to_empty_for_static_compatibility(monkeypatch): + monkeypatch.setattr(rh, "_get_settings", lambda: {}) + assert load_offloading_config() == {} + + +def test_offloading_config_is_loaded(monkeypatch): + monkeypatch.setattr( + rh, + "_get_settings", + lambda: { + "offloading": { + "algorithm": "adaptive_risk", + "adaptive_risk": {"hysteresis_margin_ms": 7}, + } + }, + ) + + assert load_offloading_config() == { + "algorithm": "adaptive_risk", + "adaptive_risk": {"hysteresis_margin_ms": 7}, + } + + +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], + ) + + 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.adaptive_state is rh.RequestHandler.offloading_states[ + "factory-device" + ] + + def test_append_and_publish_inference_cycle_keeps_local_write(monkeypatch): calls = [] From fe74ccdeeae2e16a292b63ff732aa7832a80ed79 Mon Sep 17 00:00:00 2001 From: "bedogni@unimore.it" Date: Fri, 10 Jul 2026 14:44:35 +0200 Subject: [PATCH 2/4] Surface server transport startup failures --- .../communication/transport_lifecycle.py | 14 +++++++- src/server/edge/run_edge.py | 2 +- tests/unit/test_transport_lifecycle.py | 35 +++++++++++++++++++ 3 files changed, 49 insertions(+), 2 deletions(-) diff --git a/src/server/communication/transport_lifecycle.py b/src/server/communication/transport_lifecycle.py index efdec3b..4877de2 100644 --- a/src/server/communication/transport_lifecycle.py +++ b/src/server/communication/transport_lifecycle.py @@ -127,7 +127,11 @@ def _run(self) -> None: with self._lock: self.status = TransportStatus.FAILED self._ready.clear() - logger.exception("%s transport failed", self.name) + logger.critical( + "%s transport failed during startup/runtime", + self.name, + exc_info=True, + ) else: with self._lock: if self.status is TransportStatus.STOPPING: @@ -194,6 +198,14 @@ def run_forever(self) -> list[TransportSnapshot]: self.start_all() while not self._stop_requested.is_set(): if not any(transport.is_alive() for transport in self.transports): + if self.transports: + logger.critical( + "All communication transports stopped; server is exiting" + ) + else: + logger.critical( + "No communication transports were configured; server is exiting" + ) break time.sleep(0.25) finally: diff --git a/src/server/edge/run_edge.py b/src/server/edge/run_edge.py index eb98df0..1bf0741 100644 --- a/src/server/edge/run_edge.py +++ b/src/server/edge/run_edge.py @@ -116,7 +116,7 @@ def main(config_path: str | None = None) -> int: snapshots = coordinator.run_forever() failed_transports = [snapshot for snapshot in snapshots if snapshot.failure is not None] if failed_transports: - logger.error( + logger.critical( "One or more communication transports failed: %s", ", ".join(snapshot.name for snapshot in failed_transports), ) diff --git a/tests/unit/test_transport_lifecycle.py b/tests/unit/test_transport_lifecycle.py index 5776efc..e79a286 100644 --- a/tests/unit/test_transport_lifecycle.py +++ b/tests/unit/test_transport_lifecycle.py @@ -1,4 +1,5 @@ import threading +import logging from server.communication.transport_lifecycle import ( ManagedTransport, @@ -32,6 +33,14 @@ def stop(self): self.stopped = True +class ReturningFakeTransport: + def run(self): + return None + + def stop(self): + return None + + def test_managed_transport_starts_reports_ready_and_stops(): fake = BlockingFakeTransport() transport = ManagedTransport("http", fake) @@ -72,6 +81,18 @@ def test_coordinator_keeps_healthy_transport_running_after_startup_failure(): assert healthy_transport.status is TransportStatus.STOPPED +def test_failed_transport_logs_critical_startup_error(caplog): + transport = ManagedTransport("http", FailingFakeTransport("bind failed")) + + with caplog.at_level(logging.CRITICAL): + transport.start() + transport.join(1) + + assert transport.status is TransportStatus.FAILED + assert "http transport failed during startup/runtime" in caplog.text + assert "bind failed" in caplog.text + + def test_coordinator_shutdown_stops_all_running_transports(): first = BlockingFakeTransport() second = BlockingFakeTransport() @@ -112,3 +133,17 @@ def test_run_forever_exits_when_all_transports_stop(): thread.join(1) assert not thread.is_alive() + + +def test_run_forever_logs_critical_when_all_transports_stop(caplog): + coordinator = TransportCoordinator( + [ManagedTransport("http", ReturningFakeTransport())], + startup_timeout=0.02, + join_timeout=1, + ) + + with caplog.at_level(logging.CRITICAL): + snapshots = coordinator.run_forever() + + assert snapshots[0].status is TransportStatus.STOPPED + assert "All communication transports stopped; server is exiting" in caplog.text From 8a7484da2a3a204f0feedf635b10a4ac8728de6c Mon Sep 17 00:00:00 2001 From: "bedogni@unimore.it" Date: Fri, 10 Jul 2026 15:02:13 +0200 Subject: [PATCH 3/4] Repeat client and server table headers --- src/client/python/http_client.py | 89 +++++++++++++++++---- src/server/communication/request_handler.py | 60 ++++++++++++-- 2 files changed, 126 insertions(+), 23 deletions(-) diff --git a/src/client/python/http_client.py b/src/client/python/http_client.py index fbfbdae..18a801f 100644 --- a/src/client/python/http_client.py +++ b/src/client/python/http_client.py @@ -16,7 +16,6 @@ import ipaddress from concurrent.futures import ThreadPoolExecutor, as_completed from concurrent.futures import TimeoutError as FutureTimeoutError -import cProfile from sciot.config import load_client_config @@ -67,6 +66,15 @@ session.mount("http://", _adapter) session.mount("https://", _adapter) +CLIENT_TABLE_HEADER = ( + "Offload | Acq Time (ms) | Comp Time (ms) | Net Time (ms) | " + "Net Speed (KB/s) | Data (bytes)" +) +CLIENT_TABLE_SEPARATOR = "-" * len(CLIENT_TABLE_HEADER) +CLIENT_TABLE_HEADER_REPEAT_INTERVAL = 20 +_client_table_header_printed = False +_client_table_rows_printed = 0 + # ===================================================== # OPTIMIZATION: Dedicated discovery session with a larger pool # for parallel subnet scanning (up to 50 threads). @@ -171,7 +179,7 @@ def check_server_at_ip(ip, port, timeout=1.0): pass return None - except Exception as e: + except Exception: return None @@ -201,7 +209,7 @@ def discover_server(port): print(f"Local IP: {local_ip}") print(f"Scanning network: {network}") - print(f"This may take 10-30 seconds...\n") + print("This may take 10-30 seconds...\n") # Scan network with threading for speed found_servers = [] @@ -341,7 +349,7 @@ def _preload_interpreters(): print(f"Pre-loading {LAST_OFFLOADING_LAYER + 1} TFLite interpreters...") for i in range(LAST_OFFLOADING_LAYER + 1): _get_interpreter(i) - print(f"All interpreters loaded and cached.") + print("All interpreters loaded and cached.") # Function to generate a random message ID @@ -362,7 +370,7 @@ def register_device(): print("WARNING: Server does not have this model - running LOCAL-ONLY") return False return r.status_code == 200 - except requests.exceptions.RequestException as e: + except requests.exceptions.RequestException: return False @@ -413,14 +421,14 @@ def send_image(): payload = _send_image_buffer try: - r = session.post( + session.post( url, data=payload, headers=headers, timeout=5, ) return True - except requests.exceptions.RequestException as e: + except requests.exceptions.RequestException: return False @@ -691,6 +699,47 @@ def shutdown(self): self._shutdown_complete = True +def print_client_table_header(*, reset_rows: bool = False): + global _client_table_header_printed, _client_table_rows_printed + if reset_rows: + _client_table_rows_printed = 0 + print(f"\n{CLIENT_TABLE_HEADER}") + print(CLIENT_TABLE_SEPARATOR) + _client_table_header_printed = True + + +def print_client_table_row( + *, + layer: int, + acq_time: float, + comp_time: float, + net_time_ms: float | None = None, + net_speed: float | None = None, + data_size: int | None = None, +): + global _client_table_rows_printed + if ( + not _client_table_header_printed + or ( + _client_table_rows_printed > 0 + and _client_table_rows_printed % CLIENT_TABLE_HEADER_REPEAT_INTERVAL == 0 + ) + ): + print_client_table_header() + + if net_time_ms is None or net_speed is None or data_size is None: + print( + f"{layer:7d} | {acq_time:14.2f} | {comp_time:14.2f} | " + f"{'N/A':>12} | {'N/A':>16} | {'N/A':>12}" + ) + else: + print( + f"{layer:7d} | {acq_time:14.2f} | {comp_time:14.2f} | " + f"{net_time_ms:12.2f} | {net_speed:16.2f} | {data_size:12d}" + ) + _client_table_rows_printed += 1 + + def print_upload_result(upload_result): success, net_time, net_speed, data_size, layer, acq_time, comp_time = upload_result @@ -701,12 +750,19 @@ def print_upload_result(upload_result): # Se il layer eseguito è l'ultimo del modello, l'inferenza è stata # 100% locale. La rete non ha trasportato l'immagine. if layer >= LAST_OFFLOADING_LAYER: - print(f"{layer:7d} | {acq_time:14.2f} | {comp_time:14.2f} | {'N/A':>12} | {'N/A':>16} | {'N/A':>12}") + print_client_table_row(layer=layer, acq_time=acq_time, comp_time=comp_time) else: # Altrimenti c'è stato Offload (Totale a 0, o Parziale). - print(f"{layer:7d} | {acq_time:14.2f} | {comp_time:14.2f} | {net_time * 1000:12.2f} | {net_speed:16.2f} | {data_size:12d}") + print_client_table_row( + layer=layer, + acq_time=acq_time, + comp_time=comp_time, + net_time_ms=net_time * 1000, + net_speed=net_speed, + data_size=data_size, + ) else: - print(f"{layer:7d} | {acq_time:14.2f} | {comp_time:14.2f} | {'N/A':>12} | {'N/A':>16} | {'N/A':>12}") + print_client_table_row(layer=layer, acq_time=acq_time, comp_time=comp_time) return success @@ -767,7 +823,7 @@ def run_split_inference(image, tflite_dir, stop_layer): # Apply artificial computation delay if computation_delay.enabled: - delay = computation_delay.apply_delay() + computation_delay.apply_delay() interpreter.invoke() t1 = time.time() @@ -995,10 +1051,7 @@ def main( else: print("Connected to server\n") - print( - "Offload | Acq Time (ms) | Comp Time (ms) | Net Time (ms) | Net Speed (KB/s) | Data (bytes)" - ) - print("-" * 90) + print_client_table_header(reset_rows=True) # imposto un "timer" per non fare richieste in continuazione last_reconnect_attempt = time.time() @@ -1129,7 +1182,11 @@ def main( message_id, ) else: - print(f"{best_layer:7d} | {acq_time:14.2f} | {comp_time:14.2f} | {'N/A':>12} | {'N/A':>16} | {'N/A':>12}") + print_client_table_row( + layer=best_layer, + acq_time=acq_time, + comp_time=comp_time, + ) # ─── GESTIONE ESPORTAZIONI E PROFILING ─── cicli_completati += 1 diff --git a/src/server/communication/request_handler.py b/src/server/communication/request_handler.py index a3f83da..0d8962b 100644 --- a/src/server/communication/request_handler.py +++ b/src/server/communication/request_handler.py @@ -190,6 +190,13 @@ def _io_worker(): class RequestHandler: + INFERENCE_TABLE_HEADER = ( + "Device | Offload | Acq Time (ms) | Device Comp (ms) | " + "Edge Comp (ms) | Net Time (ms) | Total (ms)" + ) + INFERENCE_TABLE_SEPARATOR = "-" * len(INFERENCE_TABLE_HEADER) + INFERENCE_TABLE_HEADER_REPEAT_INTERVAL = 20 + 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) @@ -199,6 +206,7 @@ class RequestHandler: csv_writer = None inference_counter = 0 header_printed = False + inference_table_rows_printed = 0 # Dictionary to store device profile data device_profiles = {} @@ -219,11 +227,7 @@ def __init__(self): # Log header once if not RequestHandler.header_printed: - logger.info( - "\nDevice | Offload | Acq Time (ms) | Device Comp (ms) | Edge Comp (ms) | Net Time (ms) | Total (ms)\n" - + "-" * 100 - ) - RequestHandler.header_printed = True + RequestHandler._print_inference_table_header() # Empty the debug folder every time the server starts self._cleanup_debug_folder() @@ -386,6 +390,42 @@ def should_force_local_inference(self) -> bool: return should_force + @classmethod + def _print_inference_table_header(cls) -> None: + logger.info(f"\n{cls.INFERENCE_TABLE_HEADER}\n{cls.INFERENCE_TABLE_SEPARATOR}") + cls.header_printed = True + + @classmethod + def _print_inference_table_row( + cls, + *, + device_id: str, + offloading_layer_index: int, + acquisition_time_ms: float, + device_compute_time_ms: float, + edge_compute_time_ms: float, + network_time_ms: float, + total_time_ms: float, + ) -> None: + if ( + not cls.header_printed + or ( + cls.inference_table_rows_printed > 0 + and cls.inference_table_rows_printed + % cls.INFERENCE_TABLE_HEADER_REPEAT_INTERVAL + == 0 + ) + ): + cls._print_inference_table_header() + + logger.info( + f"{device_id:13s} | {offloading_layer_index:7d} | " + f"{acquisition_time_ms:13.2f} | {device_compute_time_ms:16.2f} | " + f"{edge_compute_time_ms:14.2f} | {network_time_ms:13.2f} | " + f"{total_time_ms:10.2f}" + ) + cls.inference_table_rows_printed += 1 + def _create_offloading_algorithm( self, *, @@ -713,8 +753,14 @@ def handle_device_inference_result(self, body, received_timestamp): decision_candidates = offloading_algo.candidate_evaluations # Stampiamo la tabella SOLO se il calcolo è andato a buon fine! - logger.info( - f"{device_id:13s} | {message_data.offloading_layer_index:7d} | {acq_time:13.2f} | {device_comp_time:16.2f} | {edge_comp_time:14.2f} | {network_time:13.2f} | {total_time:10.2f}" + RequestHandler._print_inference_table_row( + device_id=device_id, + offloading_layer_index=message_data.offloading_layer_index, + acquisition_time_ms=acq_time, + device_compute_time_ms=device_comp_time, + edge_compute_time_ms=edge_comp_time, + network_time_ms=network_time, + total_time_ms=total_time, ) except IndexError: From f585fe4614533a906e4b059e69c7fa544f3b899a Mon Sep 17 00:00:00 2001 From: "bedogni@unimore.it" Date: Fri, 10 Jul 2026 15:21:47 +0200 Subject: [PATCH 4/4] Add device load aware adaptive offloading --- docs/CONFIGURATION.md | 5 +- docs/OFFLOADING_DECISION_EVENTS.md | 3 +- docs/TRANSPORTS.md | 7 +-- docs/config/server.full.yaml | 1 + src/client/python/http_client.py | 62 +++++++++++++++++-- src/sciot/config.py | 7 +++ .../communication/inference_protocol.py | 47 ++++++++------ src/server/communication/message_data.py | 11 +++- src/server/communication/request_handler.py | 4 ++ src/server/offloading_algo/adaptive_risk.py | 37 ++++++++++- src/server/offloading_algo/factory.py | 3 + src/server/settings.yaml | 1 + tests/fixtures/protocol/inference_v1.hex | 1 - .../test_http_protocol_validation.py | 17 ++++- tests/unit/test_adaptive_offloading.py | 40 ++++++++++++ tests/unit/test_config_validation.py | 7 +++ tests/unit/test_inference_protocol.py | 57 +++++++---------- tests/unit/test_message_data.py | 18 ++++++ tests/unit/test_request_handler_helpers.py | 2 + 19 files changed, 255 insertions(+), 75 deletions(-) delete mode 100644 tests/fixtures/protocol/inference_v1.hex diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 02c6e6e..3896405 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -61,11 +61,14 @@ offloading: hysteresis_margin_ms: 5 min_speed_bytes_per_second: 1 probe_probability: 0.05 + device_load_weight: 0.50 ``` `adaptive_risk` reuses the same candidate cost components as `static`, then adds penalties for unstable device/edge measurements, noisy network speed, and -small layer switches. +small layer switches. It also scales device-side compute estimates by the +reported device CPU occupancy so a busy client moves toward earlier offloading +points faster. ## Evaluation outputs diff --git a/docs/OFFLOADING_DECISION_EVENTS.md b/docs/OFFLOADING_DECISION_EVENTS.md index fbbdad9..e53ed14 100644 --- a/docs/OFFLOADING_DECISION_EVENTS.md +++ b/docs/OFFLOADING_DECISION_EVENTS.md @@ -45,7 +45,8 @@ Each candidate includes: Adaptive candidates may also include `adaptive_score`, `risk_penalty`, `switch_penalty`, `network_speed_ewma`, `network_uncertainty`, -`device_uncertainty`, and `edge_uncertainty`. +`device_uncertainty`, `edge_uncertainty`, `device_cpu_percent`, +`device_load_factor`, `raw_device_compute_cost`, and `device_load_penalty`. ## Example diff --git a/docs/TRANSPORTS.md b/docs/TRANSPORTS.md index cbe086f..8bb879f 100644 --- a/docs/TRANSPORTS.md +++ b/docs/TRANSPORTS.md @@ -73,7 +73,7 @@ Versioned inference-result payloads start with this 8-byte header: | Field | Type | Value | | --- | --- | --- | | `magic` | `char[4]` | `SCIP` | -| `version` | `uint16` | `1` | +| `version` | `uint16` | `2` | | `flags` | `uint16` | `0` | The body is: @@ -86,14 +86,13 @@ The body is: | `message_id` | ASCII `char[4]` | client message identifier | | `offloading_layer_index` | `int32` | `-1` or `0..4095` | | `acquisition_time` | `float32` | seconds, finite and non-negative | +| `device_cpu_percent` | `float32` | whole-machine CPU occupancy, `0..100` | | `layer_output_size` | `uint32` | 1 to 32 MiB, multiple of `float32` | | `layer_output` | `float32[]` | split-layer tensor data | | `layer_times_size` | `uint32` | 0 to 64 KiB, multiple of `float32` | | `layer_times` | `float32[]` | per-layer device timings, seconds | -Compatibility rule: payloads that do not start with the `SCIP` magic are decoded -as legacy unversioned payloads using the same body layout. The decoder labels -those messages as protocol version `0` internally. Payloads with an unknown +Payloads without the `SCIP` magic/header are rejected. Payloads with an unknown version, non-zero flags, invalid sizes, unsupported field values, or trailing bytes are rejected as malformed inference payloads by HTTP, WebSocket, and MQTT. diff --git a/docs/config/server.full.yaml b/docs/config/server.full.yaml index be9ddbf..4359bca 100644 --- a/docs/config/server.full.yaml +++ b/docs/config/server.full.yaml @@ -79,6 +79,7 @@ offloading: hysteresis_margin_ms: 5 min_speed_bytes_per_second: 1 probe_probability: 0.05 + device_load_weight: 0.50 delay_simulation: computation: diff --git a/src/client/python/http_client.py b/src/client/python/http_client.py index 18a801f..296dc3d 100644 --- a/src/client/python/http_client.py +++ b/src/client/python/http_client.py @@ -19,6 +19,11 @@ from sciot.config import load_client_config +try: + import psutil +except ImportError: + psutil = None + try: from .delay_simulator import DelaySimulator from .profiler import AdvancedProfiler @@ -92,6 +97,18 @@ # ===================================================== COMPRESSION_ENABLED = config.get("compression", {}).get("enabled", False) +INFERENCE_PROTOCOL_MAGIC = b"SCIP" +INFERENCE_PROTOCOL_VERSION = 2 +INFERENCE_PROTOCOL_FLAGS = 0 +INFERENCE_PROTOCOL_HEADER_FORMAT = "<4sHH" +INFERENCE_PROTOCOL_HEADER_BYTES = struct.calcsize(INFERENCE_PROTOCOL_HEADER_FORMAT) + +if psutil is not None: + try: + psutil.cpu_percent(interval=None) + except Exception: + pass + def load_config(config_filename=None): """Legge il file di configurazione YAML usando il percorso assoluto.""" @@ -301,6 +318,19 @@ def set_device_id(device_id): _DEVICE_ID_BYTES = DEVICE_ID.encode("ascii") _DEVICE_ID_HEADER = struct.pack(" float: + """Return whole-machine CPU occupancy as a bounded percentage.""" + if psutil is None: + return 0.0 + try: + cpu_percent = float(psutil.cpu_percent(interval=None)) + except Exception: + return 0.0 + if not np.isfinite(cpu_percent): + return 0.0 + return min(100.0, max(0.0, cpu_percent)) + # Initialize delay simulators DELAY_CONFIG = config.get("delay_simulation", {}) computation_delay = DelaySimulator(DELAY_CONFIG.get("device_computation")) @@ -840,9 +870,13 @@ def build_inference_payload( acq_time_ms, *, timestamp=None, + device_cpu_percent=None, ): """Serialize an inference result using the validated little-endian protocol.""" timestamp = time.time() if timestamp is None else timestamp + if device_cpu_percent is None: + device_cpu_percent = get_device_cpu_percent() + device_cpu_percent = min(100.0, max(0.0, float(device_cpu_percent))) output_bytes = np.asarray(output_data, dtype=" MessageData: """Decode one inference-result message using the bounded wire format.""" try: reader = _PayloadReader(payload) - protocol_version = INFERENCE_PROTOCOL_LEGACY_VERSION - protocol_flags = 0 + if reader.payload[: len(INFERENCE_PROTOCOL_MAGIC)] != INFERENCE_PROTOCOL_MAGIC: + raise InvalidInferencePayload("missing inference protocol header") - if reader.payload[: len(INFERENCE_PROTOCOL_MAGIC)] == INFERENCE_PROTOCOL_MAGIC: - header = reader.read_bytes( - INFERENCE_PROTOCOL_HEADER_BYTES, - "protocol_header", - ) - magic, protocol_version, protocol_flags = struct.unpack( - INFERENCE_PROTOCOL_HEADER_FORMAT, - header, + header = reader.read_bytes( + INFERENCE_PROTOCOL_HEADER_BYTES, + "protocol_header", + ) + magic, protocol_version, protocol_flags = struct.unpack( + INFERENCE_PROTOCOL_HEADER_FORMAT, + header, + ) + if magic != INFERENCE_PROTOCOL_MAGIC: + raise InvalidInferencePayload("invalid protocol magic") + if protocol_version != INFERENCE_PROTOCOL_VERSION: + raise InvalidInferencePayload( + f"unsupported inference protocol version {protocol_version}" ) - if magic != INFERENCE_PROTOCOL_MAGIC: - raise InvalidInferencePayload("invalid protocol magic") - if protocol_version != INFERENCE_PROTOCOL_VERSION: - raise InvalidInferencePayload( - f"unsupported inference protocol version {protocol_version}" - ) - if protocol_flags != 0: - raise InvalidInferencePayload("unsupported inference protocol flags") + if protocol_flags != 0: + raise InvalidInferencePayload("unsupported inference protocol flags") timestamp = reader.read_struct(" MessageData: if not math.isfinite(acquisition_time) or acquisition_time < 0: raise InvalidInferencePayload("invalid acquisition time") + device_cpu_percent = reader.read_struct(" 100 + ): + raise InvalidInferencePayload("invalid device CPU percent") + layer_output_size = reader.read_struct(" MessageData: message_content = { "offloading_layer_index": offloading_layer_index, "acquisition_time": acquisition_time, + "device_cpu_percent": device_cpu_percent, "layer_output": np.frombuffer(layer_output_bytes, dtype=" tuple: device_layers_inference_time = message_content.get( "layers_inference_time", None ) - return offloading_layer_index, layer_output, device_layers_inference_time + device_cpu_percent = message_content.get("device_cpu_percent", None) + return ( + offloading_layer_index, + layer_output, + device_layers_inference_time, + device_cpu_percent, + ) except Exception as _: - return None, None, None + return None, None, None, None diff --git a/src/server/communication/request_handler.py b/src/server/communication/request_handler.py index 0d8962b..f8b4c75 100644 --- a/src/server/communication/request_handler.py +++ b/src/server/communication/request_handler.py @@ -434,6 +434,7 @@ def _create_offloading_algorithm( device_inference_times: list, edge_inference_times: list, layers_sizes: list, + device_cpu_percent: float, ): adaptive_state = RequestHandler.offloading_states.setdefault( device_id, @@ -450,6 +451,7 @@ def _create_offloading_algorithm( 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(), ), @@ -737,6 +739,7 @@ def handle_device_inference_result(self, body, received_timestamp): device_inference_times=list(device_inference_times), edge_inference_times=list(edge_inference_times), layers_sizes=list(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__ @@ -961,6 +964,7 @@ def _extend_message_data( message_data.offloading_layer_index, message_data.layer_output, message_data.device_layers_inference_time, + message_data.device_cpu_percent, ) = MessageData.get_offloading_info(message_data.message_content) return message_data diff --git a/src/server/offloading_algo/adaptive_risk.py b/src/server/offloading_algo/adaptive_risk.py index a60c833..8faefef 100644 --- a/src/server/offloading_algo/adaptive_risk.py +++ b/src/server/offloading_algo/adaptive_risk.py @@ -17,6 +17,7 @@ class AdaptiveRiskConfig: hysteresis_margin_ms: float = 5.0 min_speed_bytes_per_second: float = 1.0 probe_probability: float = 0.05 + device_load_weight: float = 0.50 @classmethod def from_mapping(cls, value: dict[str, Any] | None) -> "AdaptiveRiskConfig": @@ -28,6 +29,7 @@ def from_mapping(cls, value: dict[str, Any] | None) -> "AdaptiveRiskConfig": hysteresis_margin_ms=float(value.get("hysteresis_margin_ms", 5.0)), min_speed_bytes_per_second=float(value.get("min_speed_bytes_per_second", 1.0)), probe_probability=float(value.get("probe_probability", 0.05)), + device_load_weight=float(value.get("device_load_weight", 0.50)), ) @@ -77,12 +79,18 @@ def __init__( config: AdaptiveRiskConfig, state: AdaptiveRiskState, variance_stats: dict[str, Any] | None = None, + device_cpu_percent: float = 0.0, **kwargs, ) -> None: super().__init__(**kwargs) self.config = config self.state = state self.variance_stats = variance_stats or {} + self.device_cpu_percent = self._normalize_device_cpu_percent(device_cpu_percent) + self.device_load_factor = 1.0 + max( + 0.0, + self.config.device_load_weight, + ) * (self.device_cpu_percent / 100.0) self.network_speed_ewma = self.state.observe_network_speed( self.avg_speed, self.config, @@ -130,6 +138,7 @@ def _record_edge_only_candidate(self) -> None: self._record_adaptive_candidate( strategy="edge_only", layer=-1, + raw_device_compute_cost=0.0, device_compute_cost=0.0, network_cost=0.0, edge_compute_cost=edge_compute_cost, @@ -152,7 +161,10 @@ def _record_split_candidates(self) -> None: total_edge = edge_prefix[self.num_layers - 1] if self.num_layers > 0 else 0.0 for layer in layers_to_check: - device_compute_cost = device_prefix[layer] + raw_device_compute_cost = device_prefix[layer] + device_compute_cost = self._load_adjusted_device_cost( + raw_device_compute_cost + ) edge_compute_cost = ( total_edge - edge_prefix[layer] if layer < self.num_layers - 1 else 0.0 ) @@ -166,6 +178,7 @@ def _record_split_candidates(self) -> None: self._record_adaptive_candidate( strategy="split", layer=layer, + raw_device_compute_cost=raw_device_compute_cost, device_compute_cost=device_compute_cost, network_cost=network_cost, edge_compute_cost=edge_compute_cost, @@ -175,7 +188,8 @@ def _record_split_candidates(self) -> None: ) def _record_device_only_candidate(self) -> None: - device_compute_cost = sum(self.inference_time_device[: self.num_layers]) + raw_device_compute_cost = sum(self.inference_time_device[: self.num_layers]) + device_compute_cost = self._load_adjusted_device_cost(raw_device_compute_cost) network_cost = ( self.layers_sizes[self.num_layers - 1] / self.network_speed_ewma if self.num_layers > 0 @@ -190,6 +204,7 @@ def _record_device_only_candidate(self) -> None: self._record_adaptive_candidate( strategy="device_only", layer=self.num_layers - 1, + raw_device_compute_cost=raw_device_compute_cost, device_compute_cost=device_compute_cost, network_cost=network_cost, edge_compute_cost=0.0, @@ -203,6 +218,7 @@ def _record_adaptive_candidate( *, strategy: str, layer: int, + raw_device_compute_cost: float, device_compute_cost: float, network_cost: float, edge_compute_cost: float, @@ -228,6 +244,12 @@ def _record_adaptive_candidate( "risk_penalty": float(risk_penalty), "switch_penalty": float(switch_penalty), "network_speed_ewma": float(self.network_speed_ewma), + "device_cpu_percent": float(self.device_cpu_percent), + "device_load_factor": float(self.device_load_factor), + "raw_device_compute_cost": float(raw_device_compute_cost), + "device_load_penalty": float( + device_compute_cost - raw_device_compute_cost + ), "network_uncertainty": float(self.network_uncertainty), "device_uncertainty": float( self._max_cv("device", self._device_layers_for(strategy, layer)) @@ -262,6 +284,17 @@ def _risk_penalty( ) return base_latency * self.config.uncertainty_weight * uncertainty + @staticmethod + def _normalize_device_cpu_percent(value: float) -> float: + try: + cpu_percent = float(value or 0.0) + except (TypeError, ValueError): + return 0.0 + return min(100.0, max(0.0, cpu_percent)) + + def _load_adjusted_device_cost(self, raw_device_compute_cost: float) -> float: + return raw_device_compute_cost * self.device_load_factor + def _max_cv(self, section: str, layers: range) -> float: stats_by_layer = self.variance_stats.get(section, {}) max_cv = 0.0 diff --git a/src/server/offloading_algo/factory.py b/src/server/offloading_algo/factory.py index 0c13b67..7eeb1af 100644 --- a/src/server/offloading_algo/factory.py +++ b/src/server/offloading_algo/factory.py @@ -20,6 +20,7 @@ "hysteresis_margin_ms": 5, "min_speed_bytes_per_second": 1, "probe_probability": 0.05, + "device_load_weight": 0.50, }, } @@ -38,6 +39,7 @@ class OffloadingContext: inference_time_device: list[float] inference_time_edge: list[float] model_dir: str + device_cpu_percent: float = 0.0 adaptive_state: AdaptiveRiskState | None = None variance_stats: dict[str, Any] | None = None @@ -89,6 +91,7 @@ def create_offloading_algorithm( config=AdaptiveRiskConfig.from_mapping(offloading_config["adaptive_risk"]), state=state, variance_stats=context.variance_stats, + device_cpu_percent=context.device_cpu_percent, ) raise ValueError(f"Unsupported offloading algorithm: {algorithm_name}") diff --git a/src/server/settings.yaml b/src/server/settings.yaml index c08d759..f4fdbfe 100644 --- a/src/server/settings.yaml +++ b/src/server/settings.yaml @@ -63,6 +63,7 @@ offloading: hysteresis_margin_ms: 5 min_speed_bytes_per_second: 1 probe_probability: 0.05 + device_load_weight: 0.50 # Splitting degli output di valutazione (CSV, JSONL e inference_cycles_*.csv). # I CSV di valutazione vengono spezzati in piu' segmenti: se ne apre uno nuovo diff --git a/tests/fixtures/protocol/inference_v1.hex b/tests/fixtures/protocol/inference_v1.hex deleted file mode 100644 index e5005d7..0000000 --- a/tests/fixtures/protocol/inference_v1.hex +++ /dev/null @@ -1 +0,0 @@ -534349500100000000000000004a93400d000000676f6c64656e2d646576696365474f4c44020000000000003e080000000000c03f000000c0080000000ad7233c0ad7a33c diff --git a/tests/integration/test_http_protocol_validation.py b/tests/integration/test_http_protocol_validation.py index 9c44bc4..4088895 100644 --- a/tests/integration/test_http_protocol_validation.py +++ b/tests/integration/test_http_protocol_validation.py @@ -23,8 +23,17 @@ def _valid_payload(): def _oversized_payload(): payload = _valid_payload() - device_id_length = struct.unpack_from("