diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 4b64e64..3896405 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -38,12 +38,38 @@ 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 + 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. 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 The server writes inference-cycle CSV output under diff --git a/docs/OFFLOADING_DECISION_EVENTS.md b/docs/OFFLOADING_DECISION_EVENTS.md index 1bca133..e53ed14 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,11 @@ 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`, `edge_uncertainty`, `device_cpu_percent`, +`device_load_factor`, `raw_device_compute_cost`, and `device_load_penalty`. + ## Example ```csv 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 604f536..4359bca 100644 --- a/docs/config/server.full.yaml +++ b/docs/config/server.full.yaml @@ -70,6 +70,17 @@ 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 + device_load_weight: 0.50 + delay_simulation: computation: enabled: false diff --git a/src/client/python/http_client.py b/src/client/python/http_client.py index fbfbdae..296dc3d 100644 --- a/src/client/python/http_client.py +++ b/src/client/python/http_client.py @@ -16,10 +16,14 @@ 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 +try: + import psutil +except ImportError: + psutil = None + try: from .delay_simulator import DelaySimulator from .profiler import AdvancedProfiler @@ -67,6 +71,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). @@ -84,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.""" @@ -171,7 +196,7 @@ def check_server_at_ip(ip, port, timeout=1.0): pass return None - except Exception as e: + except Exception: return None @@ -201,7 +226,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 = [] @@ -293,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")) @@ -341,7 +379,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 +400,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 +451,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 +729,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 +780,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 +853,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() @@ -784,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="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/sciot/config.py b/src/sciot/config.py index e347990..25a2503 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,102 @@ 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", + "device_load_weight", + } + 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" + ) + _optional_non_negative_number( + adaptive, + "device_load_weight", + errors, + path=f"{path}.adaptive_risk.device_load_weight", + ) + + def _model_reference( config: Mapping[str, Any], key: str, diff --git a/src/server/communication/inference_protocol.py b/src/server/communication/inference_protocol.py index 3cdfd7d..2da67f4 100644 --- a/src/server/communication/inference_protocol.py +++ b/src/server/communication/inference_protocol.py @@ -14,10 +14,9 @@ MAX_LAYER_TIMINGS_BYTES = 64 * 1024 MAX_REPORTED_LAYERS = 4096 INFERENCE_PROTOCOL_MAGIC = b"SCIP" -INFERENCE_PROTOCOL_VERSION = 1 +INFERENCE_PROTOCOL_VERSION = 2 INFERENCE_PROTOCOL_HEADER_FORMAT = "<4sHH" INFERENCE_PROTOCOL_HEADER_BYTES = struct.calcsize(INFERENCE_PROTOCOL_HEADER_FORMAT) -INFERENCE_PROTOCOL_LEGACY_VERSION = 0 class InvalidInferencePayload(ValueError): @@ -65,26 +64,25 @@ def decode_inference_payload(topic: str, payload: bytes) -> 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 37e3439..f8b4c75 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. @@ -180,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) @@ -189,9 +206,11 @@ class RequestHandler: csv_writer = None inference_counter = 0 header_printed = False + inference_table_rows_printed = 0 # Dictionary to store device profile data device_profiles = {} + offloading_states = {} num_layers = 0 def __init__(self): @@ -208,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() @@ -248,6 +263,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 +304,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 +390,73 @@ 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, + *, + 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( @@ -639,6 +721,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,27 +733,37 @@ 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" - ), + 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.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! - 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: @@ -676,12 +773,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 +790,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={ @@ -872,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 @@ -879,7 +972,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/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/src/server/offloading_algo/adaptive_risk.py b/src/server/offloading_algo/adaptive_risk.py new file mode 100644 index 0000000..8faefef --- /dev/null +++ b/src/server/offloading_algo/adaptive_risk.py @@ -0,0 +1,331 @@ +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 + device_load_weight: float = 0.50 + + @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)), + device_load_weight=float(value.get("device_load_weight", 0.50)), + ) + + +@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, + 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, + ) + 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, + raw_device_compute_cost=0.0, + 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: + 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 + ) + 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, + raw_device_compute_cost=raw_device_compute_cost, + 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: + 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 + 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, + raw_device_compute_cost=raw_device_compute_cost, + 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, + raw_device_compute_cost: float, + 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), + "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)) + ), + "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 + + @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 + 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..7eeb1af --- /dev/null +++ b/src/server/offloading_algo/factory.py @@ -0,0 +1,97 @@ +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, + "device_load_weight": 0.50, + }, +} + + +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 + device_cpu_percent: float = 0.0 + 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, + device_cpu_percent=context.device_cpu_percent, + ) + + 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..f4fdbfe 100644 --- a/src/server/settings.yaml +++ b/src/server/settings.yaml @@ -54,6 +54,17 @@ 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 + 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 # al primo criterio che scatta. I mirror JSONL delle decisioni usano un file per 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(" 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) + + +def test_device_cpu_load_scales_device_prefix_and_moves_split_earlier(): + config = { + "algorithm": "adaptive_risk", + "adaptive_risk": { + "hysteresis_margin_ms": 0.0, + "uncertainty_weight": 0.0, + "probe_probability": 0.0, + "device_load_weight": 2.0, + }, + } + timing_context = { + "inference_time_device": [0.01, 0.05, 0.40], + "inference_time_edge": [0.0, 0.08, 0.10], + } + + low_load_algorithm = create_offloading_algorithm( + config, + _context(**timing_context, device_cpu_percent=0.0), + ) + high_load_algorithm = create_offloading_algorithm( + config, + _context(**timing_context, device_cpu_percent=100.0), + ) + + assert low_load_algorithm.select_offloading_layer() == 1 + assert high_load_algorithm.select_offloading_layer() == 0 + + selected_candidate = next( + candidate + for candidate in high_load_algorithm.candidate_evaluations + if candidate["strategy"] == "split" + and candidate["offloading_layer_index"] == 0 + ) + assert selected_candidate["device_cpu_percent"] == 100.0 + assert selected_candidate["device_load_factor"] == pytest.approx(3.0) + assert selected_candidate["raw_device_compute_cost"] == pytest.approx(0.01) + assert selected_candidate["device_compute_cost"] == pytest.approx(0.03) + assert selected_candidate["device_load_penalty"] == pytest.approx(0.02) diff --git a/tests/unit/test_config_validation.py b/tests/unit/test_config_validation.py index bbd9651..a12e088 100644 --- a/tests/unit/test_config_validation.py +++ b/tests/unit/test_config_validation.py @@ -162,6 +162,36 @@ 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", + ), + ( + lambda cfg: cfg.__setitem__( + "offloading", {"adaptive_risk": {"device_load_weight": -0.1}} + ), + "offloading.adaptive_risk.device_load_weight: must be greater than or equal to 0", + ), ], ) def test_invalid_server_config_reports_field_errors(mutate, expected_error): @@ -172,6 +202,34 @@ 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, + "device_load_weight": 0.50, + }, + } + 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_inference_protocol.py b/tests/unit/test_inference_protocol.py index 2b08336..45f6fa6 100644 --- a/tests/unit/test_inference_protocol.py +++ b/tests/unit/test_inference_protocol.py @@ -4,7 +4,6 @@ import pytest from server.communication.inference_protocol import ( - INFERENCE_PROTOCOL_HEADER_BYTES, INFERENCE_PROTOCOL_MAGIC, INFERENCE_PROTOCOL_VERSION, InvalidInferencePayload, @@ -12,8 +11,8 @@ ) -def _legacy_payload(): - device_id = b"legacy-device" +def _payload_body(*, device_cpu_percent=42.5): + device_id = b"v2-device" output = np.array([1.0, 2.0], dtype="