Add UDP broadcast simulation

Author: manifest

examples/udp_vpv4_broadcast/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "udp_ipv4_broadcast"
+version = "0.1.0"
+edition = "2024"
+publish = false
+
+[dependencies]
+tokio = "1"
+turmoil = { path = "../.." }
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }

examples/udp_vpv4_broadcast/src/main.rs

@@ -0,0 +1,55 @@
+use std::{net::Ipv4Addr, time::Duration};
+use tracing::info;
+use turmoil::{IpVersion, net::UdpSocket};
+
+const N_STEPS: usize = 3;
+
+fn main() {
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::builder()
+                .with_default_directive(tracing::level_filters::LevelFilter::INFO.into())
+                .from_env_lossy(),
+        )
+        .init();
+
+    let tick = Duration::from_millis(100);
+    let mut sim = turmoil::Builder::new()
+        .tick_duration(tick)
+        .ip_version(IpVersion::V4)
+        .build();
+
+    let broadcast_port = 9000;
+
+    for server_index in 0..2 {
+        sim.client(format!("server-{server_index}"), async move {
+            let socket = UdpSocket::bind((Ipv4Addr::UNSPECIFIED, broadcast_port)).await?;
+
+            let mut buf = [0; 1024];
+            for _ in 0..N_STEPS {
+                let (n, addr) = socket.recv_from(&mut buf).await?;
+                let data = &buf[0..n];
+
+                info!("UDP packet from {} has been received: {:?}", addr, data);
+            }
+            Ok(())
+        });
+    }
+
+    sim.client("client", async move {
+        let dst = (Ipv4Addr::BROADCAST, broadcast_port);
+        let socket = UdpSocket::bind((Ipv4Addr::UNSPECIFIED, 0)).await?;
+        socket.set_broadcast(true)?;
+
+        for _ in 0..N_STEPS {
+            let _ = socket.send_to(&[1, 2, 3], dst).await?;
+            info!("UDP packet has been sent");
+
+            tokio::time::sleep(tick).await;
+        }
+
+        Ok(())
+    });
+
+    sim.run().unwrap();
+}

src/host.rs

@@ -13,6 +13,7 @@ use std::sync::Arc;
 use tokio::sync::{mpsc, Notify};
 use tokio::time::{Duration, Instant};
 
+const DEFAULT_BROADCAST: bool = true;
 const DEFAULT_MULTICAST_LOOP: bool = true;
 
 /// A host in the simulated network.
@@ -165,6 +166,7 @@ pub(crate) struct Udp {
 
 struct UdpBind {
     bind_addr: SocketAddr,
+    broadcast: bool,
     multicast_loop: bool,
     queue: mpsc::Sender<(Datagram, SocketAddr)>,
 }
@@ -177,17 +179,30 @@ impl Udp {
         }
     }
 
-    fn is_port_assigned(&self, port: u16) -> bool {
+    pub(crate) fn is_port_assigned(&self, port: u16) -> bool {
         self.binds.keys().any(|p| *p == port)
     }
 
+    pub(crate) fn is_broadcast_enabled(&self, port: u16) -> bool {
+        self.binds
+            .get(&port)
+            .map(|bind| bind.broadcast)
+            .unwrap_or(DEFAULT_BROADCAST)
+    }
+
     pub(crate) fn is_multicast_loop_enabled(&self, port: u16) -> bool {
         self.binds
             .get(&port)
             .map(|bind| bind.multicast_loop)
             .unwrap_or(DEFAULT_MULTICAST_LOOP)
     }
 
+    pub(crate) fn set_broadcast(&mut self, port: u16, on: bool) {
+        self.binds
+            .entry(port)
+            .and_modify(|bind| bind.broadcast = on);
+    }
+
     pub(crate) fn set_multicast_loop(&mut self, port: u16, on: bool) {
         self.binds
             .entry(port)
@@ -198,6 +213,7 @@ impl Udp {
         let (tx, rx) = mpsc::channel(self.capacity);
         let bind = UdpBind {
             bind_addr: addr,
+            broadcast: DEFAULT_BROADCAST,
             multicast_loop: DEFAULT_MULTICAST_LOOP,
             queue: tx,
         };

src/net/udp.rs

@@ -344,6 +344,29 @@ impl UdpSocket {
         }
 
         match dst {
+            SocketAddr::V4(dst) if dst.ip().is_broadcast() => {
+                let host = world.current_host();
+                match host.udp.is_broadcast_enabled(src.port()) {
+                    true => world
+                        .hosts
+                        .iter()
+                        .filter(|(_, host)| host.udp.is_port_assigned(dst.port()))
+                        .map(|(addr, _)| SocketAddr::new(*addr, dst.port()))
+                        .collect::<Vec<_>>()
+                        .into_iter()
+                        .try_for_each(|dst| match dst {
+                            dst if src.ip() == dst.ip() => {
+                                send_loopback(src, dst, Protocol::Udp(packet.clone()));
+                                Ok(())
+                            }
+                            dst => world.send_message(src, dst, Protocol::Udp(packet.clone())),
+                        }),
+                    false => Err(Error::new(
+                        ErrorKind::PermissionDenied,
+                        "Broadcast is not enabled",
+                    )),
+                }
+            }
             dst if dst.ip().is_multicast() => world
                 .multicast_groups
                 .destination_addresses(dst)
@@ -357,12 +380,38 @@ impl UdpSocket {
                         Ok(())
                     }
                     dst => world.send_message(src, dst, Protocol::Udp(packet.clone())),
-                })?,
-            dst if is_same(src, dst) => send_loopback(src, dst, Protocol::Udp(packet)),
-            _ => world.send_message(src, dst, Protocol::Udp(packet))?,
+                }),
+            dst if is_same(src, dst) => {
+                send_loopback(src, dst, Protocol::Udp(packet));
+                Ok(())
+            }
+            _ => world.send_message(src, dst, Protocol::Udp(packet)),
         }
+    }
 
-        Ok(())
+    /// Gets the value of the `SO_BROADCAST` option for this socket.
+    ///
+    /// For more information about this option, see [`set_broadcast`].
+    ///
+    /// [`set_broadcast`]: method@Self::set_broadcast
+    pub fn broadcast(&self) -> io::Result<bool> {
+        let local_port = self.local_addr.port();
+        World::current(|world| Ok(world.current_host().udp.is_broadcast_enabled(local_port)))
+    }
+
+    /// Sets the value of the `SO_BROADCAST` option for this socket.
+    ///
+    /// When enabled, this socket is allowed to send packets to a broadcast
+    /// address.
+    pub fn set_broadcast(&self, on: bool) -> io::Result<()> {
+        let local_port = match self.local_addr {
+            SocketAddr::V4(addr) => addr.port(),
+            _ => return Ok(()),
+        };
+        World::current(|world| {
+            world.current_host_mut().udp.set_broadcast(local_port, on);
+            Ok(())
+        })
     }
 
     /// Gets the value of the `IP_MULTICAST_LOOP` option for this socket.

tests/udp.rs

@@ -95,6 +95,50 @@ fn try_recv_pong(sock: &net::UdpSocket) -> Result<()> {
     Ok(())
 }
 
+#[test]
+fn udp_ipv4_broadcast() -> Result {
+    let mut sim = Builder::new()
+        .tick_duration(Duration::from_millis(50))
+        .ip_version(IpVersion::V4)
+        .build();
+
+    let non_broadcast_port = 8000;
+    let broadcast_port = 9000;
+    sim.client("server-non-broadcast-port", async move {
+        let socket = UdpSocket::bind((Ipv4Addr::UNSPECIFIED, non_broadcast_port)).await?;
+
+        let mut buf = [0; 1];
+        let is_timed_out = timeout(Duration::from_secs(1), socket.recv_from(&mut buf))
+            .await
+            .is_err();
+        assert!(is_timed_out);
+
+        Ok(())
+    });
+    for server_index in 0..3 {
+        sim.client(format!("server-{server_index}"), async move {
+            let socket = UdpSocket::bind((Ipv4Addr::UNSPECIFIED, broadcast_port)).await?;
+
+            let mut buf = [0; 1];
+            socket.recv_from(&mut buf).await?;
+            assert_eq!([1], buf);
+
+            Ok(())
+        });
+    }
+    sim.client("client", async move {
+        let dst = (Ipv4Addr::BROADCAST, broadcast_port);
+        let socket = UdpSocket::bind((Ipv4Addr::UNSPECIFIED, 0)).await?;
+        socket.set_broadcast(true)?;
+
+        let _ = socket.send_to(&[1], dst).await?;
+
+        Ok(())
+    });
+
+    sim.run()
+}
+
 #[test]
 fn udp_ipv4_multicast() -> Result {
     let mut sim = Builder::new()