Discovery

Author: bedogni@unimore.it

README.md

@@ -136,7 +136,7 @@ source .venv/bin/activate  # On macOS/Linux
 communication:
   http:
     host: 0.0.0.0
-    port: 8000
+    port: 23456
     endpoints:
       registration: /api/registration
       device_input: /api/device_input
@@ -168,7 +168,7 @@ client:
 
 http:
   server_host: "0.0.0.0"
-  server_port: 8000
+  server_port: 23456
 
 model:
   last_offloading_layer: 58

USER_GUIDE.md

@@ -69,7 +69,7 @@ communication:
 
   http:
     host: "0.0.0.0"
-    port: 8000
+    port: 23456
     model: "fomo_96x96"
     ntp_server: "pool.ntp.org"
     endpoints:
@@ -117,7 +117,7 @@ variance_detection:
 #### HTTP Client (`server_client_light/client/http_config.yaml`)
 ```yaml
 server:
-  base_url: "http://localhost:8000"
+  base_url: "http://localhost:23456"
   inference_endpoint: "/inference"
   timeout: 5
 
@@ -278,7 +278,7 @@ python install_tensorflow.py
 **2. Server connection refused**
 ```bash
 # Make sure server is running and port is correct
-netstat -an | grep 8000  # Check if port 8000 is open
+netstat -an | grep 23456  # Check if port 23456 is open
 ```
 
 **3. Model file not found**

pyproject.toml

@@ -33,6 +33,8 @@ dependencies = [
     "websockets",
     "tensorflow==2.15.0; sys_platform != 'darwin' or platform_machine != 'arm64'",
     "tensorflow-macos==2.15.0; sys_platform == 'darwin' and platform_machine == 'arm64'",
+    "ipykernel>=7.1.0",
+    "seaborn>=0.13.2",
 ]
 
 # Define optional dependencies like testing and CUDA support

server_client_light/client/http_client.py

@@ -9,6 +9,9 @@
 import yaml
 from pathlib import Path
 from delay_simulator import DelaySimulator
+import socket
+import ipaddress
+from concurrent.futures import ThreadPoolExecutor, as_completed
 
 # Get the directory where this script is located
 SCRIPT_DIR = Path(__file__).resolve().parent
@@ -22,6 +25,145 @@
 DEVICE_ID = config["client"]["device_id"]
 SERVER_HOST = config["http"]["server_host"]
 SERVER_PORT = config["http"]["server_port"]
+
+# Discovery functions
+def get_local_network():
+    """Get the local network IP range for scanning."""
+    try:
+        # Get local IP address
+        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        s.connect(("8.8.8.8", 80))
+        local_ip = s.getsockname()[0]
+        s.close()
+        
+        # Convert to network/24 (assuming typical home network)
+        network = ipaddress.IPv4Network(f"{local_ip}/24", strict=False)
+        return network, local_ip
+    except Exception as e:
+        print(f"Error getting local network: {e}")
+        return None, None
+
+def check_server_at_ip(ip, port, timeout=1.0):
+    """Check if a server is responding at the given IP:port."""
+    try:
+        # First try TCP connection
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        sock.settimeout(timeout)
+        result = sock.connect_ex((str(ip), port))
+        sock.close()
+        
+        if result == 0:
+            # Port is open, verify it's our server by trying multiple methods
+            try:
+                # Try 1: POST to registration endpoint (proper way)
+                test_url = f"http://{ip}:{port}{config['http']['endpoints']['registration']}"
+                response = requests.post(test_url, json={"device_id": "discovery_test"}, timeout=2)
+                if response.status_code in [200, 201, 400, 409]:  # Any reasonable server response
+                    print(f"  Found server at {ip}:{port} (registration: {response.status_code})")
+                    return str(ip)
+            except requests.exceptions.RequestException:
+                pass
+            
+            try:
+                # Try 2: GET request to root or any endpoint
+                response = requests.get(f"http://{ip}:{port}/", timeout=2)
+                if response.status_code < 500:  # Any response except server error
+                    print(f"  Found server at {ip}:{port} (root: {response.status_code})")
+                    return str(ip)
+            except requests.exceptions.RequestException:
+                pass
+            
+            try:
+                # Try 3: Check offloading_layer endpoint
+                test_url = f"http://{ip}:{port}{config['http']['endpoints']['offloading_layer']}"
+                response = requests.get(test_url, timeout=2)
+                if response.status_code in [200, 400, 404, 405]:
+                    print(f"  Found server at {ip}:{port} (offload endpoint: {response.status_code})")
+                    return str(ip)
+            except requests.exceptions.RequestException:
+                pass
+                
+        return None
+    except Exception as e:
+        return None
+
+def discover_server(port):
+    """Scan local network to discover server."""
+    print("="*60)
+    print("SERVER DISCOVERY MODE")
+    print("="*60)
+    
+    # First, try common local addresses
+    print(f"Checking localhost and common addresses on port {port}...\n")
+    common_hosts = ["127.0.0.1", "localhost", "0.0.0.0"]
+    
+    for host in common_hosts:
+        print(f"  Trying {host}:{port}...")
+        result = check_server_at_ip(host, port, timeout=2.0)
+        if result:
+            print(f"\nServer found at {result}:{port}")
+            return result
+    
+    print("\nNo server on localhost, scanning local network...\n")
+    
+    network, local_ip = get_local_network()
+    if not network:
+        print("ERROR: Could not determine local network")
+        return None
+    
+    print(f"Local IP: {local_ip}")
+    print(f"Scanning network: {network}")
+    print(f"This may take 10-30 seconds...\n")
+    
+    # Scan network with threading for speed
+    found_servers = []
+    total_hosts = network.num_addresses - 2  # Exclude network and broadcast
+    checked = 0
+    
+    with ThreadPoolExecutor(max_workers=50) as executor:
+        futures = {executor.submit(check_server_at_ip, ip, port): ip 
+                  for ip in network.hosts() if str(ip) != local_ip}
+        
+        for future in as_completed(futures):
+            checked += 1
+            if checked % 25 == 0:
+                print(f"Progress: {checked}/{total_hosts} hosts checked...")
+            
+            result = future.result()
+            if result:
+                found_servers.append(result)
+                print(f"\nSERVER FOUND: {result}:{port}")
+                # Cancel remaining futures for faster completion
+                for f in futures:
+                    f.cancel()
+                break
+    
+    print(f"\nScan complete: {checked}/{total_hosts} hosts checked")
+    
+    if found_servers:
+        selected = found_servers[0]
+        print(f"\nSelected server: {selected}:{port}")
+        return selected
+    else:
+        print("\nNo server found on local network")
+        print("   Please check:")
+        print("   1. Server is running")
+        print("   2. Server is on the same network")
+        print(f"   3. Server is listening on port {port}")
+        return None
+
+# Run discovery if SERVER_HOST is None
+if SERVER_HOST is None or SERVER_HOST == "None" or SERVER_HOST == "":
+    print("\nSERVER_HOST is not configured - starting discovery...\n")
+    discovered_host = discover_server(SERVER_PORT)
+    
+    if discovered_host:
+        SERVER_HOST = discovered_host
+        print(f"\nUsing discovered server: {SERVER_HOST}:{SERVER_PORT}\n")
+    else:
+        print("\nWARNING: No server discovered - will run in LOCAL-ONLY mode")
+        SERVER_HOST = "localhost"  # Fallback to avoid errors
+
 SERVER = f"http://{SERVER_HOST}:{SERVER_PORT}"
 
 MODEL_CONFIG = config["model"]

server_client_light/client/http_config.yaml

@@ -3,10 +3,7 @@ client:
   device_id: device_01
 delay_simulation:
   device_computation:
-    enabled: true
-    mean: 0.015
-    std_dev: 0.003
-    type: gaussian
+    enabled: false
   network:
     enabled: false
 http:
@@ -15,8 +12,8 @@ http:
     device_input: /api/device_input
     offloading_layer: /api/offloading_layer
     registration: /api/registration
-  server_host: 0.0.0.0
-  server_port: 8000
+  server_host: None
+  server_port: 23456
 local_inference_mode:
   enabled: false
   probability: 0.0

server_client_light/client/http_config.yaml.backup

@@ -50,7 +50,7 @@
         "device_computation_delay": {"enabled": False},
         "edge_computation_delay": {"enabled": False},
         "network_delay": {"enabled": False},
-        "duration_seconds": 600,
+        "duration_seconds": 180,
         "num_clients": 1
     },
 
@@ -65,7 +65,7 @@
             "std_dev": 0.002  # Low variance for stable high delay
         },
         "network_delay": {"enabled": False},
-        "duration_seconds": 600,
+        "duration_seconds": 180,
         "num_clients": 1
     },
 
@@ -80,7 +80,7 @@
         },
         "edge_computation_delay": {"enabled": False},
         "network_delay": {"enabled": False},
-        "duration_seconds": 600,
+        "duration_seconds": 180,
         "num_clients": 1
     },
 
@@ -95,7 +95,7 @@
             "mean": 0.050,  # 50ms average
             "std_dev": 0.025  # High variance for variable conditions
         },
-        "duration_seconds": 600,
+        "duration_seconds": 180,
         "num_clients": 1
     },
 
@@ -120,7 +120,7 @@
             "mean": 0.040,  # 40ms average
             "std_dev": 0.020  # High variance
         },
-        "duration_seconds": 600,
+        "duration_seconds": 180,
         "num_clients": 1
     },
 ]
@@ -209,6 +209,7 @@ def start_server(self):
         """Start the edge server"""
         print("\nStarting server...")
         server_script = SERVER_DIR / "run_edge.py"
+        print(f"  Launching server script: {PYTHON_EXECUTABLE}{server_script}")
 
         self.server_process = subprocess.Popen(
             [PYTHON_EXECUTABLE, str(server_script)],
@@ -228,7 +229,7 @@ def print_server_output():
         server_output_thread.start()
 
         # Wait for server to be ready
-        max_wait = 15
+        max_wait = 30
         for i in range(max_wait):
             try:
                 response = requests.get(f"http://{self.server_host}:{self.server_port}/docs", timeout=1)

simulation_runner.py

@@ -4,7 +4,7 @@
 server:
   http:
     host: "0.0.0.0"
-    port: 8000
+    port: 23456
     ntp_server: "0.it.pool.ntp.org"
 
   websocket:

src/server/client_config.yaml

@@ -124,6 +124,19 @@ def run_inference(offloading_layer_index: int, offloading_layer_output: np.array
 
             prediction_data = []
             if layer_index == start_layer_index:  # if it's the starting layer
+                # Check if this layer has multiple inputs (skip connection)
+                if hasattr(layer, '_inbound_nodes') and layer._inbound_nodes:
+                    inbound_node = layer._inbound_nodes[0]
+                    # Check if this layer expects multiple inputs
+                    if isinstance(inbound_node.inbound_layers, list) and len(inbound_node.inbound_layers) > 1:
+                        # This layer has skip connections - we can't process it because we don't have
+                        # all the required inputs from previous layers computed on device
+                        from server.logger.log import logger
+                        logger.warning(f"Cannot start edge processing at layer {layer_index}: layer {layer.name} has skip connections requiring {len(inbound_node.inbound_layers)} inputs from device-side layers")
+                        logger.warning(f"Offloading point {offloading_layer_index} is before a skip connection merge. Returning device output.")
+                        # Return the input as-is since we can't process further
+                        return input_data, []
+                
                 prediction_data.append(input_data)
             else:
                 # Get the previous layers' output tensor
@@ -158,6 +171,10 @@ def run_inference(offloading_layer_index: int, offloading_layer_output: np.array
                 if not has_all_inputs:
                     break
 
+            # Check again if we should stop (before calling predict)
+            if edge_processing_stopped:
+                break
+
             # Measure inference time for this layer
             t0 = time.time()
             prediction = model_manager.predict_single_layer(layer_index, start_layer_offset, prediction_data)

src/server/edge/edge_initialization.py

@@ -181,6 +181,18 @@ def predict_single_layer(self, layer_id: int, layer_offset: int, layer_input_dat
             output_details = cached['output_details']
 
         # set input tensor
+        # Handle both single input and multiple inputs (e.g., from skip connections)
+        if not isinstance(layer_input_data, list):
+            # If layer_input_data is not a list, wrap it in one
+            layer_input_data = [layer_input_data]
+        
+        # Verify we have the correct number of inputs
+        if len(layer_input_data) != len(input_details):
+            logger.error(f"Input mismatch: model expects {len(input_details)} inputs but received {len(layer_input_data)}")
+            logger.error(f"Input shapes expected: {[detail['shape'] for detail in input_details]}")
+            logger.error(f"Input data shapes received: {[data.shape if hasattr(data, 'shape') else type(data) for data in layer_input_data]}")
+            raise ValueError(f"Model expects {len(input_details)} inputs but received {len(layer_input_data)}")
+        
         for i, input_detail in enumerate(input_details):
             interpreter.set_tensor(input_detail['index'], layer_input_data[i].reshape(input_detail['shape']))