From 5badc2d9f4d6f2ed50096bdc6059d970ff9c8694 Mon Sep 17 00:00:00 2001 From: chazmm Date: Thu, 10 Sep 2026 03:10:15 +0100 Subject: [PATCH] Remove fixed model transfer rate caps and stale queue drops --- crates/skate-net/src/blob.rs | 20 +-- crates/skate-net/src/lobby.rs | 1 - .../skate-net/tests/appearance_bandwidth.rs | 135 ++++++++++++++++++ crates/skate-steam-relay/src/main.rs | 14 +- docs/multiplayer-test.md | 12 +- 5 files changed, 156 insertions(+), 26 deletions(-) create mode 100644 crates/skate-net/tests/appearance_bandwidth.rs diff --git a/crates/skate-net/src/blob.rs b/crates/skate-net/src/blob.rs index 71a2fda1..6d07f9ce 100644 --- a/crates/skate-net/src/blob.rs +++ b/crates/skate-net/src/blob.rs @@ -131,10 +131,7 @@ pub struct Blobs { cache: BTreeMap<[u8; 32], Blob>, sending: BTreeMap<(u64, u64), Sending>, acknowledgements: BTreeSet<(u64, u64)>, - credits: BTreeMap, - last: u64, round: usize, - pub congested: bool, } impl Blobs { fn reserve(&mut self, size: usize) -> bool { @@ -318,8 +315,6 @@ impl Blobs { self.sending.retain(|(peer, id), _| { members.contains(id) && peers.iter().any(|(p, _, _)| p == peer) }); - self.credits - .retain(|peer, _| peers.iter().any(|(p, _, _)| p == peer)); let mut out = vec![]; for (peer, origin) in std::mem::take(&mut self.acknowledgements) { if !peers.iter().any(|(p, _, _)| *p == peer) { @@ -341,11 +336,9 @@ impl Blobs { data.extend(mask); out.push(Outgoing { peer, data }); } - let dt = now.saturating_sub(self.last).min(1000) as f64 / 1000.; - self.last = now; - // Same rates for direct and Steam. Bursts cover slow render frames; ACKs - // bound outstanding data and relay congestion reduces bulk throughput. - let rate = if self.congested { 128_000. } else { 4_000_000. }; + // ACKs and the loss-responsive window control throughput, rather than + // a fixed bytes/second cap. Bound each service call so model uploads + // cannot monopolize the game thread or overflow the local IPC queues. let mut global = 2_000_000usize; let mut peers = peers.to_vec(); if !peers.is_empty() { @@ -357,8 +350,7 @@ impl Blobs { if recipient == 0 { continue; } - let credit = self.credits.entry(peer).or_default(); - *credit = (*credit + dt * rate).min(512_000.); + let mut budget = 512_000usize; let mut origins: Vec<_> = self .actors .iter() @@ -429,7 +421,7 @@ impl Blobs { if global < size { return out; } - if *credit < size as f64 { + if budget < size { continue 'peers; } let mut data = packed::header(session, *origin, DATA, seq); @@ -440,7 +432,7 @@ impl Blobs { .unwrap() .sent .insert(i, now); - *credit -= size as f64; + budget -= size; global -= size; out.push(Outgoing { peer, data }); sent = true; diff --git a/crates/skate-net/src/lobby.rs b/crates/skate-net/src/lobby.rs index 575c2e94..fd0a1360 100644 --- a/crates/skate-net/src/lobby.rs +++ b/crates/skate-net/src/lobby.rs @@ -231,7 +231,6 @@ impl Session { self.is_host() || self.received_roster != 0 } pub fn set_congested(&mut self, congested: bool) { - self.blobs.congested = congested; self.link_budget = if congested { DEFAULT_LINK_BUDGET * 0.5 } else { diff --git a/crates/skate-net/tests/appearance_bandwidth.rs b/crates/skate-net/tests/appearance_bandwidth.rs new file mode 100644 index 00000000..2b6b2709 --- /dev/null +++ b/crates/skate-net/tests/appearance_bandwidth.rs @@ -0,0 +1,135 @@ +use skate_net::{ + Body, Pose, + lobby::{Info, Session}, + packed::{self, BodyState, Packed}, +}; +use std::{ + net::UdpSocket, + time::{Duration, Instant}, +}; + +#[test] +fn model_crosses_real_udp_host_while_movement_stays_current() { + let sockets: Vec<_> = (0..3) + .map(|_| { + let s = UdpSocket::bind("127.0.0.1:0").unwrap(); + skate_net::socket::configure(&s).unwrap(); + s + }) + .collect(); + let addresses: Vec<_> = sockets.iter().map(|s| s.local_addr().unwrap()).collect(); + let mut sessions: Vec<_> = (0..3) + .map(|i| { + Session::new( + 91, + Info { + id: 10 + i, + map: 1, + rig: 1, + physics: 1, + appearance: 1, + }, + if i == 0 { None } else { Some(1) }, + ) + }) + .collect(); + let model: Vec = (0..12_000_000).map(|i| (i * 17) as u8).collect(); + assert!(sessions[1].blobs.publish(11, model.clone())); + let start = Instant::now(); + let mut frame = 0; + let mut buffer = [0; 1500]; + let mut completed = None; + let mut last_data = 0; + let mut lost = std::collections::BTreeSet::new(); + loop { + let now = start.elapsed().as_millis() as u64; + assert!( + now < 15_000, + "transfer stalled at {:?}", + sessions[2].blobs.progress(12) + ); + for i in 0..3 { + loop { + match sockets[i].recv_from(&mut buffer) { + Ok((n, from)) => { + let peer = addresses.iter().position(|a| *a == from).unwrap() + 1; + // Lose selected first attempts on the client-host leg. + if i == 0 && n >= 33 && buffer[24] == skate_net::blob::DATA { + let chunk = u32::from_le_bytes(buffer[29..33].try_into().unwrap()); + if chunk % 37 == 0 && lost.insert(chunk) { + continue; + } + } + sessions[i].receive(peer as u64, &buffer[..n], now); + } + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => break, + Err(e) => panic!("{e}"), + } + } + // A stale Steam congestion sample must not impose 128 KB/s. + sessions[i].set_congested(true); + let pose = Pose { + p: [frame as f32 * 0.001, 0., 0.], + q: [0., 0., 0., 1.], + }; + sessions[i].publish( + packed::BODY, + Packed::body(&BodyState { + root: pose, + enabled: (1u64 << 33) - 1, + bodies: vec![ + Body { + pose, + velocity: [0.; 3], + angular: [0.; 3] + }; + 33 + ], + }) + .unwrap(), + now, + ); + let packets = sessions[i].service(now); + assert!(packets.iter().map(|p| p.data.len()).sum::() < 2_100_000); + for p in packets { + if p.data[24] == skate_net::blob::DATA { + last_data = now; + } + sockets[i] + .send_to(&p.data, addresses[p.peer as usize - 1]) + .unwrap(); + } + } + if now > 1000 { + assert!( + sessions[2] + .actors + .get(&11) + .and_then(|a| a.body.latest()) + .is_some_and(|b| now - b.received < 300), + "model transfer starved movement" + ); + } + if completed.is_none() && sessions[2].blobs.ready(11).is_some_and(|(_, b)| b == model) { + eprintln!("12 MB client-host-client UDP transfer: {now} ms at 15 FPS"); + assert!(now < 12_000); + completed = Some(now); + } + if let Some(done) = completed { + assert!( + sessions[2].actors[&11] + .body + .latest() + .is_some_and(|b| now - b.received < 300) + ); + if now > done + 1000 { + assert!(last_data < done + 500); + break; + } + } + frame += 1; + if let Some(wait) = Duration::from_millis(frame * 67).checked_sub(start.elapsed()) { + std::thread::sleep(wait); + } + } +} diff --git a/crates/skate-steam-relay/src/main.rs b/crates/skate-steam-relay/src/main.rs index b9743687..e2eb4d1c 100644 --- a/crates/skate-steam-relay/src/main.rs +++ b/crates/skate-steam-relay/src/main.rs @@ -52,15 +52,15 @@ fn run() -> Result<(), String> { return Err("Same Steam identity".into()); } client.networking_utils().init_relay_network_access(); - // SDK defaults cap model uploads at 256 KB/s. Raise only the ceiling; - // retain Steam's congestion control and the lobby's bulk pacing/backoff. + // Remove our throughput ceiling without forcing a minimum send rate. + // Steam still controls the connection; the game bounds in-flight data. // SAFETY: Steam is initialized and owns this interface for `client`'s lifetime. unsafe { let utils = steamworks::sys::SteamAPI_SteamNetworkingUtils_SteamAPI_v004(); if utils.is_null() || !steamworks::sys::SteamAPI_ISteamNetworkingUtils_SetGlobalConfigValueInt32( utils, steamworks::sys::ESteamNetworkingConfigValue::k_ESteamNetworkingConfig_SendRateMax, - 4_500_000, + i32::MAX, ) { return Err("Could not configure Steam model transfer bandwidth".into()); } @@ -200,10 +200,10 @@ fn run() -> Result<(), String> { } parent_seen = Instant::now(); let state = matches!(kind, skate_net::packed::BODY | skate_net::packed::POSE); - if (state || kind == skate_net::blob::DATA) && peers.get(&target).is_some_and(|(_, queue)| *queue > 50) { - dropped += 1; - continue; - } + // Do not discard packets based on the once-per-second + // queue sample. That creates long artificial loss bursts. + // Steam handles backpressure; state remains NoDelay, and + // bulk retries/ACK windows remain owned by skate-net. let flags = if state { SendFlags::UNRELIABLE_NO_DELAY } else { diff --git a/docs/multiplayer-test.md b/docs/multiplayer-test.md index 5c1857e0..c49ef687 100644 --- a/docs/multiplayer-test.md +++ b/docs/multiplayer-test.md @@ -16,10 +16,14 @@ The host forwards available chunks while downloading. A sliding selective-ACK window replaces the old 32-chunk request/wait cycle. Each peer receives the current model once; completed content is cached for later swaps, and a late join receives the current selection. New selections cancel obsolete transfers. -Movement/collision packets are serviced first. Bulk traffic has its own bounded -4 MB/s per-link ceiling, grows its window as chunks are acknowledged, backs off -on loss/relay congestion, and retries missing -chunks. Datagram queues hold slow render frames' traffic on both direct sockets +Movement/collision packets are serviced first. Bulk traffic has no fixed +bytes-per-second cap: its acknowledgement window grows with successful delivery +and backs off on packet loss. Each service call is bounded to 512 KB per peer +and 2 MB overall to protect frame time and local IPC queues. Missing chunks retry. +Steam's configured maximum is the highest representable rate; its minimum is +unchanged, so this does not force traffic beyond Steam's connection control. +The relay no longer discards packets based on a once-per-second queue snapshot. +Movement retains Steam's NoDelay flag. Datagram queues hold slow render frames' traffic on both direct sockets and Steam IPC. The Steam send-rate ceiling is raised from its small-message default. The HUD shows character transfer progress. Each remote keeps its visible character until the new scenes, textures and every