diff --git a/packages/cultmesh-rs/src/lib.rs b/packages/cultmesh-rs/src/lib.rs index 7999f18b..135314bb 100644 --- a/packages/cultmesh-rs/src/lib.rs +++ b/packages/cultmesh-rs/src/lib.rs @@ -955,6 +955,7 @@ fn connect_rudp_client( socket.set_read_timeout(Some(poll_interval))?; let mut client = CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + media_delivery: None, runtime_id: runtime_id.to_string(), socket, mode: cultnet_rs::CultNetRudpSocketMode::Client, diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index c0f191b4..df14b4ea 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -163,6 +163,7 @@ fn client(target: SocketAddr, id: u32) -> Result +//! media_delivery_probe send +//! ``` +//! +//! The sender reports amplification: wire bytes divided by offered payload +//! bytes. At 1.0 nothing was retransmitted. Well above 1.0 under loss is the +//! pathology. + +use std::env; +use std::net::{SocketAddr, UdpSocket}; +use std::time::{Duration, Instant}; + +use anyhow::{Result, anyhow}; +use cultnet_rs::{ + CultNetRudpServerHubOptions, CultNetRudpSocketTransportConnection, + CultNetRudpSocketTransportOptions, CultNetTransportDelivery, +}; + +const CHANNEL: &str = "media"; +const DRAIN: Duration = Duration::from_secs(5); + +fn main() -> Result<()> { + let args: Vec = env::args().collect(); + match args.get(1).map(String::as_str) { + Some("recv") => recv( + args.get(2).ok_or_else(|| anyhow!("bind addr"))?.parse()?, + args.get(3).ok_or_else(|| anyhow!("expect count"))?.parse()?, + ), + Some("send") => send( + args.get(2).ok_or_else(|| anyhow!("endpoint"))?.parse()?, + match args.get(3).map(String::as_str) { + Some("unreliable") => CultNetTransportDelivery::Unreliable, + Some("reliable") => CultNetTransportDelivery::Reliable, + other => return Err(anyhow!("delivery must be reliable|unreliable, got {other:?}")), + }, + args.get(4).ok_or_else(|| anyhow!("payload bytes"))?.parse()?, + args.get(5).ok_or_else(|| anyhow!("count"))?.parse()?, + args.get(6).map(|v| v.parse()).transpose()?.unwrap_or(12.0), + ), + _ => Err(anyhow!( + "usage: media_delivery_probe recv | send " + )), + } +} + +fn recv(bind: SocketAddr, expect: u64) -> Result<()> { + let socket = UdpSocket::bind(bind)?; + socket.set_nonblocking(true)?; + let mut hub = cultnet_rs::CultNetRudpServerHub::new(CultNetRudpServerHubOptions::new( + "probe-recv", + socket, + 0x0BE0_0001, + ))?; + + println!("listening on {bind}, expecting {expect} frames"); + let started = Instant::now(); + let mut frames = 0u64; + let mut payload_bytes = 0u64; + let mut last_progress = Instant::now(); + + loop { + while let Some(event) = hub.receive_event_once()? { + if let cultnet_rs::CultNetRudpServerEvent::Frame { frame, .. } = event + && frame.channel_id == CHANNEL + { + frames += 1; + payload_bytes += frame.payload.len() as u64; + last_progress = Instant::now(); + } + } + if frames >= expect || last_progress.elapsed() > DRAIN { + break; + } + std::thread::sleep(Duration::from_micros(200)); + } + + let stats = hub.stats(); + let elapsed = started.elapsed().as_secs_f64(); + println!("RESULT recv frames={frames} expected={expect} payload_bytes={payload_bytes}"); + println!( + "RESULT recv wire_bytes_in={} wire_bytes_out={} elapsed_s={elapsed:.2} delivered_pct={:.2}", + stats.bytes_received, + stats.bytes_sent, + (frames as f64 / expect as f64) * 100.0 + ); + Ok(()) +} + +fn send( + endpoint: SocketAddr, + delivery: CultNetTransportDelivery, + payload_bytes: usize, + count: u64, + target_mbps: f64, +) -> Result<()> { + let socket = UdpSocket::bind("0.0.0.0:0")?; + socket.set_nonblocking(true)?; + let mut options = + CultNetRudpSocketTransportOptions::client("probe-send", socket, endpoint, 0x0BE0_0001); + options.media_delivery = Some(delivery); + options.media_reliable_expire_after_ms = None; + + let mut transport = CultNetRudpSocketTransportConnection::new(options)?; + transport.connect(b"probe".to_vec())?; + let deadline = Instant::now() + Duration::from_secs(10); + while !transport.connected() { + let _ = transport.receive_once()?; + transport.poll_resends()?; + if Instant::now() >= deadline { + return Err(anyhow!("timed out connecting to {endpoint}")); + } + std::thread::sleep(Duration::from_millis(2)); + } + + // Pace to a realistic media rate. An unpaced blast overruns the receiver's + // socket buffer and manufactures loss that has nothing to do with the link. + let per_packet = Duration::from_secs_f64((payload_bytes as f64 * 8.0) / (target_mbps * 1e6)); + let payload = vec![0xA5u8; payload_bytes]; + let offered = payload_bytes as u64 * count; + let started = Instant::now(); + let mut sent = 0u64; + let mut would_block = 0u64; + + while sent < count { + match transport.send(CHANNEL, payload.clone()) { + Ok(()) => sent += 1, + Err(error) if error.to_string().contains("would block") => { + would_block += 1; + transport.poll_resends()?; + let _ = transport.receive_once()?; + std::thread::sleep(Duration::from_micros(500)); + } + Err(error) => return Err(error), + } + if sent % 64 == 0 { + let _ = transport.receive_once()?; + transport.poll_resends()?; + } + let due = started + per_packet.mul_f64(sent as f64); + if let Some(wait) = due.checked_duration_since(Instant::now()) { + std::thread::sleep(wait); + } + } + let send_elapsed = started.elapsed(); + + // Let retransmission run: this is where a reliable channel spends what an + // unreliable one does not. + let drain_start = Instant::now(); + while drain_start.elapsed() < DRAIN { + let _ = transport.receive_once()?; + transport.poll_resends()?; + std::thread::sleep(Duration::from_millis(1)); + } + + let stats = transport.stats(); + let amplification = stats.bytes_sent as f64 / offered as f64; + println!( + "RESULT send delivery={delivery:?} payload_bytes={payload_bytes} count={count} offered={offered} target_mbps={target_mbps}" + ); + println!( + "RESULT send wire_bytes={} amplification={amplification:.3} send_s={:.2} total_s={:.2} would_block={would_block} reliable_expired={}", + stats.bytes_sent, + send_elapsed.as_secs_f64(), + started.elapsed().as_secs_f64(), + stats.reliable_packets_expired + ); + Ok(()) +} diff --git a/packages/cultnet-rs/src/cultmesh.rs b/packages/cultnet-rs/src/cultmesh.rs index efc8acad..14998396 100644 --- a/packages/cultnet-rs/src/cultmesh.rs +++ b/packages/cultnet-rs/src/cultmesh.rs @@ -319,6 +319,7 @@ impl CultMesh { let socket = bind_rudp_socket(&options)?; CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { media_reliable_expire_after_ms: None, + media_delivery: None, runtime_id: runtime_id.into(), socket, mode: CultNetRudpSocketMode::Server, @@ -343,6 +344,7 @@ impl CultMesh { let socket = bind_rudp_socket(&options)?; CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { media_reliable_expire_after_ms: None, + media_delivery: None, runtime_id: runtime_id.into(), socket, mode: CultNetRudpSocketMode::Client, diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index f1fd4ca4..b7019b1b 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -1210,6 +1210,8 @@ pub struct CultNetRudpSocketTransportOptions { pub max_pending_reliable_packets: Option, pub reconnect_policy: Option, pub media_reliable_expire_after_ms: Option, + /// `None` means the historical default, `Reliable`. + pub media_delivery: Option, } impl CultNetRudpSocketTransportOptions { @@ -1221,6 +1223,7 @@ impl CultNetRudpSocketTransportOptions { ) -> Self { Self { media_reliable_expire_after_ms: Some(DEFAULT_MEDIA_RELIABLE_EXPIRE_AFTER_MS), + media_delivery: None, runtime_id: runtime_id.into(), socket, mode: CultNetRudpSocketMode::Client, @@ -1239,6 +1242,7 @@ impl CultNetRudpSocketTransportOptions { pub fn server(runtime_id: impl Into, socket: UdpSocket, connection_id: u32) -> Self { Self { media_reliable_expire_after_ms: Some(DEFAULT_MEDIA_RELIABLE_EXPIRE_AFTER_MS), + media_delivery: None, runtime_id: runtime_id.into(), socket, mode: CultNetRudpSocketMode::Server, @@ -1280,12 +1284,15 @@ pub struct CultNetRudpServerHubOptions { pub max_pending_reliable_packets: Option, pub max_peers: usize, pub media_reliable_expire_after_ms: Option, + /// `None` means the historical default, `Reliable`. + pub media_delivery: Option, } impl CultNetRudpServerHubOptions { pub fn new(runtime_id: impl Into, socket: UdpSocket, connection_id: u32) -> Self { Self { media_reliable_expire_after_ms: Some(DEFAULT_MEDIA_RELIABLE_EXPIRE_AFTER_MS), + media_delivery: None, runtime_id: runtime_id.into(), socket, connection_id, @@ -1372,6 +1379,7 @@ impl CultNetRudpServerHub { options.runtime_id, RudpTransportProfileOptions { media_reliable_expire_after_ms: options.media_reliable_expire_after_ms, + media_delivery: options.media_delivery, transport_id: options.transport_id, host: Some(local_addr.ip().to_string()), port: Some(local_addr.port()), @@ -1451,6 +1459,7 @@ impl CultNetRudpServerHub { channel_id, now_ms(), channel_reliable_expire_after_ms(&self.profile, channel_id), + channel_delivery(&self.profile, channel_id), ), self.max_fragment_bytes, )? @@ -1671,6 +1680,7 @@ impl CultNetRudpSocketTransportConnection { options.runtime_id, RudpTransportProfileOptions { media_reliable_expire_after_ms: options.media_reliable_expire_after_ms, + media_delivery: options.media_delivery, transport_id: options.transport_id, host: Some(local_addr.ip().to_string()), port: Some(local_addr.port()), @@ -1738,6 +1748,7 @@ impl CultNetRudpSocketTransportConnection { channel_id, now_ms(), channel_reliable_expire_after_ms(&self.profile, channel_id), + channel_delivery(&self.profile, channel_id), ); let packets = self.session @@ -1759,6 +1770,7 @@ impl CultNetRudpSocketTransportConnection { channel_id, now_ms(), channel_reliable_expire_after_ms(&self.profile, channel_id), + channel_delivery(&self.profile, channel_id), ); // This lineage has no reliable-expiry concept, so every reliable channel is // already non-expiring. Restore the expiry half of this guard if one is added. @@ -2073,6 +2085,8 @@ pub struct RudpTransportProfileOptions { pub max_pending_reliable_packets: Option, pub reconnect_policy: Option, pub media_reliable_expire_after_ms: Option, + /// `None` means the historical default, `Reliable`. + pub media_delivery: Option, } pub fn create_rudp_transport_profile( @@ -2125,7 +2139,9 @@ pub fn create_rudp_transport_profile( }, CultNetTransportChannel { channel_id: "media".to_string(), - delivery: CultNetTransportDelivery::Reliable, + delivery: options + .media_delivery + .unwrap_or(CultNetTransportDelivery::Reliable), ordering: CultNetTransportOrdering::Unordered, max_payload_bytes: options.max_payload_bytes, max_fragment_bytes: options.max_fragment_bytes, @@ -2254,10 +2270,30 @@ fn channel_reliable_expire_after_ms( .and_then(|channel| channel.reliable_expire_after_ms) } +/// The advertised profile owns each channel's delivery, for the same reason it +/// owns expiry: a send path that hardcodes reliability can disagree with the +/// profile the peer was shown, and then no one can say what the channel does. +/// +/// Falls back to `Reliable` for a channel the profile does not describe, which +/// is what every channel got before delivery was configurable. +fn channel_delivery( + profile: &CultNetTransportProfile, + channel_id: &str, +) -> CultNetTransportDelivery { + profile + .transports + .iter() + .flat_map(|transport| transport.channels.iter()) + .find(|channel| channel.channel_id == channel_id) + .map(|channel| channel.delivery) + .unwrap_or(CultNetTransportDelivery::Reliable) +} + fn channel_send_options( channel_id: &str, now_ms: u64, reliable_expire_after_ms: Option, + delivery: CultNetTransportDelivery, ) -> CultNetRudpSendOptions { match channel_id { "schema" => CultNetRudpSendOptions { @@ -2274,8 +2310,11 @@ fn channel_send_options( now_ms, reliable_expire_after_ms: None, }, + // The one channel whose delivery a caller chooses. Media that has gone + // stale is worth dropping, not retransmitting: a reliable media channel + // under loss adds load exactly when the link has least to give. "media" => CultNetRudpSendOptions { - reliable: true, + reliable: matches!(delivery, CultNetTransportDelivery::Reliable), ordered: false, sequenced: false, now_ms, diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index 851646cf..7be6ea18 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -359,6 +359,7 @@ fn rudp_transport_profile_advertises_state_and_realtime_channels() { let profile = create_rudp_transport_profile( "rust-rudp", cultnet_rs::RudpTransportProfileOptions { + media_delivery: None, transport_id: Some("public-rudp".to_string()), host: Some("127.0.0.1".to_string()), port: Some(7777), @@ -1248,6 +1249,7 @@ fn rudp_socket_transport_handshakes_and_carries_reliable_ordered_schema_frames() let connection_id = 0x1020_3040; let mut server = CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + media_delivery: None, runtime_id: "rust-rudp-server".to_string(), socket: server_socket, mode: CultNetRudpSocketMode::Server, @@ -1264,6 +1266,7 @@ fn rudp_socket_transport_handshakes_and_carries_reliable_ordered_schema_frames() })?; let mut client = CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + media_delivery: None, runtime_id: "rust-rudp-client".to_string(), socket: client_socket, mode: CultNetRudpSocketMode::Client, @@ -1340,6 +1343,7 @@ fn rudp_socket_transport_carries_fragmented_reliable_ordered_schema_frames() -> let connection_id = 0x1020_3041; let mut server = CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + media_delivery: None, runtime_id: "rust-rudp-fragment-server".to_string(), socket: server_socket, mode: CultNetRudpSocketMode::Server, @@ -1356,6 +1360,7 @@ fn rudp_socket_transport_carries_fragmented_reliable_ordered_schema_frames() -> })?; let mut client = CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + media_delivery: None, runtime_id: "rust-rudp-fragment-client".to_string(), socket: client_socket, mode: CultNetRudpSocketMode::Client, @@ -1470,6 +1475,7 @@ fn rudp_socket_flush_waits_for_large_fragment_delivery() -> Result<()> { let connection_id = 0x1020_3043; let mut server = CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + media_delivery: None, runtime_id: "rust-rudp-window-server".into(), socket: server_socket, mode: CultNetRudpSocketMode::Server, @@ -1486,6 +1492,7 @@ fn rudp_socket_flush_waits_for_large_fragment_delivery() -> Result<()> { })?; let mut client = CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + media_delivery: None, runtime_id: "rust-rudp-window-client".into(), socket: client_socket, mode: CultNetRudpSocketMode::Client, diff --git a/packages/cultnet-rs/tests/media_channel_delivery.rs b/packages/cultnet-rs/tests/media_channel_delivery.rs new file mode 100644 index 00000000..499be0d3 --- /dev/null +++ b/packages/cultnet-rs/tests/media_channel_delivery.rs @@ -0,0 +1,92 @@ +//! Media delivery is the caller's choice, and the advertised profile is where +//! that choice lives. +//! +//! Before this, `media` was hardcoded `Reliable` in two places — the profile +//! builder and the per-channel send options — so no caller could say otherwise. +//! A producer asking for an unreliable media lane got a reliable one that +//! retransmits under loss, which adds load exactly when the link has least to +//! give. Muninn asked for that lane by passing no expiry, meaning "unreliable, +//! so expiry is moot", and got reliable-with-no-expiry instead: retransmit +//! forever. +//! +//! The default stays `Reliable` so every existing consumer is untouched. + +use std::net::UdpSocket; + +use cultnet_rs::{ + CultNetRudpSocketTransportConnection, CultNetRudpSocketTransportOptions, + CultNetTransportDelivery, +}; + +fn media_channel_delivery( + media_delivery: Option, +) -> (CultNetTransportDelivery, Option) { + let socket = UdpSocket::bind("127.0.0.1:0").expect("binds"); + let endpoint = "127.0.0.1:5204".parse().expect("parses"); + let mut options = CultNetRudpSocketTransportOptions::client("test-media", socket, endpoint, 7); + options.media_delivery = media_delivery; + options.media_reliable_expire_after_ms = None; + + let transport = CultNetRudpSocketTransportConnection::new(options).expect("opens"); + let channel = transport + .profile + .transports + .first() + .expect("a transport") + .channels + .iter() + .find(|channel| channel.channel_id == "media") + .expect("a media channel"); + (channel.delivery, channel.reliable_expire_after_ms) +} + +#[test] +fn media_defaults_to_reliable_so_existing_consumers_do_not_move() { + let (delivery, _) = media_channel_delivery(None); + assert_eq!(delivery, CultNetTransportDelivery::Reliable); +} + +#[test] +fn a_caller_can_ask_for_an_unreliable_media_lane() { + let (delivery, expiry) = media_channel_delivery(Some(CultNetTransportDelivery::Unreliable)); + assert_eq!(delivery, CultNetTransportDelivery::Unreliable); + assert_eq!( + expiry, None, + "an unreliable channel has nothing to expire; the pairing should stay honest" + ); +} + +#[test] +fn asking_for_reliable_explicitly_matches_the_default() { + let (explicit, _) = media_channel_delivery(Some(CultNetTransportDelivery::Reliable)); + let (defaulted, _) = media_channel_delivery(None); + assert_eq!(explicit, defaulted); +} + +/// The channels whose semantics are fixed by design must not drift with this. +#[test] +fn the_other_channels_keep_their_fixed_delivery() { + let socket = UdpSocket::bind("127.0.0.1:0").expect("binds"); + let endpoint = "127.0.0.1:5204".parse().expect("parses"); + let mut options = CultNetRudpSocketTransportOptions::client("test-media", socket, endpoint, 8); + options.media_delivery = Some(CultNetTransportDelivery::Unreliable); + + let transport = CultNetRudpSocketTransportConnection::new(options).expect("opens"); + let channels = &transport + .profile + .transports + .first() + .expect("a transport") + .channels; + + let delivery_of = |id: &str| { + channels + .iter() + .find(|channel| channel.channel_id == id) + .unwrap_or_else(|| panic!("channel {id}")) + .delivery + }; + + assert_eq!(delivery_of("schema"), CultNetTransportDelivery::Reliable); + assert_eq!(delivery_of("realtime"), CultNetTransportDelivery::Unreliable); +}