diff --git a/crates/punktfunk-core/Cargo.toml b/crates/punktfunk-core/Cargo.toml index fa56ce50..91542792 100644 --- a/crates/punktfunk-core/Cargo.toml +++ b/crates/punktfunk-core/Cargo.toml @@ -77,7 +77,14 @@ libc = "0.2" # windows-sys (raw FFI, the quinn-udp choice): the high-level `windows` crate doesn't bind the # `WSASendMsg` extension function. WinSock feature gives WSASendMsg + WSAMSG/WSABUF/CMSGHDR. # Win32_System_IO too: WSASendMsg's signature references OVERLAPPED, so it's gated on that feature. -windows-sys = { version = "0.59", features = ["Win32_Networking_WinSock", "Win32_System_IO"] } +# Win32_NetworkManagement_QoS + Win32_Foundation: the qWAVE flow API for real on-the-wire DSCP +# marking (transport/qos_windows.rs) — plain IP_TOS is stripped by the Windows stack. +windows-sys = { version = "0.59", features = [ + "Win32_Networking_WinSock", + "Win32_System_IO", + "Win32_Foundation", + "Win32_NetworkManagement_QoS", +] } [dev-dependencies] proptest = "1" diff --git a/crates/punktfunk-core/src/transport/mod.rs b/crates/punktfunk-core/src/transport/mod.rs index 72c918c3..32768cd4 100644 --- a/crates/punktfunk-core/src/transport/mod.rs +++ b/crates/punktfunk-core/src/transport/mod.rs @@ -3,10 +3,12 @@ mod loopback; mod qos; +#[cfg(windows)] +mod qos_windows; mod udp; pub use loopback::{loopback_pair, LoopbackTransport}; -pub use qos::{grow_socket_buffers, set_dscp_default, set_media_qos, MediaClass}; +pub use qos::{grow_socket_buffers, set_dscp_default, set_media_qos, MediaClass, QosFlow}; /// Windows-only: reusable USO (UDP Send Offload) batch send for callers that own their own connected /// socket (the GameStream video sender) rather than going through [`UdpTransport`]. #[cfg(target_os = "windows")] diff --git a/crates/punktfunk-core/src/transport/qos.rs b/crates/punktfunk-core/src/transport/qos.rs index ceef4314..19eed00a 100644 --- a/crates/punktfunk-core/src/transport/qos.rs +++ b/crates/punktfunk-core/src/transport/qos.rs @@ -8,9 +8,10 @@ //! QoS-aware path (Wi-Fi WMM access categories, a managed switch, a shaped uplink) can prioritize it //! over bulk flows. Mirrors what Apollo/Sunshine tag — DSCP **CS5** for video, **CS6** for audio. It //! is **opt-in** (`PUNKTFUNK_DSCP=1`, or [`set_dscp_default`] from an embedder — the Android client -//! ties it to its experimental low-latency mode): DSCP can interact badly with some consumer ISPs/routers, and on -//! Windows a plain `IP_TOS` is silently stripped unless a qWAVE policy is active (Apollo uses the -//! qWAVE API there — that port is a follow-up; today this is a no-op on the wire on Windows). +//! ties it to its experimental low-latency mode): DSCP can interact badly with some consumer +//! ISPs/routers. On Windows a plain `IP_TOS` is silently stripped from the wire, so the marking +//! goes through qWAVE flows instead (see [`super::qos_windows`]) — the caller holds the returned +//! [`QosFlow`] guard for as long as the socket sends media. use std::net::UdpSocket; use std::sync::atomic::{AtomicBool, Ordering}; @@ -60,7 +61,7 @@ pub enum MediaClass { impl MediaClass { /// DSCP code point (the high 6 bits of the IPv4 TOS / IPv6 traffic-class byte). - const fn dscp(self) -> u32 { + pub(super) const fn dscp(self) -> u32 { match self { MediaClass::Video => 40, // CS5 MediaClass::Audio => 48, // CS6 @@ -92,20 +93,43 @@ pub(crate) fn dscp_enabled() -> bool { } } +/// RAII token for a socket's QoS marking. On Windows it is the qWAVE flow membership +/// ([`super::qos_windows::QosFlow`]) — dropping it removes the marking, so hold it for as long +/// as the socket sends media. Elsewhere DSCP rides the socket option itself and the token is +/// inert (and never constructed — [`set_media_qos`] returns `None`). +#[cfg(windows)] +pub use super::qos_windows::QosFlow; +#[cfg(not(windows))] +pub struct QosFlow { + _never: std::convert::Infallible, +} + /// Best-effort: tag `socket`'s outgoing packets for prioritized delivery of its media class. A no-op /// unless `PUNKTFUNK_DSCP=1`. Every step is best-effort (failures logged at debug, never fatal) — QoS /// is a nicety, not required for correctness. /// -/// IPv4 only (all current media sockets bind `0.0.0.0`); a v6 socket simply isn't tagged. On Windows -/// the `IP_TOS` set succeeds but the OS doesn't tag the wire without a qWAVE policy (follow-up). -pub fn set_media_qos(socket: &UdpSocket, class: MediaClass) { - if dscp_enabled() { +/// The socket must already be `connect`ed (Windows derives the qWAVE flow from the connected +/// 5-tuple). IPv4 only (all current media sockets bind `0.0.0.0`); a v6 socket simply isn't +/// tagged. Returns the [`QosFlow`] guard on Windows — keep it alive with the socket; `None` +/// elsewhere (the marking is a plain socket option) and whenever a step refused. +pub fn set_media_qos(socket: &UdpSocket, class: MediaClass) -> Option { + if !dscp_enabled() { + return None; + } + #[cfg(windows)] + { + super::qos_windows::add_media_flow(socket, class) + } + #[cfg(not(windows))] + { apply_media_qos(socket, class); + None } } /// The unconditional QoS application, factored out of [`set_media_qos`] so it is directly testable /// without touching the process-global `PUNKTFUNK_DSCP` env. Best-effort (every step logs-and-continues). +#[cfg_attr(windows, allow(dead_code))] fn apply_media_qos(socket: &UdpSocket, class: MediaClass) { let sock = socket2::SockRef::from(socket); // DSCP occupies the high 6 bits of the TOS byte → shift left 2. @@ -143,8 +167,8 @@ mod tests { fn qos_and_buffer_growth_are_best_effort_and_never_panic() { let sock = UdpSocket::bind("127.0.0.1:0").unwrap(); // No PUNKTFUNK_DSCP in the test env → early return; must not panic regardless. - set_media_qos(&sock, MediaClass::Video); - set_media_qos(&sock, MediaClass::Audio); + assert!(set_media_qos(&sock, MediaClass::Video).is_none()); + assert!(set_media_qos(&sock, MediaClass::Audio).is_none()); grow_socket_buffers(&sock); } diff --git a/crates/punktfunk-core/src/transport/qos_windows.rs b/crates/punktfunk-core/src/transport/qos_windows.rs new file mode 100644 index 00000000..ea5f0854 --- /dev/null +++ b/crates/punktfunk-core/src/transport/qos_windows.rs @@ -0,0 +1,134 @@ +//! qWAVE (qos2.h) DSCP marking — the Windows path of [`super::qos::set_media_qos`]. +//! +//! On Windows a plain `IP_TOS` setsockopt succeeds but the stack strips the mark from the wire: +//! marking requires membership in a qWAVE flow, which is how Apollo/Sunshine tag (qwave.dll). +//! [`QOSAddSocketToFlow`] with a traffic type yields the OS default marking (AudioVideo → DSCP +//! 40, Voice → 56, both WMM-mapped); the follow-up `QOSSetFlow(QOSSetOutgoingDSCPValue)` pins +//! the exact CS5/CS6 code points the other platforms mark. The pin needs an elevated process or +//! the "allow non-admin DSCP" group policy and silently keeps the traffic-type default +//! otherwise — the host runs as the SYSTEM service (`PunktfunkHost`), so the exact-DSCP path +//! applies exactly where it matters (the video egress); user-mode clients keep traffic-type +//! defaults (still WMM-useful). +//! +//! Same contract as the rest of [`super::qos`]: opt-in (`dscp_enabled`), and every step +//! debug-logs and continues — QoS is a nicety, never required for correctness. + +use super::qos::MediaClass; +use std::net::UdpSocket; +use std::os::windows::io::AsRawSocket; +use std::sync::OnceLock; +use windows_sys::Win32::Foundation::{GetLastError, HANDLE}; +use windows_sys::Win32::NetworkManagement::QoS::{ + QOSAddSocketToFlow, QOSCreateHandle, QOSRemoveSocketFromFlow, QOSSetFlow, + QOSSetOutgoingDSCPValue, QOSTrafficTypeAudioVideo, QOSTrafficTypeVoice, QOS_NON_ADAPTIVE_FLOW, + QOS_VERSION, +}; + +/// The process-wide qWAVE handle (`QOSCreateHandle` once, cached). `None` = qWAVE unavailable +/// (e.g. the QWAVE service is disabled) — every flow request then no-ops. Deliberately never +/// closed: it lives as long as the process, like the media sockets whose flows it carries. +fn qos_handle() -> Option { + static SLOT: OnceLock> = OnceLock::new(); + SLOT.get_or_init(|| { + let version = QOS_VERSION { + MajorVersion: 1, + MinorVersion: 0, + }; + let mut handle: HANDLE = std::ptr::null_mut(); + // SAFETY: both pointers are valid for the duration of the synchronous call. + if unsafe { QOSCreateHandle(&version, &mut handle) } == 0 { + tracing::debug!( + err = unsafe { GetLastError() }, + "QOSCreateHandle failed — qWAVE DSCP marking unavailable" + ); + None + } else { + Some(handle as usize) + } + }) + .map(|h| h as HANDLE) +} + +/// RAII qWAVE flow membership: while held, the socket's egress carries the flow's marking; +/// dropping removes the socket from the flow. (Closing the socket also removes it implicitly — +/// the guard makes teardown explicit and ordered, and must outlive the socket's traffic.) +pub struct QosFlow { + /// Raw `SOCKET` value only — never dereferenced; qWAVE tolerates an already-closed socket + /// on remove (the error is deliberately ignored). + socket: u64, + flow_id: u32, +} + +impl Drop for QosFlow { + fn drop(&mut self) { + if let Some(handle) = qos_handle() { + // SAFETY: handle/flow_id came from the successful add; a stale socket just errors. + unsafe { QOSRemoveSocketFromFlow(handle, self.socket as _, self.flow_id, 0) }; + } + } +} + +/// Put a **connected** media socket on a qWAVE flow of its class (video → +/// `QOSTrafficTypeAudioVideo`, audio → `QOSTrafficTypeVoice`), then best-effort pin the exact +/// DSCP the other platforms mark (CS5 = 40 / CS6 = 48). Returns the flow guard, or `None` when +/// a required step refused (logged at debug). +pub(super) fn add_media_flow(socket: &UdpSocket, class: MediaClass) -> Option { + let handle = qos_handle()?; + let traffic_type = match class { + MediaClass::Video => QOSTrafficTypeAudioVideo, + MediaClass::Audio => QOSTrafficTypeVoice, + }; + let raw = socket.as_raw_socket(); + let mut flow_id = 0u32; + // NULL destination = derive the flow's 5-tuple from the connected socket (every media + // socket is `connect`ed before it is tagged). + // SAFETY: the socket is live for the call; `flow_id` is a valid out-pointer. + let ok = unsafe { + QOSAddSocketToFlow( + handle, + raw as _, + std::ptr::null(), + traffic_type, + QOS_NON_ADAPTIVE_FLOW, + &mut flow_id, + ) + }; + if ok == 0 { + tracing::debug!( + err = unsafe { GetLastError() }, + ?class, + "QOSAddSocketToFlow failed — DSCP marking skipped" + ); + return None; + } + // Construct the guard FIRST so an early return below still removes the flow membership. + let flow = QosFlow { + socket: raw as u64, + flow_id, + }; + // Pin the exact code point. Succeeds for elevated processes or under the "allow non-admin + // DSCP" policy; otherwise the traffic-type default marking stands (40 / 56 — WMM-useful). + let dscp: u32 = class.dscp(); + // SAFETY: `buffer` points at 4 valid bytes for the synchronous (no OVERLAPPED) call. + let ok = unsafe { + QOSSetFlow( + handle, + flow_id, + QOSSetOutgoingDSCPValue, + 4, + &dscp as *const u32 as *const _, + 0, + std::ptr::null_mut(), + ) + }; + if ok == 0 { + tracing::debug!( + err = unsafe { GetLastError() }, + ?class, + "QOSSetFlow(OutgoingDSCPValue) refused — traffic-type default marking stands" + ); + } else { + tracing::debug!(?class, dscp, flow_id, "qWAVE flow pinned to exact DSCP"); + } + Some(flow) +} diff --git a/crates/punktfunk-core/src/transport/udp.rs b/crates/punktfunk-core/src/transport/udp.rs index 0c6b48ce..977cf1a2 100644 --- a/crates/punktfunk-core/src/transport/udp.rs +++ b/crates/punktfunk-core/src/transport/udp.rs @@ -446,6 +446,9 @@ pub fn spawn_data_punch(sock: UdpSocket, stop: std::sync::Arc, socket: UdpSocket, } @@ -464,10 +467,14 @@ impl UdpTransport { socket.connect(peer)?; super::qos::grow_socket_buffers(&socket); // The native data plane is video-dominant — tag it as the video class (opt-in via - // PUNKTFUNK_DSCP). Each end marks its own egress. - super::qos::set_media_qos(&socket, super::qos::MediaClass::Video); + // PUNKTFUNK_DSCP). Each end marks its own egress; the socket is connected by now, as + // the Windows qWAVE flow requires. + let qos_flow = super::qos::set_media_qos(&socket, super::qos::MediaClass::Video); socket.set_nonblocking(true)?; - Ok(UdpTransport { socket }) + Ok(UdpTransport { + _qos_flow: qos_flow, + socket, + }) } /// Host side of the data plane for clients that may sit behind NAT / a stateful inter-VLAN @@ -524,9 +531,15 @@ impl UdpTransport { socket.connect(target.as_deref().unwrap_or(fallback_peer))?; socket.set_read_timeout(None)?; super::qos::grow_socket_buffers(&socket); - super::qos::set_media_qos(&socket, super::qos::MediaClass::Video); + let qos_flow = super::qos::set_media_qos(&socket, super::qos::MediaClass::Video); socket.set_nonblocking(true)?; - Ok((UdpTransport { socket }, punched)) + Ok(( + UdpTransport { + _qos_flow: qos_flow, + socket, + }, + punched, + )) } /// A second handle to the data socket, for sending hole-punch keepalives ([`PUNCH_MAGIC`]) diff --git a/crates/punktfunk-host/src/gamestream/audio.rs b/crates/punktfunk-host/src/gamestream/audio.rs index 52af1888..2c80dbd3 100644 --- a/crates/punktfunk-host/src/gamestream/audio.rs +++ b/crates/punktfunk-host/src/gamestream/audio.rs @@ -257,9 +257,9 @@ fn run( audio_cap: &std::sync::Mutex>>, ) -> Result<()> { let sock = UdpSocket::bind(("0.0.0.0", AUDIO_PORT)).context("bind audio UDP")?; - // Grow SO_SNDBUF/RCVBUF + opt-in DSCP/QoS-tag this as the audio class (PUNKTFUNK_DSCP=1). + // Grow SO_SNDBUF/RCVBUF; the opt-in DSCP/QoS tag happens after connect below (Windows + // qWAVE derives the flow from the connected 5-tuple). punktfunk_core::transport::grow_socket_buffers(&sock); - punktfunk_core::transport::set_media_qos(&sock, punktfunk_core::transport::MediaClass::Audio); // The client pings the audio port (~every 500ms) so we learn where to send. sock.set_read_timeout(Some(Duration::from_secs(10)))?; tracing::info!(port = AUDIO_PORT, "audio: awaiting client ping"); @@ -269,6 +269,10 @@ fn run( .context("audio: no client ping within 10s")?; sock.connect(client) .context("connect client audio endpoint")?; + // Opt-in DSCP/QoS-tag this as the audio class (PUNKTFUNK_DSCP=1); the guard keeps the + // Windows qWAVE flow alive for the whole stream (this function's scope IS the stream). + let _qos_flow = + punktfunk_core::transport::set_media_qos(&sock, punktfunk_core::transport::MediaClass::Audio); tracing::info!(%client, "audio: client endpoint learned"); // Reuse the persistent capturer when its channel count still matches (drain stale diff --git a/crates/punktfunk-host/src/gamestream/stream.rs b/crates/punktfunk-host/src/gamestream/stream.rs index 99bab24b..efa75654 100644 --- a/crates/punktfunk-host/src/gamestream/stream.rs +++ b/crates/punktfunk-host/src/gamestream/stream.rs @@ -95,10 +95,10 @@ fn run( encode::validate_dimensions(cfg.codec, cfg.width, cfg.height) .context("client-requested video mode")?; let sock = UdpSocket::bind(("0.0.0.0", VIDEO_PORT)).context("bind video UDP")?; - // Grow SO_SNDBUF/RCVBUF (avoid host-side ENOBUFS at high bitrate) like the native plane, and - // opt-in DSCP/QoS-tag this as the video class (PUNKTFUNK_DSCP=1). + // Grow SO_SNDBUF/RCVBUF (avoid host-side ENOBUFS at high bitrate) like the native plane. + // The opt-in DSCP/QoS tag happens after connect below (Windows qWAVE derives the flow from + // the connected 5-tuple). punktfunk_core::transport::grow_socket_buffers(&sock); - punktfunk_core::transport::set_media_qos(&sock, punktfunk_core::transport::MediaClass::Video); // The client pings the video port so we learn where to send; it re-pings until video // flows, so a missed early ping is fine. sock.set_read_timeout(Some(Duration::from_secs(10)))?; @@ -112,6 +112,10 @@ fn run( .context("video: no client ping within 10s")?; sock.connect(client) .context("connect client video endpoint")?; + // Opt-in DSCP/QoS-tag this as the video class (PUNKTFUNK_DSCP=1); the guard keeps the + // Windows qWAVE flow alive for the whole stream (this function's scope IS the stream). + let _qos_flow = + punktfunk_core::transport::set_media_qos(&sock, punktfunk_core::transport::MediaClass::Video); tracing::info!(%client, "video: client endpoint learned"); // Short label for web-console stats captures: the client's peer IP. let client_label = client.ip().to_string();