From c645eb2ca3ee1d1c8eb5c57188e67bc1a0c2ef4e Mon Sep 17 00:00:00 2001 From: "Aaron D. Lee" Date: Fri, 3 Jul 2026 21:58:49 -0400 Subject: [PATCH] =?UTF-8?q?feat(binary):=20MediaThread=20=E2=80=94=20dedic?= =?UTF-8?q?ated=20std::thread=20for=20the=2020ms=20loop=20(slice-4=20?= =?UTF-8?q?=C2=A74)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ARCHITECTURE.md mandate ("never the shared tokio pool") finally landed. One std::thread owns all RtcSessions exclusively; axum routes via command channel (Register/AcceptOffer/Delete/Shutdown). The Reflex wrapper is wired here on Connected via RtcSession::set_pipe. loop_driver + rtc_session untouched (seam holds). Signed-off-by: Aaron D. Lee --- crates/rutster/src/lib.rs | 1 + crates/rutster/src/media_thread.rs | 308 +++++++++++++++++++++++++++++ 2 files changed, 309 insertions(+) create mode 100644 crates/rutster/src/media_thread.rs diff --git a/crates/rutster/src/lib.rs b/crates/rutster/src/lib.rs index 6f98e10..0d89088 100644 --- a/crates/rutster/src/lib.rs +++ b/crates/rutster/src/lib.rs @@ -22,6 +22,7 @@ //! - [slice-1 spec §4](../../docs/superpowers/specs/2026-06-28-slice-1-webrtc-loopback-design.md) //! - [ARCHITECTURE.md](../../docs/ARCHITECTURE.md) — fused vertical. +pub mod media_thread; pub mod routes; pub mod session_map; pub mod tap_engine; diff --git a/crates/rutster/src/media_thread.rs b/crates/rutster/src/media_thread.rs new file mode 100644 index 0000000..f5fb155 --- /dev/null +++ b/crates/rutster/src/media_thread.rs @@ -0,0 +1,308 @@ +//! # MediaThread — the dedicated 20ms media loop on a std::thread +//! (slice-4 spec §2.2, §4) +//! +//! ARCHITECTURE.md mandates "dedicated timing threads, not the shared +//! tokio pool." slice-1 ran the poll on tokio as an acknowledged +//! deviation; slice-4 graduates it. ONE `std::thread::spawn` at binary +//! startup owns `HashMap` exclusively; all access +//! from axum is via a command channel. The 10ms meta-tick between loop +//! iterations is `std::thread::sleep(Duration::from_millis(10))`. +//! +//! # Why one thread, not per-session +//! +//! Spearhead scale (see slice-4 spec §6.3). The command-channel seam +//! makes the later threadpool-shard graduation localized. +//! +//! # The seam (loop_driver + rtc_session byte-identical) +//! +//! `MediaThread` calls `RtcSession::run_poll_once(now)` — the unchanged +//! `loop_driver::drive`. The `Reflex` wrapper is wired in +//! here on the `Connected` transition (via `RtcSession::set_pipe`), not +//! inside `rtc_session.rs`. The seam holds. + +use std::collections::HashMap; +use std::thread::JoinHandle; +use std::time::{Duration, Instant}; + +use rutster_call_model::ChannelId; +use rutster_media::RtcSession; +use tokio::sync::{mpsc, oneshot}; +use tracing::{debug, info}; + +use crate::tap_engine::spawn_tap_engine; + +/// The 10ms meta-tick. Finer than the 20ms outbound encode tick so str0m's +/// `Timeout` outputs are honored promptly. +const META_TICK: Duration = Duration::from_millis(10); + +/// Capacity for the command channel from axum to the media thread. +const CMD_CHANNEL_CAPACITY: usize = 64; + +/// Commands axum sends to the media thread (cold-path only — NEVER on +/// the 20ms tick). The thread owns RtcSessions exclusively; this is the +/// ONLY entry point for axum-side mutation. +#[derive(Debug)] +pub enum MediaCmd { + /// Construct a fresh RtcSession, store it under a new ChannelId, reply. + /// The thread constructs RtcSession::new() (keeps all RtcSession + /// construction on the thread that owns it). + Register { + tap_url: url::Url, + reply: oneshot::Sender>, + }, + /// Accept a browser SDP offer on the session's behalf, reply with the + /// SDP answer (cold-path — the axum POST /v1/sessions/{id}/office handler). + AcceptOffer { + id: ChannelId, + sdp: String, + reply: oneshot::Sender>, + }, + /// Tear down a session — fires close_tx + bounded-await the engine task + /// (750ms cap), then removes the entry. + Delete { + id: ChannelId, + reply: oneshot::Sender<()>, + }, + /// Graceful shutdown — drain + drop + join. + Shutdown { reply: oneshot::Sender<()> }, +} + +/// The handle returned to the binary. Clone the `cmd_tx` per-session; +/// the `JoinHandle` is dropped on shutdown to detach (the binary's +/// graceful-shutdown signal fires `Shutdown` first). +pub struct MediaThread { + cmd_tx: mpsc::Sender, + join: Option>, +} + +impl MediaThread { + /// Return a clone of the command-channel sender, so the binary (axum + /// routes / `AppState`) can issue cold-path commands to the thread. + pub fn cmd_tx(&self) -> mpsc::Sender { + self.cmd_tx.clone() + } + + /// Spawn the dedicated media thread. Captures a `tokio::runtime::Handle` + /// so the thread can `handle.spawn(spawn_tap_engine(...))` on the + /// `Connected` transition. The thread owns `HashMap` exclusively. + /// + /// Returns `std::io::Error` if the OS cannot create the std::thread + /// (for example: the process has hit its thread limit). Thread creation + /// is part of the cold startup path, not the 20ms media loop, so the + /// error is propagated for the binary to decide whether to abort, retry, + /// or degrade. Inside the loop we still follow the "drop + observe, don't + /// crash" policy. + pub fn spawn( + default_tap_url: url::Url, + tokio_handle: tokio::runtime::Handle, + ) -> Result { + let (cmd_tx, cmd_rx) = mpsc::channel(CMD_CHANNEL_CAPACITY); + let join = std::thread::Builder::new() + .name("rutster-media".into()) + .spawn(move || { + run_media_thread(cmd_rx, default_tap_url, tokio_handle); + })?; + Ok(Self { + cmd_tx, + join: Some(join), + }) + } + + /// Graceful shutdown — drains commands + joins the thread. + pub fn shutdown(mut self) { + let (reply, rx) = oneshot::channel(); + let _ = self.cmd_tx.blocking_send(MediaCmd::Shutdown { reply }); + let _ = rx.blocking_recv(); + if let Some(join) = self.join.take() { + let _ = join.join(); + } + } +} + +impl Drop for MediaThread { + fn drop(&mut self) { + if let Some(join) = self.join.take() { + // Best-effort: if shutdown wasn't called explicitly, just + // detach. The thread will exit when cmd_rx is dropped. + // (We don't block on join in Drop — that could deadlock + // if the thread is mid-call into a tokio runtime handle + // that's being torn down.) + debug!(name = ?join.thread().name(), "media thread detached on drop"); + } + } +} + +/// The per-session state owned by the media thread. +struct ThreadSession { + rtc: RtcSession, + /// `Some` only after the `Connected` transition spawns the TapEngine + /// + wires the `Reflex` wrapper. + tap_conn: Option, +} + +fn run_media_thread( + mut cmd_rx: mpsc::Receiver, + default_tap_url: url::Url, + tokio_handle: tokio::runtime::Handle, +) { + let mut sessions: HashMap = HashMap::new(); + info!("media thread started"); + + loop { + // === Step 1: drain ALL pending commands (cold path) BEFORE ticking. === + while let Ok(cmd) = cmd_rx.try_recv() { + match cmd { + MediaCmd::Register { tap_url, reply } => match RtcSession::new() { + Ok(session) => { + let id = session.channel_id(); + sessions.insert( + id, + ThreadSession { + rtc: session, + tap_conn: None, + }, + ); + let _ = reply.send(Ok(id)); + debug!(channel_id = %id, %tap_url, "session registered"); + } + Err(e) => { + let _ = reply.send(Err(format!("RtcSession::new: {e}"))); + } + }, + MediaCmd::AcceptOffer { id, sdp, reply } => { + let result = match sessions.get_mut(&id) { + Some(s) => s.rtc.accept_offer(&sdp).map_err(|e| format!("{e}")), + None => Err(format!("session {id} not found")), + }; + let _ = reply.send(result); + } + MediaCmd::Delete { id, reply } => { + if let Some(mut s) = sessions.remove(&id) { + if let Some(mut conn) = s.tap_conn.take() { + let _ = conn.close_tx.send(()); + let teardown = tokio_handle.block_on(tokio::time::timeout( + Duration::from_millis(750), + &mut conn.join, + )); + match teardown { + Ok(Ok(())) => { + info!(channel_id = %id, "tap engine torn down via Delete (graceful)"); + } + _ => { + conn.join.abort(); + info!(channel_id = %id, "tap engine torn down via Delete (abort after timeout)"); + } + } + } + s.rtc.channel.tap = None; + s.rtc.channel.state = rutster_call_model::ChannelState::Closed; + } + let _ = reply.send(()); + } + MediaCmd::Shutdown { reply } => { + info!( + "media thread shutdown; dropping {} sessions", + sessions.len() + ); + sessions.clear(); + let _ = reply.send(()); + return; + } + } + } + + // === Step 2: the 10ms meta-tick over all sessions. === + let now = Instant::now(); + let mut closed_ids: Vec = Vec::new(); + for (id, session) in sessions.iter_mut() { + // Drain flush side-channel BEFORE run_poll_once (slice-2 §5.3 step 4). + if let Some(conn) = session.tap_conn.as_mut() { + if let Some(rx) = conn.flush_rx.as_mut() { + let mut should_flush = false; + while let Ok(()) = rx.try_recv() { + should_flush = true; + } + if should_flush { + session.rtc.clear_playout_ring(); + } + } + } + let _ = session.rtc.run_poll_once(now); + + // === relay the Connected transition: spawn TapEngine + wire Reflex. === + use rutster_call_model::ChannelState; + if let ChannelState::Connected = session.rtc.channel.state { + if session.rtc.channel.tap.is_none() { + let url = default_tap_url.clone(); + // The media thread owns the advisory channel (multi-producer: + // tokio::sync::mpsc::Sender is Clone). Both the brain's + // advisories (via spawn_tap_engine) AND the local VAD's + // trips (via LocalVadReflex) push to the SAME mpsc; the + // Reflex drains both uniformly. This means spawn_tap_engine + // takes advisory_tx as a PARAMETER (Task 5's signature + // changes: spawn_tap_engine(session_id, tap_url, app_state, + // advisory_tx) — see Task 5's revision note). + let (advisory_tx, advisory_rx) = + mpsc::channel::(16); + let (pipe, conn) = tokio_handle.block_on(async { + spawn_tap_engine( + *id, + url, + crate::session_map::AppState::default(), + advisory_tx.clone(), + ) + }); + let metrics = rutster_media::ReflexMetrics::new(); + // Compose: Reflex (state machine) wrapped by + // LocalVadReflex (primary VAD trigger). Both feed advisory_tx. + let reflex = rutster_media::Reflex::new(pipe, advisory_rx, metrics); + let vad = rutster_media::LocalVadReflex::new(reflex, advisory_tx); + session.rtc.set_pipe(vad); + session.rtc.channel.tap = Some(rutster_call_model::TapHandle::new()); + session.tap_conn = Some(conn); + info!(channel_id = %id, "tap engine + reflex + local VAD wired on Connected"); + continue; + } + } + + if session.rtc.is_closed() { + closed_ids.push(*id); + } + } + for id in closed_ids { + sessions.remove(&id); + debug!(channel_id = %id, "session evicted after close"); + } + + // === Step 3: sleep META_TICK. === + std::thread::sleep(META_TICK); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn media_thread_register_and_shutdown_round_trips() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let handle = rt.handle().clone(); + let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap(); + let thread = MediaThread::spawn(url, handle).expect("media thread spawn in test"); + let (reply, rx) = oneshot::channel(); + thread + .cmd_tx() + .blocking_send(MediaCmd::Register { + tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(), + reply, + }) + .unwrap(); + let id = rx + .blocking_recv() + .expect("register reply") + .expect("session"); + assert_eq!(format!("{}", id).len(), 36, "UUID-shaped ChannelId"); + thread.shutdown(); + } +}