io/pipeline: replace polling-send with kernel-wakeup channel

The halt-aware send loop polled try_send on a 50 ms sleep slice when
the channel was full. That capped producer throughput at 1 / 50 ms =
20 frames/sec ≈ 1 MB/s at typical PES frame sizes — way below NFS,
let alone local SSD/NVMe. Multi-day diagnostic 2026-05-13/14 surfaced
it as the root cause of the 18 → 3 MB/s mux throughput regression on
0.21.x.

Replaced std::sync::mpsc::sync_channel with crossbeam_channel::bounded
and switched send_with_halt to send_timeout. Producer now BLOCKS on
consumer drain (kernel-wakeup) instead of polling — the timeout slice
(250 ms) only fires when the producer is waiting for stop signal
observation, never on the happy path.

Result: no throughput cap from this primitive at any medium speed.
Mux is bounded by actual storage / network bandwidth, not by our
channel implementation.

Same fix needs to land in autorip's mux.rs producer-side
sync_channel (separate sibling commit).

Documented in freemkv-private/memory/
feedback_send_with_halt_poll_throttle.md.

Cargo.toml

@@ -28,6 +28,12 @@ base64 = "0.22.1"
 # under CLAUDE.md ("Acceptable strings: debug/trace logging"). Consumers (autorip)
 # wire a tracing subscriber and pipe events into the JSONL debug log.
 tracing = "0.1"
+# Bounded MPSC channel with kernel-wakeup send_timeout. Used by `io::pipeline`
+# so the halt-aware send/finish loops can BLOCK on consumer drain instead of
+# polling — the 50 ms poll cadence of the previous mpsc-based impl capped mux
+# throughput at ~1 MB/s (see freemkv-private/memory/
+# feedback_send_with_halt_poll_throttle.md, 0.21.7).
+crossbeam-channel = "0.5"
 
 [target.'cfg(target_os = "linux")'.dependencies]
 libc = "0.2"

src/io/pipeline.rs

@@ -34,10 +34,11 @@
 //! consumer lag detection). This is critical for diagnosing stalls.
 
 use std::io;
-use std::sync::mpsc::{SyncSender, TrySendError, sync_channel};
 use std::thread::{self, JoinHandle};
 use std::time::{Duration, Instant};
 
+use crossbeam_channel::{Sender, TrySendError, bounded};
+
 use crate::error::Error;
 use crate::halt::Halt;
 
@@ -57,12 +58,21 @@ pub const JOIN_TIMEOUT_SECS: u64 = 600;
 /// responsive across both primitives.
 const POLL_INTERVAL: Duration = Duration::from_millis(250);
 
-/// Polling slice for the halt-aware send loop. Smaller than
+/// Halt-check cadence for the send loop. Producer blocks on
-/// [`POLL_INTERVAL`] because send latency tolerance is much lower —
+/// [`crossbeam_channel::Sender::send_timeout`] for this slice — the
-/// frames must move through the channel at hundreds-of-Hz on the
+/// kernel wakes it the instant the consumer drains a slot, so on the
-/// happy path; 50 ms keeps backpressure-driven wakeups fine-grained
+/// happy path there's no throughput cap from this primitive at all
-/// without busy-looping.
+/// (the cap is whatever the underlying medium can sustain). When the
-const SEND_POLL_INTERVAL: Duration = Duration::from_millis(50);
+/// consumer is genuinely wedged, the timeout fires every 250 ms and
+/// the producer checks the halt token; that's the latency a stop
+/// request will observe.
+///
+/// 0.21.7 replaced an old `std::sync::mpsc::sync_channel` + 50 ms
+/// `thread::sleep` polling loop that capped mux throughput at
+/// ~20 frames/sec ≈ 1 MB/s on saturated channels. See
+/// freemkv-private/memory/feedback_send_with_halt_poll_throttle.md
+/// for the multi-day diagnostic that surfaced it.
+const SEND_HALT_CHECK_INTERVAL: Duration = Duration::from_millis(250);
 
 /// Check if verbose debug logging is enabled via FREEMKV_DEBUG env var.
 pub fn debug_enabled() -> bool {
@@ -135,7 +145,7 @@ pub trait Sink<I>: Send + 'static {
 /// Bounded producer/consumer pipeline. Holds the producer-side
 /// channel and the consumer thread's join handle.
 pub struct Pipeline<I: Send + 'static, R: Send + 'static> {
-    tx: SyncSender<I>,
+    tx: Sender<I>,
     handle: JoinHandle<Result<R, Error>>,
 }
 
@@ -170,7 +180,7 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
         depth: usize,
         sink: S,
     ) -> Result<Self, Error> {
-        let (tx, rx) = sync_channel::<I>(depth);
+        let (tx, rx) = bounded::<I>(depth);
         let handle = thread::Builder::new()
             .name(name.into())
             .spawn(move || -> Result<R, Error> {
@@ -281,14 +291,20 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
     /// `Err`. Useful for best-effort signalling (e.g. sweep's
     /// throttled `StatsRequest`) where dropping the message is
     /// preferable to blocking the producer.
-    pub fn try_send(&self, item: I) -> Result<(), std::sync::mpsc::TrySendError<I>> {
+    pub fn try_send(&self, item: I) -> Result<(), TrySendError<I>> {
         self.tx.try_send(item)
     }
 
     /// Halt-aware bounded variant of [`Pipeline::send`].
     ///
-    /// Polls `try_send` on a 50 ms slice. Between slices, checks
+    /// Uses [`crossbeam_channel::Sender::send_timeout`] so the producer
-    /// (1) the [`Halt`] token and (2) the per-call `deadline`. Returns:
+    /// thread BLOCKS on consumer drain (kernel-wakeup) rather than
+    /// polling. The timeout slice is just the halt-observation cadence
+    /// ([`SEND_HALT_CHECK_INTERVAL`]) — on the happy path the producer
+    /// wakes the instant the consumer drains a slot, so there is no
+    /// throughput cap from this primitive at any medium speed.
+    ///
+    /// Returns:
     ///
     /// - `Ok(())` once the item lands in the channel.
     /// - `Err(item)` if the consumer disconnected, the halt fired, or
@@ -302,36 +318,45 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
     /// Unlike [`Pipeline::send`], this never blocks the producer
     /// thread inside an unkillable `mpsc::send` — if the consumer is
     /// wedged inside an unkillable syscall, the producer can still
-    /// observe `/api/stop` and unwind.
+    /// observe `/api/stop` and unwind within
+    /// [`SEND_HALT_CHECK_INTERVAL`].
     pub fn send_with_halt(&self, item: I, halt: &Halt, deadline: Duration) -> Result<(), I> {
+        use crossbeam_channel::SendTimeoutError;
         let end = Instant::now() + deadline;
         let mut pending = item;
         loop {
-            match self.tx.try_send(pending) {
+            // Pre-check the cheap exit conditions before parking.
-                Ok(()) => return Ok(()),
+            if halt.is_cancelled() {
-                Err(TrySendError::Full(returned)) => {
+                if debug_enabled() {
-                    pending = returned;
+                    tracing::debug!(
-                    if halt.is_cancelled() {
+                        "Pipeline send_with_halt: halt observed, returning item={}",
-                        if debug_enabled() {
+                        std::any::type_name::<I>()
-                            tracing::debug!(
+                    );
-                                "Pipeline send_with_halt: halt observed, returning item={}",
-                                std::any::type_name::<I>()
-                            );
-                        }
-                        return Err(pending);
-                    }
-                    if Instant::now() >= end {
-                        if debug_enabled() {
-                            tracing::debug!(
-                                "Pipeline send_with_halt: deadline elapsed, returning item={}",
-                                std::any::type_name::<I>()
-                            );
-                        }
-                        return Err(pending);
-                    }
-                    thread::sleep(SEND_POLL_INTERVAL);
                 }
-                Err(TrySendError::Disconnected(returned)) => {
+                return Err(pending);
+            }
+            let now = Instant::now();
+            if now >= end {
+                if debug_enabled() {
+                    tracing::debug!(
+                        "Pipeline send_with_halt: deadline elapsed, returning item={}",
+                        std::any::type_name::<I>()
+                    );
+                }
+                return Err(pending);
+            }
+            // Wait for space-available or halt-check tick, whichever
+            // is sooner. Crossbeam's send_timeout is kernel-wakeup
+            // based: the consumer's recv on a saturated channel
+            // signals this thread the moment a slot opens up.
+            let slice = SEND_HALT_CHECK_INTERVAL.min(end.saturating_duration_since(now));
+            match self.tx.send_timeout(pending, slice) {
+                Ok(()) => return Ok(()),
+                Err(SendTimeoutError::Timeout(returned)) => {
+                    pending = returned;
+                    // loop: re-check halt + deadline, then park again
+                }
+                Err(SendTimeoutError::Disconnected(returned)) => {
                     if debug_enabled() {
                         tracing::debug!(
                             "Pipeline send_with_halt: consumer disconnected, item={}",