Repository: quartiq/stabilizer (language: rust)

Issue:
stream_target in flash crashes due to DHCP not yet complete
The local IP isn't set yet but it already tries opening the stream and crashes.

```
WARN - Network link DOWN                                                                                                                                                                             
INFO - Opening stream                                                                             
panicked at /home/rj/.cargo/git/checkouts/smoltcp-nal-f2ec3ae73a86846c/3e3373e/src/lib.rs:569:14:                                                                                                    
called `Option::unwrap()` on a `None` value
```

https://github.com/quartiq/smoltcp-nal/blob/3e3373e/src/lib.rs#L569


Discussion / hints:
I think:

* let's make smoltcp-nal not crash here but Err instead
* and then let's make sure that the stream udp socket gets reset/reconnected on DHCP IP change the same way the MQTT stuff gets reset (IIRC that works nicely).
> * and then let's make sure that the stream udp socket gets reset/reconnected on DHCP IP change the same way the MQTT stuff gets reset (IIRC that works nicely).

The socket closure on address update is handled at the smoltcp-nal layer:
https://github.com/quartiq/smoltcp-nal/blob/1f7338059538848346051f8b498f897962394b0a/src/lib.rs#L254-L261

We'll need to add logic to Stabilizer to reconnect though, I'll open a PR.

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3abfe7f4d..feebf7fe6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -24,6 +24,7 @@ console.
 * MSRV removed. Stabilizer uses latest stable rust.
 * The IIR (biquad) filter used for PID action has changed its serialization format.
   See also the `iir_coefficients` Python CLI implementation.
+* The stream target is now configures as a `1.2.3.4:4321` string
 
 ### Fixed
 * Fixed an issue where the device would sometimes not enumerate on Windows
diff --git a/Cargo.lock b/Cargo.lock
index 7bb0d3fd4..47dd0fc5d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1175,6 +1175,29 @@ dependencies = [
  "syn 2.0.66",
 ]
 
+[[package]]
+name = "serde_with"
+version = "3.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0ad483d2ab0149d5a5ebcd9972a3852711e0153d863bf5a5d0391d28883c4a20"
+dependencies = [
+ "serde",
+ "serde_derive",
+ "serde_with_macros",
+]
+
+[[package]]
+name = "serde_with_macros"
+version = "3.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "65569b702f41443e8bc8bbb1c5779bd0450bbe723b56198980e80ec45780bce2"
+dependencies = [
+ "darling",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.66",
+]
+
 [[package]]
 name = "serial-settings"
 version = "0.1.0"
@@ -1239,9 +1262,9 @@ dependencies = [
 
 [[package]]
 name = "smoltcp-nal"
-version = "0.5.0"
+version = "0.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c3ac46812539eefac6ffe76393317b6dd4851fef6f31621cce27e9924d586570"
+checksum = "975f15c4c77e7798c236b542032891f0cafe5f2e103debe59d069d50a645ed8a"
 dependencies = [
  "embedded-nal",
  "embedded-time",
@@ -1299,6 +1322,7 @@ dependencies = [
  "sequential-storage",
  "serde",
  "serde-json-core",
+ "serde_with",
  "serial-settings",
  "shared-bus",
  "smoltcp-nal",
diff --git a/Cargo.toml b/Cargo.toml
index 98d8785d7..803eeef7b 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -74,6 +74,7 @@ smoltcp-nal = { version = "0.5", features = ["shared-stack"] }
 postcard = "1"
 bit_field = "0.10.2"
 embassy-futures = { version = "0.1", default-features = false }
+serde_with = { version = "3.8", default-features = false, features = ["macros"] }
 
 [build-dependencies]
 built = { version = "0.7", features = ["git2"], default-features = false }
diff --git a/hitl/streaming.py b/hitl/streaming.py
index 988cd33ff..d44a6116b 100644
--- a/hitl/streaming.py
+++ b/hitl/streaming.py
@@ -42,16 +42,12 @@ async def _main():
 
     conf = await miniconf.Miniconf.create(args.broker, prefix)
 
-    stream_target = [int(x) for x in args.ip.split('.')]
     if ipaddress.ip_address(args.ip).is_unspecified:
-        stream_target = get_local_ip(args.broker)
+        args.ip = get_local_ip(args.broker)
 
     logger.info("Starting stream")
     await conf.set(
-        "/stream_target", {
-            "ip": stream_target,
-            "port": args.port
-        }, retain=False)
+        "/stream_target", f"{args.ip}:{args.port}", retain=False)
 
     try:
         logger.info("Testing stream reception")
@@ -64,7 +60,7 @@ async def _main():
     finally:
         logger.info("Stopping stream")
         await conf.set(
-            "/stream_target", {"ip": [0, 0, 0, 0], "port": 0}, retain=False)
+            "/stream_target", "0.0.0.0:0", retain=False)
 
     logger.info("Draining queue")
     await asyncio.sleep(.1)
diff --git a/py/stabilizer/stream.py b/py/stabilizer/stream.py
index 4452c9c34..66d423729 100644
--- a/py/stabilizer/stream.py
+++ b/py/stabilizer/stream.py
@@ -30,10 +30,9 @@ def get_local_ip(remote):
     sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
     try:
         sock.connect((remote, 1883))
-        address = sock.getsockname()[0]
+        return sock.getsockname()[0]
     finally:
         sock.close()
-    return list(map(int, address.split(".")))
 
 
 class AdcDac:
diff --git a/src/bin/dual-iir.rs b/src/bin/dual-iir.rs
index a4a04be88..c5724a8dd 100644
--- a/src/bin/dual-iir.rs
+++ b/src/bin/dual-iir.rs
@@ -483,8 +483,9 @@ mod app {
                 }
             }
 
-            let target = settings.dual_iir.stream_target.into();
-            c.shared.network.lock(|net| net.direct_stream(target));
+            c.shared
+                .network
+                .lock(|net| net.direct_stream(settings.dual_iir.stream_target));
 
             c.shared
                 .active_settings
diff --git a/src/bin/lockin.rs b/src/bin/lockin.rs
index e0cfb58fe..f317a2116 100644
--- a/src/bin/lockin.rs
+++ b/src/bin/lockin.rs
@@ -521,8 +521,9 @@ mod app {
             c.local.afes.0.set_gain(settings.lockin.afe[0]);
             c.local.afes.1.set_gain(settings.lockin.afe[1]);
 
-            let target = settings.lockin.stream_target.into();
-            c.shared.network.lock(|net| net.direct_stream(target));
+            c.shared
+                .network
+                .lock(|net| net.direct_stream(settings.lockin.stream_target));
 
             c.shared
                 .active_settings
diff --git a/src/net/data_stream.rs b/src/net/data_stream.rs
index f8c1f1660..9f1d4dd12 100644
--- a/src/net/data_stream.rs
+++ b/src/net/data_stream.rs
@@ -22,14 +22,16 @@
 //! # Example
 //! A sample Python script is available in `scripts/stream_throughput.py` to demonstrate reception
 //! of livestreamed data.
-use core::mem::MaybeUninit;
+use core::{fmt::Write, mem::MaybeUninit};
 use heapless::{
     pool::{Box, Init, Pool, Uninit},
     spsc::{Consumer, Producer, Queue},
+    String,
 };
 use num_enum::IntoPrimitive;
-use serde::{Deserialize, Serialize};
-use smoltcp_nal::embedded_nal::{IpAddr, Ipv4Addr, SocketAddr, UdpClientStack};
+use serde::Serialize;
+use serde_with::DeserializeFromStr;
+use smoltcp_nal::embedded_nal::{nb, SocketAddr, UdpClientStack};
 
 use super::NetworkReference;
 
@@ -58,17 +60,47 @@ type Frame = [MaybeUninit<u8>; FRAME_SIZE];
 /// Represents the destination for the UDP stream to send data to.
 ///
 /// # Miniconf
-/// `{"ip": <addr>, "port": <port>}`
+/// `<addr>:<port>`
 ///
-/// * `<addr>` is an array of 4 bytes. E.g. `[192, 168, 0, 1]`
+/// * `<addr>` is an IPv4 address. E.g. `192.168.0.1`
 /// * `<port>` is any unsigned 16-bit value.
 ///
 /// ## Example
-/// `{"ip": [192, 168,0, 1], "port": 1111}`
-#[derive(Copy, Clone, Debug, Serialize, Deserialize, Default)]
-pub struct StreamTarget {
-    pub ip: [u8; 4],
-    pub port: u16,
+/// `192.168.0.1:1234`
+#[derive(Copy, Clone, Debug, DeserializeFromStr, PartialEq, Eq)]
+pub struct StreamTarget(pub SocketAddr);
+
+impl Default for StreamTarget {
+    fn default() -> Self {
+        Self("0.0.0.0:0".parse().unwrap())
+    }
+}
+
+impl Serialize for StreamTarget {
+    fn serialize<S: serde::Serializer>(
+        &self,
+        serializer: S,
+    ) -> Result<S::Ok, S::Error> {
+        let mut display: String<30> = String::new();
+        write!(&mut display, "{}", self.0).unwrap();
+        serializer.serialize_str(&display)
+    }
+}
+
+impl core::str::FromStr for StreamTarget {
+    type Err = &'static str;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let addr = SocketAddr::from_str(s)
+            .map_err(|_| "Invalid socket address format")?;
+        Ok(Self(addr))
+    }
+}
+
+impl core::fmt::Display for StreamTarget {
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        self.0.fmt(f)
+    }
 }
 
 /// Specifies the format of streamed data
@@ -92,20 +124,6 @@ pub enum StreamFormat {
     Fls = 2,
 }
 
-impl From<StreamTarget> for SocketAddr {
-    fn from(target: StreamTarget) -> SocketAddr {
-        SocketAddr::new(
-            IpAddr::V4(Ipv4Addr::new(
-                target.ip[0],
-                target.ip[1],
-                target.ip[2],
-                target.ip[3],
-            )),
-            target.port,
-        )
-    }
-}
-
 /// Configure streaming on a device.
 ///
 /// # Args
@@ -273,7 +291,7 @@ pub struct DataStream {
     socket: Option<<NetworkReference as UdpClientStack>::UdpSocket>,
     queue: Consumer<'static, StreamFrame, FRAME_QUEUE_SIZE>,
     frame_pool: &'static Pool<Frame>,
-    remote: SocketAddr,
+    remote: StreamTarget,
 }
 
 impl DataStream {
@@ -291,7 +309,7 @@ impl DataStream {
         Self {
             stack,
             socket: None,
-            remote: StreamTarget::default().into(),
+            remote: StreamTarget::default(),
             queue: consumer,
             frame_pool,
         }
@@ -309,20 +327,22 @@ impl DataStream {
     fn open(&mut self) -> Result<(), ()> {
         // If there is already a socket of if remote address is unspecified,
         // do not open a new socket.
-        if self.socket.is_some() || self.remote.ip().is_unspecified() {
+        if self.socket.is_some() || self.remote.0.ip().is_unspecified() {
             return Err(());
         }
 
-        log::info!("Opening stream");
-
         let mut socket = self.stack.socket().or(Err(()))?;
 
-        // Note(unwrap): We only connect with a new socket, so it is guaranteed to not already be
-        // bound.
-        self.stack.connect(&mut socket, self.remote).unwrap();
+        // We may fail to connect if we don't have an IP address yet.
+        if self.stack.connect(&mut socket, self.remote.0).is_err() {
+            self.stack.close(socket).unwrap();
+            return Err(());
+        }
 
         self.socket.replace(socket);
 
+        log::info!("Opening stream");
+
         Ok(())
     }
 
@@ -330,7 +350,7 @@ impl DataStream {
     ///
     /// # Args
     /// * `remote` - The destination to send stream data to.
-    pub fn set_remote(&mut self, remote: SocketAddr) {
+    pub fn set_remote(&mut self, remote: StreamTarget) {
         // Close socket to be reopened if the remote has changed.
         if remote != self.remote {
             self.close();
@@ -360,7 +380,27 @@ impl DataStream {
                             core::mem::size_of_val(buf),
                         )
                     };
-                    self.stack.send(handle, data).ok();
+
+                    // If we fail to send, it can only be because the socket got closed on us (i.e.
+                    // address update due to DHCP). If this happens, reopen the socket.
+                    match self.stack.send(handle, data) {
+                        Ok(_) => {},
+
+                        // Our IP address may have changedm so handle reopening the UDP stream.
+                        Err(nb::Error::Other(smoltcp_nal::NetworkError::UdpWriteFailure(smoltcp_nal::smoltcp::socket::udp::SendError::Unaddressable))) => {
+                            log::warn!( "IP address updated during stream. Reopening socket");
+                            let socket = self.socket.take().unwrap();
+                            self.stack.close(socket).unwrap();
+                        }
+
+                        // The buffer should clear up once ICMP resolves the IP address, so ignore
+                        // this error.
+                        Err(nb::Error::Other(smoltcp_nal::NetworkError::UdpWriteFailure(smoltcp_nal::smoltcp::socket::udp::SendError::BufferFull))) => {}
+
+                        Err(other) => {
+                            log::warn!("Unexpected UDP error during data stream: {other:?}");
+                        }
+                    }
                     self.frame_pool.free(frame.buffer)
                 }
             }
diff --git a/src/net/mod.rs b/src/net/mod.rs
index a10b7cdb6..b0123d4eb 100644
--- a/src/net/mod.rs
+++ b/src/net/mod.rs
@@ -18,7 +18,7 @@ use crate::hardware::{
     SystemTimer,
 };
 use crate::settings::NetSettings;
-use data_stream::{DataStream, FrameGenerator};
+use data_stream::{DataStream, FrameGenerator, StreamTarget};
 use network_processor::NetworkProcessor;
 use telemetry::TelemetryClient;
 
@@ -26,7 +26,6 @@ use core::fmt::Write;
 use heapless::String;
 use miniconf::JsonCoreSlash;
 use miniconf_mqtt::minimq;
-use smoltcp_nal::embedded_nal::SocketAddr;
 
 pub type NetworkReference =
     smoltcp_nal::shared::NetworkStackProxy<'static, NetworkStack>;
@@ -174,7 +173,7 @@ where
     ///
     /// # Args
     /// * `remote` - The destination for the streamed data.
-    pub fn direct_stream(&mut self, remote: SocketAddr) {
+    pub fn direct_stream(&mut self, remote: StreamTarget) {
         if self.generator.is_none() {
             self.stream.set_remote(remote);
         }


Test patch: