fallback to unicast when multicast unavailable

Author: Jepson2k

parol6/client/status_subscriber.py

@@ -13,26 +13,26 @@
 logger = logging.getLogger(__name__)
 
 
-class MulticastProtocol(asyncio.DatagramProtocol):
-    """Protocol handler for multicast UDP datagrams that works with uvloop."""
+class UDPProtocol(asyncio.DatagramProtocol):
+    """Protocol handler for UDP datagrams (multicast or unicast)."""
 
     def __init__(self, queue: asyncio.Queue):
         self.queue = queue
         self.transport = None
         self.receive_count = 0
         self.last_log_time = time.time()
 
-        # EMA rate tracking for multicast RX
+        # EMA rate tracking for RX
         self._rx_count = 0
         self._rx_last_time = time.monotonic()
-        self._rx_ema_period = 0.05  # Initialize with 20 Hz expected
+        self._rx_ema_period = 0.05  # Initialize with ~20 Hz expected
         self._rx_last_log_time = time.monotonic()
 
     def connection_made(self, transport):
         self.transport = transport
 
     def datagram_received(self, data, addr):
-        # Track multicast RX rate with EMA
+        # Track RX rate with EMA
         now = time.monotonic()
         if self._rx_count > 0:  # Skip first sample for period calculation
             period = now - self._rx_last_time
@@ -45,7 +45,7 @@ def datagram_received(self, data, addr):
         # Log rate every 3 seconds
         if now - self._rx_last_log_time >= 3.0 and self._rx_ema_period > 0:
             rx_hz = 1.0 / self._rx_ema_period
-            logger.debug(f"Multicast RX: {rx_hz:.1f} Hz (count={self._rx_count})")
+            logger.debug(f"Status RX: {rx_hz:.1f} Hz (count={self._rx_count})")
             self._rx_last_log_time = now
 
         try:
@@ -55,7 +55,7 @@ def datagram_received(self, data, addr):
             try:
                 self.queue.get_nowait()
                 self.queue.put_nowait((data, addr))
-            except:
+            except Exception:
                 pass
 
     def error_received(self, exc):
@@ -115,19 +115,35 @@ def _detect_primary_ip() -> str:
     return sock
 
 
+def _create_unicast_socket(port: int, host: str) -> socket.socket:
+    """Create and configure a plain UDP socket for unicast reception.
+
+    Binds to the provided host (default 127.0.0.1) and port with large RCVBUF.
+    """
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    try:
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
+    except Exception:
+        pass
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1 << 20)
+    try:
+        sock.bind((host, port))
+    except OSError:
+        # Fallback to wildcard
+        sock.bind(("", port))
+    sock.setblocking(False)
+    return sock
+
+
 async def subscribe_status(
     group: str | None = None, port: int | None = None, iface_ip: str | None = None
 ) -> AsyncIterator[StatusAggregate]:
     """
-    Async generator that yields decoded STATUS dicts from the UDP multicast broadcaster.
+    Async generator that yields decoded STATUS dicts from the UDP broadcaster.
 
     Uses create_datagram_endpoint for uvloop compatibility.
 
-    Usage:
-        async for status in subscribe_status():
-            # status is a dict with keys pose, angles, io, gripper (or None on parse failure)
-            ...
-
     Notes:
     - Uses loopback multicast by default (cfg.MCAST_* values).
     - Yields only messages that decode successfully via decode_status; otherwise skips.
@@ -136,19 +152,25 @@ async def subscribe_status(
     port = port or cfg.MCAST_PORT
     iface_ip = iface_ip or cfg.MCAST_IF
 
-    logger.info(f"subscribe_status starting: group={group}, port={port}, iface_ip={iface_ip}")
+    logger.info(
+        f"subscribe_status starting: transport={cfg.STATUS_TRANSPORT} group={group}, port={port}, iface_ip={iface_ip}"
+    )
 
     loop = asyncio.get_running_loop()
     queue = asyncio.Queue(maxsize=100)  # type: ignore
 
-    # Create the socket with multicast configuration
-    sock = _create_multicast_socket(group, port, iface_ip)
+    # Create the socket based on configured transport
+    if cfg.STATUS_TRANSPORT == "UNICAST":
+        sock = _create_unicast_socket(port, cfg.STATUS_UNICAST_HOST)
+    else:
+        # Multicast socket bound to ("", port) will also receive unicast datagrams to that port
+        sock = _create_multicast_socket(group, port, iface_ip)
 
     # Create the datagram endpoint with our protocol
     transport = None
     try:
         transport, _ = await loop.create_datagram_endpoint(
-            lambda: MulticastProtocol(queue), sock=sock
+            lambda: UDPProtocol(queue), sock=sock
         )
 
         while True:
@@ -162,7 +184,9 @@ async def subscribe_status(
                     yield parsed
 
             except TimeoutError:
-                logger.warning(f"No multicast received for 2s on {group}:{port} (iface={iface_ip})")
+                logger.warning(
+                    f"No status received for 2s on {('unicast' if cfg.STATUS_TRANSPORT=='UNICAST' else 'multicast')} {group}:{port} (iface={iface_ip})"
+                )
                 continue
 
     except asyncio.CancelledError:
@@ -178,5 +202,5 @@ async def subscribe_status(
             transport.close()
         try:
             sock.close()
-        except:
+        except Exception:
             pass

parol6/config.py

@@ -57,6 +57,11 @@ def _trace(self, msg, *args, **kwargs):
 MCAST_TTL: int = int(os.getenv("PAROL6_MCAST_TTL", "1"))
 MCAST_IF: str = os.getenv("PAROL6_MCAST_IF", "127.0.0.1")
 
+# Transport selection for status updates. Default MULTICAST; set to UNICAST on CI if multicast is not available.
+STATUS_TRANSPORT: str = os.getenv("PAROL6_STATUS_TRANSPORT", "MULTICAST").strip().upper()
+# Host to use for unicast fallback (defaults to loopback)
+STATUS_UNICAST_HOST: str = os.getenv("PAROL6_STATUS_UNICAST_HOST", "127.0.0.1")
+
 # Status update/broadcast rates
 STATUS_RATE_HZ: float = float(os.getenv("PAROL6_STATUS_RATE_HZ", "50"))
 STATUS_STALE_S: float = float(os.getenv("PAROL6_STATUS_STALE_S", "0.2"))

parol6/server/status_broadcast.py

@@ -14,13 +14,21 @@
 
 class StatusBroadcaster(threading.Thread):
     """
-    Broadcasts ASCII STATUS frames via UDP multicast.
+    Broadcasts ASCII STATUS frames via UDP.
 
-    Config:
+    Transport:
+      - cfg.STATUS_TRANSPORT: "MULTICAST" (default) or "UNICAST"
+
+    Multicast Config (used when STATUS_TRANSPORT == MULTICAST):
       - cfg.MCAST_GROUP (default "239.255.0.101")
-      - cfg.MCAST_PORT (default 50510)
-      - cfg.MCAST_TTL  (default 1)
-      - cfg.MCAST_IF   (default "127.0.0.1")
+      - cfg.MCAST_PORT  (default 50510)
+      - cfg.MCAST_TTL   (default 1)
+      - cfg.MCAST_IF    (default "127.0.0.1")
+
+    Unicast Config (used when STATUS_TRANSPORT == UNICAST):
+      - cfg.STATUS_UNICAST_HOST (default "127.0.0.1")
+
+    General:
       - cfg.STATUS_RATE_HZ (default 50)
       - cfg.STATUS_STALE_S (default 0.2) -> skip broadcast if cache is stale
     """
@@ -44,56 +52,103 @@ def __init__(
         self._period = 1.0 / max(rate_hz, 1.0)
         self._stale_s = stale_s
 
+        # Negotiated transport (can be forced via env or auto-fallback at runtime)
+        self._use_unicast: bool = (cfg.STATUS_TRANSPORT == "UNICAST")
+
         self._sock: socket.socket | None = None
         self._running = threading.Event()
         self._running.set()
 
-        # EMA rate tracking for multicast TX
+        # EMA rate tracking for TX
         self._tx_count = 0
         self._tx_last_time = time.monotonic()
         self._tx_ema_period = 1.0 / rate_hz  # Initialize with expected period
         self._tx_last_log_time = time.monotonic()  # For 3-second logging interval
 
+    def _detect_primary_ip(self) -> str:
+        tmp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        try:
+            tmp.connect(("1.1.1.1", 80))
+            return tmp.getsockname()[0]
+        except Exception:
+            return "127.0.0.1"
+        finally:
+            try:
+                tmp.close()
+            except Exception:
+                pass
+
     def _setup_socket(self) -> None:
+        # UNICAST: simple UDP socket without multicast options
+        if self._use_unicast:
+            sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1 << 20)
+            self._sock = sock
+            logger.info(
+                f"StatusBroadcaster (UNICAST) -> dest={cfg.STATUS_UNICAST_HOST}:{self.port}"
+            )
+            return
+
+        # MULTICAST: configure multicast TTL/IF with verification and fallback
         sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
         sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, self.ttl)
         sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, 1)
 
-        # Prefer loopback interface for multicast; if that fails, fall back to primary NIC IP
-        def _detect_primary_ip() -> str:
-            tmp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-            try:
-                tmp.connect(("1.1.1.1", 80))
-                return tmp.getsockname()[0]
-            except Exception:
-                return "127.0.0.1"
-            finally:
-                try:
-                    tmp.close()
-                except Exception:
-                    pass
-
         try:
             sock.setsockopt(
                 socket.IPPROTO_IP, socket.IP_MULTICAST_IF, socket.inet_aton(self.iface_ip)
             )
-        except Exception:
+            # Verify interface actually routes by sending a tiny dummy datagram
             try:
-                primary_ip = _detect_primary_ip()
+                sock.sendto(b"\0", (self.group, self.port))
+            except OSError as e:
+                raise RuntimeError(
+                    f"Initial multicast send failed on iface {self.iface_ip}: {e}"
+                )
+        except Exception as e:
+            logger.warning(
+                f"StatusBroadcaster: interface {self.iface_ip} failed verification: {e}"
+            )
+            try:
+                primary_ip = self._detect_primary_ip()
                 sock.setsockopt(
                     socket.IPPROTO_IP, socket.IP_MULTICAST_IF, socket.inet_aton(primary_ip)
                 )
                 logger.info(f"StatusBroadcaster: fallback IP_MULTICAST_IF to {primary_ip}")
-            except Exception as e:
-                logger.warning(f"StatusBroadcaster: failed to set IP_MULTICAST_IF: {e}")
+                # Verify fallback
+                sock.sendto(b"\0", (self.group, self.port))
+            except Exception as e2:
+                logger.warning(f"StatusBroadcaster: failed to set IP_MULTICAST_IF: {e2}")
+                # As a last resort, switch to UNICAST
+                try:
+                    sock.close()
+                except Exception:
+                    pass
+                self._use_unicast = True
+                usock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+                usock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1 << 20)
+                self._sock = usock
+                logger.info(
+                    f"StatusBroadcaster (UNICAST-FALLBACK) -> dest={cfg.STATUS_UNICAST_HOST}:{self.port}"
+                )
+                return
 
         sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1 << 20)
         self._sock = sock
+        logger.info(
+            f"StatusBroadcaster (MULTICAST) -> group={self.group} port={self.port} iface={self.iface_ip} ttl={self.ttl}"
+        )
 
     def run(self) -> None:
         self._setup_socket()
         cache = get_cache()
-        dest = (self.group, self.port)
+
+        # Destination based on negotiated transport
+        if self._use_unicast:
+            dest = (cfg.STATUS_UNICAST_HOST, self.port)
+        else:
+            dest = (self.group, self.port)
+
         sock = self._sock
         if sock is None:
             logger.error("StatusBroadcaster socket not initialized")
@@ -113,24 +168,30 @@ def run(self) -> None:
             # Skip broadcast if cache is stale (e.g., serial disconnected)
             if cache.age_s() <= self._stale_s:
                 payload = cache.to_ascii().encode("ascii", errors="ignore")
-                # memoryview avoids an extra copy in some implementations
-                sock.sendto(memoryview(payload), dest)
-
-                # Track multicast TX rate with EMA
-                now = time.monotonic()
-                if self._tx_count > 0:  # Skip first sample for period calculation
-                    period = now - self._tx_last_time
-                    if period > 0:
-                        # EMA update: 0.1 * new + 0.9 * old
-                        self._tx_ema_period = 0.1 * period + 0.9 * self._tx_ema_period
-                self._tx_last_time = now
-                self._tx_count += 1
-
-                # Log rate every 3 seconds
-                if now - self._tx_last_log_time >= 3.0 and self._tx_ema_period > 0:
-                    tx_hz = 1.0 / self._tx_ema_period
-                    logger.debug(f"Multicast TX: {tx_hz:.1f} Hz (count={self._tx_count})")
-                    self._tx_last_log_time = now
+                try:
+                    sock.sendto(memoryview(payload), dest)
+                except OSError as e:
+                    # Log occasionally to avoid flooding
+                    if time.monotonic() - self._tx_last_log_time >= 5.0:
+                        logger.warning(f"StatusBroadcaster send failed: {e}")
+                        self._tx_last_log_time = time.monotonic()
+                    # Do not stop thread; continue trying
+                else:
+                    # Track TX rate with EMA
+                    now = time.monotonic()
+                    if self._tx_count > 0:  # Skip first sample for period calculation
+                        period = now - self._tx_last_time
+                        if period > 0:
+                            # EMA update: 0.1 * new + 0.9 * old
+                            self._tx_ema_period = 0.1 * period + 0.9 * self._tx_ema_period
+                    self._tx_last_time = now
+                    self._tx_count += 1
+
+                    # Log rate every 3 seconds
+                    if now - self._tx_last_log_time >= 3.0 and self._tx_ema_period > 0:
+                        tx_hz = 1.0 / self._tx_ema_period
+                        logger.debug(f"Status TX: {tx_hz:.1f} Hz (count={self._tx_count})")
+                        self._tx_last_log_time = now
 
             # Sleep until next deadline (compensates for work time)
             sleep_time = next_deadline - time.monotonic()