From c74f336b7c01ab1f109be9604cdc0f2a0993c5d8 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 12 Nov 2026 09:34:58 +0000 Subject: [PATCH] Track synchronisation universes separately from data universes A data packet with a sync address made the receiver call listen_universes for that address. That joined its multicast group, which is needed for the sync packets, but also added it to the user's data universes, so data sent on the sync universe was passed up although the user never listened to it, and the group was never left. Sync addresses are now kept in their own map with the time each was last named. Their groups are joined without adding them to the data universes, and left once no data packet has named them for the data loss timeout. listen_universes and mute_universe account for a group that is also in use for synchronisation, so it isn't joined twice or left early. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01KzaQVPkrkczgX2g7wGkip5 --- src/receive.rs | 179 +++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 174 insertions(+), 5 deletions(-) diff --git a/src/receive.rs b/src/receive.rs index 15f4141..dfbf295 100644 --- a/src/receive.rs +++ b/src/receive.rs @@ -197,6 +197,14 @@ pub struct SacnReceiver { /// Flag which indicates if an `UniverseTimeout` error should be thrown if it is detected that a source has timed out. announce_timeout: bool, + + /// Synchronisation universes that received data packets have named, with the time each was last named. + /// + /// The receiver joins the multicast group of each so that it receives their synchronisation packets, without adding + /// them to `universes`: data sent on a synchronisation universe is only passed up if the user listens to it. Entries + /// that haven't been named for `E131_NETWORK_DATA_LOSS_TIMEOUT` are removed and their group left, unless the user + /// listens to that universe. + sync_universes: HashMap, } /// Represents an sACN source/sender on the network that has been discovered by this sACN receiver by receiving universe discovery packets. @@ -306,6 +314,7 @@ impl SacnReceiver { announce_source_discovery: ANNOUNCE_SOURCE_DISCOVERY_DEFAULT, announce_stream_termination: ANNOUNCE_STREAM_TERMINATION_DEFAULT, announce_timeout: ANNOUNCE_TIMEOUT_DEFAULT, + sync_universes: HashMap::new(), }; sri.listen_universes(&[E131_DISCOVERY_UNIVERSE])?; @@ -408,7 +417,8 @@ impl SacnReceiver { // Value not found, i is the position it should be inserted self.universes.insert(i, *u); - if self.is_multicast_enabled() { + // The group may already have been joined to receive synchronisation packets for this universe. + if self.is_multicast_enabled() && !self.sync_universes.contains_key(u) { self.receiver.listen_multicast_universe(*u)?; } } else { @@ -435,9 +445,13 @@ impl SacnReceiver { Err(SacnError::UniverseNotFound(universe)) } Ok(i) => { - // If value found then don't insert to avoid duplicates. self.universes.remove(i); - self.receiver.mute_multicast_universe(universe) + if self.sync_universes.contains_key(&universe) { + // Still needed for synchronisation packets; the group is left once it is no longer used for that. + Ok(()) + } else { + self.receiver.mute_multicast_universe(universe) + } } } } @@ -510,6 +524,7 @@ impl SacnReceiver { // always check timeouts self.sequences.check_timeouts(self.announce_timeout)?; self.check_waiting_data_timeouts(); + self.expire_sync_universes(); return Err(io::Error::new( // Use the right expected error for the operating system. if cfg!(target_os = "windows") { @@ -531,6 +546,7 @@ impl SacnReceiver { loop { self.sequences.check_timeouts(self.announce_timeout)?; self.check_waiting_data_timeouts(); + self.expire_sync_universes(); // In the case of `timeout` being longer than `E131_NETWORK_DATA_LOSS_TIMEOUT`: // Forces the actual timeout used for receiving from the underlying network to never exceed E131_NETWORK_DATA_LOSS_TIMEOUT. @@ -754,7 +770,7 @@ impl SacnReceiver { } else { // As per ANSI E1.31-2018 Appendix B.2 the receiver should listen at the synchronisation address when a data packet is received with a non-zero // synchronisation address. - self.listen_universes(&[data_pkt.synchronization_address])?; + self.track_sync_universe(data_pkt.synchronization_address)?; let vals: Vec = data_pkt.data.property_values.into_owned(); let dmx_data: DMXData = DMXData { @@ -847,7 +863,11 @@ impl SacnReceiver { cid: Uuid, sync_pkt: SynchronizationPacketFramingLayer, ) -> Result>> { - if !self.is_listening(&sync_pkt.synchronization_address) { + if !self.is_listening(&sync_pkt.synchronization_address) + && !self + .sync_universes + .contains_key(&sync_pkt.synchronization_address) + { return Ok(None); // If not listening for this universe then ignore the packet. } @@ -972,6 +992,42 @@ impl SacnReceiver { None // No source fully discovered. } + /// Records that a received data packet named `sync_uni` as its synchronisation address, joining that universe's + /// multicast group if it isn't joined already so that the synchronisation packets are received. + /// + /// # Errors + /// Returns an error if the multicast group cannot be joined. The universe isn't recorded in that case, so joining is + /// tried again with the next data packet that names it. + fn track_sync_universe(&mut self, sync_uni: u16) -> Result<()> { + if !self.sync_universes.contains_key(&sync_uni) + && !self.is_listening(&sync_uni) + && self.is_multicast_enabled() + { + self.receiver.listen_multicast_universe(sync_uni)?; + } + self.sync_universes.insert(sync_uni, Instant::now()); + Ok(()) + } + + /// Stops tracking synchronisation universes that no data packet has named for `E131_NETWORK_DATA_LOSS_TIMEOUT`, + /// leaving their multicast groups unless the user listens to that universe. + fn expire_sync_universes(&mut self) { + let expired: Vec = self + .sync_universes + .iter() + .filter(|(_, last)| last.elapsed() >= E131_NETWORK_DATA_LOSS_TIMEOUT) + .map(|(u, _)| *u) + .collect(); + + for u in expired { + self.sync_universes.remove(&u); + if !self.is_listening(&u) && self.is_multicast_enabled() { + // Nothing useful can be done if leaving fails; the membership ends when the socket closes. + let _ = self.receiver.mute_multicast_universe(u); + } + } + } + /// Goes through all the waiting data and removes any which has timed out as a sync-packet for it hasn't been received within the `E131_NETWORK_DATA_LOSS_TIMEOUT` /// period as specified by ANSI E1.31-2018 Section 11.1.2. fn check_waiting_data_timeouts(&mut self) { @@ -993,6 +1049,17 @@ impl SacnReceiver { impl Drop for SacnReceiver { fn drop(&mut self) { let universes = self.universes.clone(); + // Leave the groups joined only for synchronisation first; mute_universe would otherwise skip them. + let sync_only: Vec = self + .sync_universes + .keys() + .filter(|u| !self.is_listening(u)) + .copied() + .collect(); + self.sync_universes.clear(); + for u in sync_only { + let _ = self.receiver.mute_multicast_universe(u); + } for u in universes { // Cannot return an error or pass it onto the user because drop might be called during a panic. // Therefore if there is an error cleaning up the only options are ignore, notify or panic. @@ -3359,4 +3426,106 @@ mod test { res1 ); } + + /// Builds a data packet for `universe` that waits for a synchronisation packet on `sync_uni`. + fn sync_data_packet<'a>(universe: u16, sync_uni: u16, seq: u8) -> DataPacketFramingLayer<'a> { + let mut pkt = generate_data_packet_framing_layer_seq_num(universe, seq); + pkt.synchronization_address = sync_uni; + pkt + } + + /// A synchronisation address named by a data packet is joined for its sync packets, but data sent on that universe + /// isn't passed up unless the user listens to it. + #[test] + fn test_sync_universe_not_added_to_data_universes() { + const DATA_UNI: u16 = 1; + const SYNC_UNI: u16 = 7; + let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), ACN_SDT_MULTICAST_PORT); + let mut rcv = SacnReceiver::with_ip(addr, None).unwrap(); + rcv.listen_universes(&[DATA_UNI]).unwrap(); + let src = Uuid::new_v4(); + + assert!( + rcv.handle_data_packet(src, sync_data_packet(DATA_UNI, SYNC_UNI, 0)) + .unwrap() + .is_none() + ); + assert!(!rcv.is_listening(&SYNC_UNI)); + assert!(rcv.sync_universes.contains_key(&SYNC_UNI)); + + // Data on the synchronisation universe is ignored: the user never listened to it. + let on_sync_uni = generate_data_packet_framing_layer_seq_num(SYNC_UNI, 0); + assert!(rcv.handle_data_packet(src, on_sync_uni).unwrap().is_none()); + + // The sync packet still releases the waiting data. + let released = rcv + .handle_sync_packet(src, generate_sync_packet_framing_layer_seq_num(SYNC_UNI, 0)) + .unwrap() + .expect("sync packet should release the waiting data"); + assert_eq!(released.len(), 1); + assert_eq!(released[0].universe, DATA_UNI); + } + + /// A synchronisation universe nobody has named for the data loss timeout is forgotten, so its sync packets are + /// ignored and its multicast group can be left. + #[test] + fn test_sync_universe_expires() { + const DATA_UNI: u16 = 1; + const SYNC_UNI: u16 = 7; + let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), ACN_SDT_MULTICAST_PORT); + let mut rcv = SacnReceiver::with_ip(addr, None).unwrap(); + rcv.listen_universes(&[DATA_UNI]).unwrap(); + let src = Uuid::new_v4(); + + rcv.handle_data_packet(src, sync_data_packet(DATA_UNI, SYNC_UNI, 0)) + .unwrap(); + let expired = Instant::now() - E131_NETWORK_DATA_LOSS_TIMEOUT - Duration::from_millis(1); + rcv.sync_universes.insert(SYNC_UNI, expired); + rcv.expire_sync_universes(); + assert!(!rcv.sync_universes.contains_key(&SYNC_UNI)); + + let res = rcv + .handle_sync_packet(src, generate_sync_packet_framing_layer_seq_num(SYNC_UNI, 0)) + .unwrap(); + assert!(res.is_none(), "sync packet on a forgotten universe was acted on"); + + // Naming it again joins it again without error. + rcv.handle_data_packet(src, sync_data_packet(DATA_UNI, SYNC_UNI, 1)) + .unwrap(); + assert!(rcv.sync_universes.contains_key(&SYNC_UNI)); + } + + /// Listening to, then muting, a universe that is also in use for synchronisation must not join its group twice or + /// leave it while it is still needed for sync packets. + #[test] + fn test_listen_and_mute_universe_also_used_for_sync() { + const DATA_UNI: u16 = 1; + const SYNC_UNI: u16 = 7; + let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), ACN_SDT_MULTICAST_PORT); + let mut rcv = SacnReceiver::with_ip(addr, None).unwrap(); + rcv.listen_universes(&[DATA_UNI]).unwrap(); + let src = Uuid::new_v4(); + + rcv.handle_data_packet(src, sync_data_packet(DATA_UNI, SYNC_UNI, 0)) + .unwrap(); + + // Joining the same group twice fails on Linux (EADDRINUSE), so this would error if it re-joined. + rcv.listen_universes(&[SYNC_UNI]).unwrap(); + assert!(rcv.is_listening(&SYNC_UNI)); + + rcv.mute_universe(SYNC_UNI).unwrap(); + assert!(!rcv.is_listening(&SYNC_UNI)); + assert!(rcv.sync_universes.contains_key(&SYNC_UNI)); + + // Still joined: the group can be left once, when the sync universe expires. Leaving a group that isn't joined + // errors, so a second leave here would fail. + rcv.sync_universes.insert( + SYNC_UNI, + Instant::now() - E131_NETWORK_DATA_LOSS_TIMEOUT - Duration::from_millis(1), + ); + rcv.expire_sync_universes(); + #[cfg(not(target_os = "windows"))] + assert!(rcv.receiver.mute_multicast_universe(SYNC_UNI).is_err()); + } } +