Matthew Jackson
71686f1407
Every other repo's CI now runs clippy with --all-targets. libfreemkv, the crate the other seven build against and the one held up as the reference workflow, was the last one still linting the library only — so its ~3,000 tests, by far the largest body of test code in the project, had never been linted at all. Turning the flag on surfaced 74 findings. Most were mechanical and applied with clippy --fix. The rest, by hand: - Four discarded Results in decrypt.rs. css::descramble_region returns a Result and four CSS tests threw it away, so a descramble that FAILED would have surfaced as a confusing buffer-comparison mismatch instead of the actual error. They expect() now. - A dead `kp` field on the PlantedWalk fixture. The test deliberately asserts Kp as the explicit AES-G3(dk, 1) relation from [C] §3.2.4 rather than against a stored value — its doc comment says so — which makes the field not just unused but a trap: the obvious "fix" of asserting against it would quietly weaken the test to comparing the fixture with itself. Removed. - Two hand-rolled ICB counters in the HD-DVD fixtures, a needless mut, three vec!s that only ever needed arrays, a filter_map whose every arm was Some, and a Vec::new()+push chain. - Doc list indentation in mkv.rs and mp4/read.rs, which was mis-rendering in the generated docs. - A five-[u8; 16]-tuple return type named FourLevelParts. Three lints are allowed at the specific sites, with reasons, because they are wrong for this domain: the underscores in the bitstream-header literals mark BITFIELD boundaries, not digit groups, so regrouping them uniformly would satisfy the lint by destroying the only thing they encode; and in three table-validation loops the loop variable is the domain value under test (a DTS SFREQ code, an AMODE value, a palette entry number), which is what the assertion messages name.
663 lines
30 KiB
Rust
663 lines
30 KiB
Rust
//! `DemuxThread` — runs the read+decrypt+demux pipeline on a
|
|
//! dedicated thread, feeding completed `PesPacket` batches to the
|
|
//! caller via a bounded channel.
|
|
//!
|
|
//! ## Why a second worker thread
|
|
//!
|
|
//! With [`crate::sector::PrefetchedSectorSource`] alone, read+decrypt
|
|
//! already runs on a producer thread; the *consumer* (main) thread
|
|
//! still serialises `ts_demuxer.feed` (M2TS parsing) with the codec
|
|
//! parsers. Profiling showed feed at ~37 % and
|
|
//! codec parse at ~44 % of consumer wall time — i.e. feed is heavy
|
|
//! enough that pipelining it with parse pays for itself.
|
|
//!
|
|
//! Splitting them: feed runs in [`DemuxThread`]; the consumer thread
|
|
//! receives `Vec<PesPacket>` batches and runs codec parse + frame
|
|
//! emission only. Total throughput becomes `1/max(feed, parse)`
|
|
//! instead of `1/(feed + parse)`.
|
|
//!
|
|
//! ## Lifecycle
|
|
//!
|
|
//! [`DemuxThread::spawn_zero_copy`] consumes the prefetch channels and
|
|
//! the demuxer state, returning a handle plus a `Receiver<DemuxBatch>`.
|
|
//! Dropping the handle closes the channel which signals the worker to
|
|
//! exit; the join in `Drop::drop` blocks until the worker observes
|
|
//! channel closure and returns (no timeout — a wedged downstream would
|
|
//! block the drop until it releases the channel).
|
|
|
|
use crate::halt::Halt;
|
|
use crossbeam_channel::{Receiver, Sender, bounded};
|
|
use std::thread::JoinHandle;
|
|
|
|
/// Output channel depth. Two batches in flight keeps the consumer
|
|
/// (codec parser) busy without piling up demuxed bytes if it stalls.
|
|
const DEMUX_CHANNEL_DEPTH: usize = 2;
|
|
|
|
/// One demuxed batch flowing from the demux thread to the consumer.
|
|
pub enum DemuxBatch {
|
|
/// Successfully demuxed PesPackets — non-empty.
|
|
Ts(Vec<super::ts::PesPacket>),
|
|
Ps(Vec<super::ps::PsPacket>),
|
|
/// Underlying reader returned an error. Terminal.
|
|
Err(std::io::Error),
|
|
/// Explicit clean-completion sentinel. The worker sends this as its
|
|
/// LAST message on every non-error exit (input exhausted, or halt
|
|
/// cancelled) so the consumer can distinguish a normal end-of-stream
|
|
/// from a bare channel disconnection. A worker that panics mid-stream
|
|
/// drops `tx` without sending this, so the consumer sees `RecvError`
|
|
/// and reports the panic rather than silently truncating output.
|
|
Eof,
|
|
}
|
|
|
|
/// Spawned demux thread. Drop joins.
|
|
///
|
|
/// In zero-copy mode the thread also owns an opaque
|
|
/// `producer_shell: Option<Box<dyn Send>>` — the join handle of the
|
|
/// upstream producer (sector or byte prefetcher). Dropping the
|
|
/// `DemuxThread` runs the shell's `Drop`, which joins the producer.
|
|
/// `Box<dyn Send>` rather than a concrete type so the same demux
|
|
/// worker can be wired behind either prefetcher kind.
|
|
pub struct DemuxThread {
|
|
handle: Option<JoinHandle<()>>,
|
|
#[allow(dead_code)]
|
|
producer_shell: Option<Box<dyn Send>>,
|
|
}
|
|
|
|
impl DemuxThread {
|
|
/// Spawn the demux thread. Instead of taking a `SectorSource` and
|
|
/// memcpy-ing through its `read_sectors` API, this constructor
|
|
/// consumes the prefetch channels directly: filled buffers come
|
|
/// in via `prefetch_rx`, the demux thread feeds them, then
|
|
/// returns them to `recycle_tx` for the producer to re-fill.
|
|
/// Eliminates the 16 MiB memcpy per batch that the SectorSource
|
|
/// adapter incurred (and, with the producer-side recycling pool,
|
|
/// also eliminates the per-batch heap alloc / cross-thread free
|
|
/// that was costing 40 %+ of demux-thread time before).
|
|
///
|
|
/// `producer_shell` is an opaque handle whose only purpose is to
|
|
/// outlive the demux thread and join the upstream producer when
|
|
/// dropped. Both
|
|
/// [`crate::sector::PrefetchedSectorSource::into_channels`] and
|
|
/// [`crate::io::byte_prefetcher::BytePrefetcher::into_channels`]
|
|
/// hand back a shell that fits — pass either.
|
|
pub fn spawn_zero_copy<S: Send + 'static>(
|
|
prefetch_rx: Receiver<std::io::Result<Vec<u8>>>,
|
|
recycle_tx: Sender<Vec<u8>>,
|
|
producer_shell: S,
|
|
halt: Option<Halt>,
|
|
ts: Option<super::ts::TsDemuxer>,
|
|
ps: Option<super::ps::PsDemuxer>,
|
|
) -> crate::error::Result<(Self, Receiver<DemuxBatch>)> {
|
|
let (tx, rx) = bounded::<DemuxBatch>(DEMUX_CHANNEL_DEPTH);
|
|
let mut ts = ts;
|
|
let mut ps = ps;
|
|
|
|
// SAFETY (no teardown deadlock on spawn failure): the worker closure is
|
|
// `move`, so it OWNS `prefetch_rx` and `recycle_tx`. If `spawn` fails it
|
|
// consumes and drops the closure, which drops those channel ends — so the
|
|
// upstream producer observes disconnection and exits on its own BEFORE we
|
|
// join it. `producer_shell` (whose Drop joins the producer) is NOT captured
|
|
// by the closure, so dropping it on the Err path below joins a producer that
|
|
// has already exited → non-blocking.
|
|
let spawn_result = std::thread::Builder::new()
|
|
.name("freemkv-demux".into())
|
|
.spawn(move || {
|
|
let prof = std::env::var_os("FREEMKV_PROFILE").is_some();
|
|
let mut prof_started = std::time::Instant::now();
|
|
let mut prof_last_dump = prof_started;
|
|
let mut prof_read_ns: u128 = 0;
|
|
let mut prof_feed_ns: u128 = 0;
|
|
let mut prof_bytes: u64 = 0;
|
|
// Liveness heartbeat: the feed loop blocks on prefetch_rx.recv()
|
|
// and on tx.send(); a stuck upstream/downstream shows up as the
|
|
// beat going silent. Total is unknown for a stream, so `pos` is
|
|
// cumulative bytes fed.
|
|
let mut hb = crate::progress::Heartbeat::new("demux_feed");
|
|
let mut fed_bytes: u64 = 0;
|
|
loop {
|
|
hb.tick(fed_bytes, 0);
|
|
if halt.as_ref().map(|h| h.is_cancelled()).unwrap_or(false) {
|
|
// Caller-initiated stop is a clean termination —
|
|
// send the Eof sentinel so the consumer doesn't
|
|
// mistake it for a worker panic.
|
|
let _ = tx.send(DemuxBatch::Eof);
|
|
return;
|
|
}
|
|
let t0 = if prof {
|
|
Some(std::time::Instant::now())
|
|
} else {
|
|
None
|
|
};
|
|
let buf = match prefetch_rx.recv() {
|
|
Ok(Ok(b)) => b,
|
|
Ok(Err(e)) => {
|
|
let _ = tx.send(DemuxBatch::Err(e));
|
|
return;
|
|
}
|
|
Err(_) => break, // producer done → EOF
|
|
};
|
|
let t1 = if prof {
|
|
Some(std::time::Instant::now())
|
|
} else {
|
|
None
|
|
};
|
|
let n = buf.len();
|
|
// Source byte offset of this buffer's first byte = bytes fed
|
|
// so far. Threaded into the demuxer so every PES it cuts is
|
|
// stamped with its SourcePos (carried, not reconstructed).
|
|
let buf_base = fed_bytes;
|
|
fed_bytes += n as u64;
|
|
if let Some(ref mut d) = ts {
|
|
let pkts = d.feed_at(buf_base, &buf);
|
|
let t2 = if prof {
|
|
Some(std::time::Instant::now())
|
|
} else {
|
|
None
|
|
};
|
|
// Recycle the buffer back to the producer
|
|
// before pushing the demuxed packets. If the
|
|
// recycle channel is closed the producer has
|
|
// exited; we drop the buffer and continue.
|
|
let _ = recycle_tx.send(buf);
|
|
// Always send the batch — even when empty (null /
|
|
// untracked PIDs only). send() is how we detect an early
|
|
// consumer disconnect; on mostly-null extents spanning
|
|
// gigabytes of disc the batch can stay empty for a long
|
|
// time, and skipping empty sends would hide the
|
|
// disconnect until a (possibly never-arriving) non-empty
|
|
// batch. An empty batch yields no frames downstream.
|
|
if tx.send(DemuxBatch::Ts(pkts)).is_err() {
|
|
return;
|
|
}
|
|
if prof {
|
|
prof_read_ns += t1.unwrap().duration_since(t0.unwrap()).as_nanos();
|
|
prof_feed_ns += t2.unwrap().duration_since(t1.unwrap()).as_nanos();
|
|
prof_bytes += n as u64;
|
|
let now = std::time::Instant::now();
|
|
if now.duration_since(prof_last_dump)
|
|
>= std::time::Duration::from_secs(5)
|
|
{
|
|
let el = now.duration_since(prof_started).as_millis().max(1);
|
|
let mbps = prof_bytes as u128 * 1000 / 1_000_000 / el;
|
|
tracing::debug!(
|
|
target: "mux",
|
|
"[demux] elapsed={}ms in={}MB/s read={}% feed={}%",
|
|
el,
|
|
mbps,
|
|
prof_read_ns / 10_000 / el,
|
|
prof_feed_ns / 10_000 / el,
|
|
);
|
|
prof_last_dump = now;
|
|
prof_started = now;
|
|
prof_read_ns = 0;
|
|
prof_feed_ns = 0;
|
|
prof_bytes = 0;
|
|
}
|
|
}
|
|
} else if let Some(ref mut d) = ps {
|
|
let pkts = d.feed_at(buf_base, &buf);
|
|
let _ = recycle_tx.send(buf);
|
|
// Always send (even empty) — same early-disconnect
|
|
// detection rationale as the TS branch above.
|
|
if tx.send(DemuxBatch::Ps(pkts)).is_err() {
|
|
return;
|
|
}
|
|
} else {
|
|
let _ = recycle_tx.send(buf);
|
|
// No demuxer (a BdTs title with zero streams): still send
|
|
// an empty batch so an early consumer disconnect is
|
|
// detected here too, exactly like the ts/ps branches above.
|
|
// Without it this worker reads the whole disc even after
|
|
// the consumer has dropped.
|
|
if tx.send(DemuxBatch::Ts(Vec::new())).is_err() {
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
// Flush tail packets at EOF.
|
|
if let Some(ref mut d) = ts {
|
|
let tail = d.flush();
|
|
if !tail.is_empty() {
|
|
let _ = tx.send(DemuxBatch::Ts(tail));
|
|
}
|
|
} else if let Some(ref mut d) = ps {
|
|
let tail = d.flush();
|
|
if !tail.is_empty() {
|
|
let _ = tx.send(DemuxBatch::Ps(tail));
|
|
}
|
|
}
|
|
// Clean end-of-stream sentinel. Reaching here means no
|
|
// panic occurred; a panic during `feed`/`flush` skips
|
|
// this and drops `tx`, which the consumer reads as an
|
|
// error rather than a clean EOF.
|
|
let _ = tx.send(DemuxBatch::Eof);
|
|
});
|
|
|
|
let handle = match spawn_result {
|
|
Ok(h) => h,
|
|
Err(e) => {
|
|
// `prefetch_rx`/`recycle_tx` were moved into the (now-dropped)
|
|
// failed spawn closure, so the producer already sees disconnection.
|
|
// Dropping producer_shell here joins that already-exiting producer.
|
|
drop(producer_shell);
|
|
return Err(crate::error::Error::IoError { source: e });
|
|
}
|
|
};
|
|
|
|
Ok((
|
|
Self {
|
|
handle: Some(handle),
|
|
producer_shell: Some(Box::new(producer_shell)),
|
|
},
|
|
rx,
|
|
))
|
|
}
|
|
}
|
|
|
|
impl Drop for DemuxThread {
|
|
fn drop(&mut self) {
|
|
if let Some(h) = self.handle.take() {
|
|
let _ = h.join();
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::halt::Halt;
|
|
use crossbeam_channel::bounded;
|
|
use std::time::Duration;
|
|
|
|
/// Build one 192-byte BD-TS packet on `pid` carrying a complete PES with
|
|
/// a `00 00 01 E0` start, hdr_len 0, then `payload` as ES. The TS payload
|
|
/// region after the PES header is padded with a stuffing adaptation field
|
|
/// so `payload` is the exact ES (no zero padding the unbounded PES would
|
|
/// absorb). ISO 13818-1 packet layout: sync 0x47 at TS offset 0 (BD off 4).
|
|
fn bdts_pes_packet(pid: u16, payload: &[u8]) -> Vec<u8> {
|
|
const SYNC: u8 = 0x47;
|
|
use crate::consts::TS_PAYLOAD_BYTES;
|
|
let mut pes = vec![0x00, 0x00, 0x01, 0xE0, 0x00, 0x00, 0x80, 0x00, 0x00];
|
|
pes.extend_from_slice(payload);
|
|
assert!(pes.len() <= TS_PAYLOAD_BYTES);
|
|
let mut pkt = vec![0u8; 192];
|
|
pkt[4] = SYNC;
|
|
pkt[5] = (((pid >> 8) as u8) & 0x1F) | 0x40; // PUSI
|
|
pkt[6] = (pid & 0xFF) as u8;
|
|
let pad = TS_PAYLOAD_BYTES - pes.len();
|
|
if pad == 0 {
|
|
pkt[7] = 0x10; // payload only
|
|
pkt[8..8 + pes.len()].copy_from_slice(&pes);
|
|
} else {
|
|
pkt[7] = 0x30; // AF + payload
|
|
let af_field_len = pad - 1;
|
|
pkt[8] = af_field_len as u8;
|
|
if af_field_len >= 1 {
|
|
pkt[9] = 0x00; // flags
|
|
for b in pkt.iter_mut().skip(10).take(af_field_len - 1) {
|
|
*b = 0xFF;
|
|
}
|
|
}
|
|
let off = 8 + pad;
|
|
pkt[off..off + pes.len()].copy_from_slice(&pes);
|
|
}
|
|
pkt
|
|
}
|
|
|
|
/// Drain a receiver into a Vec, blocking up to `budget` total.
|
|
fn collect_batches(rx: &Receiver<DemuxBatch>, budget: Duration) -> Vec<DemuxBatch> {
|
|
let mut out = Vec::new();
|
|
let deadline = std::time::Instant::now() + budget;
|
|
loop {
|
|
let now = std::time::Instant::now();
|
|
if now >= deadline {
|
|
break;
|
|
}
|
|
match rx.recv_timeout(deadline - now) {
|
|
Ok(b) => {
|
|
let is_terminal = matches!(b, DemuxBatch::Eof | DemuxBatch::Err(_));
|
|
out.push(b);
|
|
if is_terminal {
|
|
break;
|
|
}
|
|
}
|
|
Err(_) => break,
|
|
}
|
|
}
|
|
out
|
|
}
|
|
|
|
#[test]
|
|
fn clean_eof_sentinel_sent_after_input_exhausted() {
|
|
// The worker must send exactly one Eof as its LAST message on a
|
|
// normal end-of-stream so the consumer can distinguish clean
|
|
// completion from a panic (which drops tx without Eof).
|
|
let (pf_tx, pf_rx) = bounded::<std::io::Result<Vec<u8>>>(4);
|
|
let (rc_tx, _rc_rx) = bounded::<Vec<u8>>(4);
|
|
let pid = 0x1011;
|
|
let ts = super::super::ts::TsDemuxer::new(&[pid]);
|
|
let (_dt, rx) =
|
|
DemuxThread::spawn_zero_copy(pf_rx, rc_tx, (), None, Some(ts), None).unwrap();
|
|
|
|
pf_tx.send(Ok(bdts_pes_packet(pid, &[0xDE, 0xAD]))).unwrap();
|
|
drop(pf_tx); // producer done → EOF
|
|
|
|
let batches = collect_batches(&rx, Duration::from_secs(5));
|
|
// Last batch must be the Eof sentinel.
|
|
assert!(
|
|
matches!(batches.last(), Some(DemuxBatch::Eof)),
|
|
"stream must terminate with the Eof sentinel"
|
|
);
|
|
// The PES bytes must surface before EOF (the demuxer holds the PES
|
|
// until flush at EOF since there's no following PUSI).
|
|
let saw_pes = batches.iter().any(|b| match b {
|
|
DemuxBatch::Ts(p) => p.iter().any(|pes| pes.data == vec![0xDE, 0xAD]),
|
|
_ => false,
|
|
});
|
|
assert!(saw_pes, "the demuxed PES must be delivered");
|
|
}
|
|
|
|
#[test]
|
|
fn flush_tail_emitted_before_eof() {
|
|
// A PES with no trailing PUSI is only completed by flush() at EOF.
|
|
// The worker must flush after the producer disconnects, emitting the
|
|
// tail PES BEFORE the Eof sentinel — never dropping the last frame.
|
|
let (pf_tx, pf_rx) = bounded::<std::io::Result<Vec<u8>>>(4);
|
|
let (rc_tx, _rc_rx) = bounded::<Vec<u8>>(4);
|
|
let pid = 0x1011;
|
|
let ts = super::super::ts::TsDemuxer::new(&[pid]);
|
|
let (_dt, rx) =
|
|
DemuxThread::spawn_zero_copy(pf_rx, rc_tx, (), None, Some(ts), None).unwrap();
|
|
|
|
pf_tx
|
|
.send(Ok(bdts_pes_packet(pid, &[0x11, 0x22, 0x33])))
|
|
.unwrap();
|
|
drop(pf_tx);
|
|
|
|
let batches = collect_batches(&rx, Duration::from_secs(5));
|
|
// Find the tail PES and the Eof; tail must precede Eof.
|
|
let pes_idx = batches.iter().position(|b| {
|
|
matches!(b, DemuxBatch::Ts(p) if p.iter().any(|x| x.data == vec![0x11, 0x22, 0x33]))
|
|
});
|
|
let eof_idx = batches.iter().position(|b| matches!(b, DemuxBatch::Eof));
|
|
assert!(pes_idx.is_some(), "flushed tail PES delivered");
|
|
assert!(eof_idx.is_some(), "Eof delivered");
|
|
assert!(pes_idx.unwrap() < eof_idx.unwrap(), "tail before Eof");
|
|
}
|
|
|
|
#[test]
|
|
fn halt_cancellation_sends_eof_not_panic() {
|
|
// A caller-initiated halt is a CLEAN termination — the worker must
|
|
// send Eof (not just drop tx), so the consumer doesn't mistake the
|
|
// stop for a worker panic.
|
|
let (pf_tx, pf_rx) = bounded::<std::io::Result<Vec<u8>>>(4);
|
|
let (rc_tx, _rc_rx) = bounded::<Vec<u8>>(4);
|
|
let halt = Halt::new();
|
|
halt.cancel(); // already cancelled before the loop runs
|
|
let ts = super::super::ts::TsDemuxer::new(&[0x1011]);
|
|
let (_dt, rx) =
|
|
DemuxThread::spawn_zero_copy(pf_rx, rc_tx, (), Some(halt), Some(ts), None).unwrap();
|
|
// Keep pf_tx alive so the ONLY exit is the halt path, not producer
|
|
// disconnect.
|
|
let batches = collect_batches(&rx, Duration::from_secs(5));
|
|
drop(pf_tx);
|
|
assert!(
|
|
matches!(batches.last(), Some(DemuxBatch::Eof)),
|
|
"halt cancellation must yield a clean Eof sentinel"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn upstream_error_is_propagated_as_err_terminal() {
|
|
// An error from the prefetch channel must be forwarded as a terminal
|
|
// DemuxBatch::Err — the worker then returns (no Eof after an error).
|
|
let (pf_tx, pf_rx) = bounded::<std::io::Result<Vec<u8>>>(4);
|
|
let (rc_tx, _rc_rx) = bounded::<Vec<u8>>(4);
|
|
let ts = super::super::ts::TsDemuxer::new(&[0x1011]);
|
|
let (_dt, rx) =
|
|
DemuxThread::spawn_zero_copy(pf_rx, rc_tx, (), None, Some(ts), None).unwrap();
|
|
|
|
pf_tx.send(Err(std::io::Error::other("boom"))).unwrap();
|
|
drop(pf_tx);
|
|
|
|
let batches = collect_batches(&rx, Duration::from_secs(5));
|
|
assert!(
|
|
matches!(batches.last(), Some(DemuxBatch::Err(_))),
|
|
"upstream error must terminate the stream with Err"
|
|
);
|
|
// No Eof must follow an Err (the worker returns immediately).
|
|
assert!(
|
|
!batches.iter().any(|b| matches!(b, DemuxBatch::Eof)),
|
|
"Err is terminal; no Eof after it"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn buffers_are_recycled_to_producer() {
|
|
// The worker must return each consumed buffer to recycle_tx so the
|
|
// producer can re-fill it (the zero-copy pool contract). Verify a
|
|
// fed buffer comes back on the recycle channel.
|
|
let (pf_tx, pf_rx) = bounded::<std::io::Result<Vec<u8>>>(4);
|
|
let (rc_tx, rc_rx) = bounded::<Vec<u8>>(4);
|
|
let pid = 0x1011;
|
|
let ts = super::super::ts::TsDemuxer::new(&[pid]);
|
|
let (_dt, _rx) =
|
|
DemuxThread::spawn_zero_copy(pf_rx, rc_tx, (), None, Some(ts), None).unwrap();
|
|
|
|
pf_tx.send(Ok(bdts_pes_packet(pid, &[0xAA]))).unwrap();
|
|
let recycled = rc_rx.recv_timeout(Duration::from_secs(5));
|
|
assert!(recycled.is_ok(), "consumed buffer must be recycled");
|
|
assert_eq!(recycled.unwrap().len(), 192, "the original buffer returned");
|
|
drop(pf_tx);
|
|
}
|
|
|
|
#[test]
|
|
fn ps_path_demuxes_and_eofs() {
|
|
// The PS branch must demux MPEG-2 Program Stream input and also send
|
|
// the Eof sentinel on clean exit. Feed a complete PES + program-end
|
|
// delimiter so the PsDemuxer emits it without waiting for flush.
|
|
let (pf_tx, pf_rx) = bounded::<std::io::Result<Vec<u8>>>(4);
|
|
let (rc_tx, _rc_rx) = bounded::<Vec<u8>>(4);
|
|
let ps = super::super::ps::PsDemuxer::new();
|
|
let (_dt, rx) =
|
|
DemuxThread::spawn_zero_copy(pf_rx, rc_tx, (), None, None, Some(ps)).unwrap();
|
|
|
|
// PES (video 0xE0, bounded length 5) + program-end delimiter.
|
|
let mut buf = vec![
|
|
0x00, 0x00, 0x01, 0xE0, 0x00, 0x05, 0x80, 0x00, 0x00, 0x77, 0x88,
|
|
];
|
|
buf.extend_from_slice(&[0x00, 0x00, 0x01, 0xB9]); // program end
|
|
pf_tx.send(Ok(buf)).unwrap();
|
|
drop(pf_tx);
|
|
|
|
let batches = collect_batches(&rx, Duration::from_secs(5));
|
|
assert!(
|
|
matches!(batches.last(), Some(DemuxBatch::Eof)),
|
|
"PS path sends Eof"
|
|
);
|
|
let saw = batches.iter().any(|b| match b {
|
|
DemuxBatch::Ps(p) => p.iter().any(|x| x.data == vec![0x77, 0x88]),
|
|
_ => false,
|
|
});
|
|
assert!(saw, "PS PES must be demuxed and delivered");
|
|
}
|
|
|
|
#[test]
|
|
fn no_demuxer_configured_still_recycles_and_eofs() {
|
|
// With neither ts nor ps set, the worker must still recycle buffers
|
|
// and terminate with Eof — also forward an empty batch per buffer for disconnect detection.
|
|
let (pf_tx, pf_rx) = bounded::<std::io::Result<Vec<u8>>>(4);
|
|
let (rc_tx, rc_rx) = bounded::<Vec<u8>>(4);
|
|
let (_dt, rx) = DemuxThread::spawn_zero_copy(pf_rx, rc_tx, (), None, None, None).unwrap();
|
|
|
|
pf_tx.send(Ok(vec![0u8; 192])).unwrap();
|
|
assert!(
|
|
rc_rx.recv_timeout(Duration::from_secs(5)).is_ok(),
|
|
"buffer recycled"
|
|
);
|
|
drop(pf_tx);
|
|
|
|
let batches = collect_batches(&rx, Duration::from_secs(5));
|
|
// The no-demuxer branch now forwards an empty Ts batch per buffer for
|
|
// early consumer-disconnect detection (same rationale as the ts/ps
|
|
// branches), then the Eof sentinel.
|
|
assert_eq!(
|
|
batches.len(),
|
|
2,
|
|
"empty Ts disconnect-probe batch, then Eof"
|
|
);
|
|
assert!(matches!(batches[0], DemuxBatch::Ts(ref v) if v.is_empty()));
|
|
assert!(matches!(batches[1], DemuxBatch::Eof));
|
|
}
|
|
|
|
#[test]
|
|
fn empty_batches_are_forwarded_for_disconnect_detection() {
|
|
// The worker forwards EVERY batch, including empty ones, so an early
|
|
// consumer disconnect is detected promptly via `send()` (crossbeam's
|
|
// Sender has no non-destructive disconnect check). An empty batch is
|
|
// harmless downstream: `pump_one_batch` consumes 0 packets and returns
|
|
// Ok(true) — only the explicit `Eof` sentinel ends the stream. A buffer
|
|
// that yields no complete PES therefore produces an empty Ts batch
|
|
// followed by Eof.
|
|
let (pf_tx, pf_rx) = bounded::<std::io::Result<Vec<u8>>>(4);
|
|
let (rc_tx, _rc_rx) = bounded::<Vec<u8>>(4);
|
|
let pid = 0x1011;
|
|
let ts = super::super::ts::TsDemuxer::new(&[pid]);
|
|
let (_dt, rx) =
|
|
DemuxThread::spawn_zero_copy(pf_rx, rc_tx, (), None, Some(ts), None).unwrap();
|
|
|
|
// A non-PUSI packet on a tracked PID with header_remaining 0 and no
|
|
// active PES: process_packet pushes nothing (asm inactive), so feed
|
|
// returns empty and flush also returns empty.
|
|
const SYNC: u8 = 0x47;
|
|
let mut pkt = vec![0u8; 192];
|
|
pkt[4] = SYNC;
|
|
pkt[5] = ((pid >> 8) as u8) & 0x1F; // no PUSI
|
|
pkt[6] = (pid & 0xFF) as u8;
|
|
pkt[7] = 0x10; // payload only
|
|
pf_tx.send(Ok(pkt)).unwrap();
|
|
drop(pf_tx);
|
|
|
|
let batches = collect_batches(&rx, Duration::from_secs(5));
|
|
assert_eq!(batches.len(), 2, "empty Ts batch forwarded, then Eof");
|
|
assert!(matches!(batches[0], DemuxBatch::Ts(ref v) if v.is_empty()));
|
|
assert!(matches!(batches[1], DemuxBatch::Eof));
|
|
}
|
|
|
|
/// Regression: worker must detect consumer disconnect even when every
|
|
/// demux batch is empty (no matching PIDs / null packets).
|
|
///
|
|
/// Before the fix, `tx.send()` was never called for empty batches so the
|
|
/// worker never observed the consumer drop — it would spin through ALL
|
|
/// remaining extents before exiting, causing DemuxThread::drop's join()
|
|
/// to block for minutes on a mostly-untracked disc region.
|
|
///
|
|
/// The watchdog: if the worker doesn't exit within 1 s of the consumer
|
|
/// drop the test fails (rather than hanging forever as the bug would).
|
|
#[test]
|
|
fn worker_exits_promptly_on_consumer_drop_during_empty_batches() {
|
|
// Use an untracked PID so every batch the demuxer produces is empty.
|
|
let tracked_pid = 0x1011u16;
|
|
let untracked_pid = 0x0100u16;
|
|
|
|
const SYNC: u8 = 0x47;
|
|
// Build a non-PUSI continuation packet on the untracked PID so
|
|
// TsDemuxer.feed() returns an empty Vec every call.
|
|
let mut empty_pkt = vec![0u8; 192];
|
|
empty_pkt[4] = SYNC;
|
|
empty_pkt[5] = ((untracked_pid >> 8) as u8) & 0x1F; // no PUSI
|
|
empty_pkt[6] = (untracked_pid & 0xFF) as u8;
|
|
empty_pkt[7] = 0x10; // payload only
|
|
|
|
// Large prefetch channel — enough that the worker will be spinning
|
|
// through empty batches long after the consumer drops.
|
|
let (pf_tx, pf_rx) = bounded::<std::io::Result<Vec<u8>>>(64);
|
|
let (rc_tx, _rc_rx) = bounded::<Vec<u8>>(64);
|
|
let ts = super::super::ts::TsDemuxer::new(&[tracked_pid]);
|
|
let (dt, rx) =
|
|
DemuxThread::spawn_zero_copy(pf_rx, rc_tx, (), None, Some(ts), None).unwrap();
|
|
|
|
// Fill the prefetch channel with empty-batch buffers.
|
|
for _ in 0..64 {
|
|
pf_tx.send(Ok(empty_pkt.clone())).unwrap();
|
|
}
|
|
|
|
// Drop the consumer — the worker should notice during the next
|
|
// empty-batch iteration (is_disconnected() check).
|
|
drop(rx);
|
|
|
|
// Give the worker a generous but bounded window to observe the
|
|
// disconnect and exit. A regression (spin-until-exhaustion) would
|
|
// take >> 1 s; correct behaviour exits almost immediately.
|
|
let join_done = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
|
|
let join_done2 = join_done.clone();
|
|
let watchdog = std::thread::spawn(move || {
|
|
drop(dt); // joins the worker
|
|
join_done2.store(true, std::sync::atomic::Ordering::Relaxed);
|
|
});
|
|
|
|
// Also close the producer so the worker doesn't block on prefetch_rx
|
|
// if somehow is_disconnected is not triggered (belt-and-suspenders).
|
|
drop(pf_tx);
|
|
|
|
watchdog.join().unwrap();
|
|
assert!(
|
|
join_done.load(std::sync::atomic::Ordering::Relaxed),
|
|
"worker must exit promptly after consumer drop during empty batches"
|
|
);
|
|
}
|
|
|
|
/// Regression: on thread-spawn failure the channels must be dropped BEFORE
|
|
/// the producer shell so the upstream producer observes disconnection and
|
|
/// exits, allowing join() to complete without hanging.
|
|
///
|
|
/// A true EAGAIN/pids-limit spawn failure cannot be reliably forced in a
|
|
/// unit test without root or ulimit co-operation, so we test the
|
|
/// drop-order contract directly: a mock shell that panics if join() is
|
|
/// called while either channel end is still open.
|
|
///
|
|
/// The test constructs a `(prefetch_tx, prefetch_rx)` pair where the tx
|
|
/// side is held by a sentinel that stays alive as long as either channel
|
|
/// end is open, then asserts that the sentinel is gone by the time
|
|
/// producer_shell's join logic would run. Because we can't force a real
|
|
/// spawn failure, we instead verify the helper logic in isolation: drop
|
|
/// `prefetch_rx` and `recycle_tx` first, then observe the producer-side
|
|
/// sender is disconnected, which is the property the fix relies on.
|
|
#[test]
|
|
fn channels_disconnected_before_producer_join_on_spawn_failure() {
|
|
// Build a prefetch channel pair. The producer "thread" is simulated by
|
|
// holding prefetch_tx; we verify it observes disconnection after we
|
|
// drop prefetch_rx (and only after — not before).
|
|
// crossbeam channels expose disconnection only through send/recv
|
|
// results (there is no is_disconnected()), so we probe it that way.
|
|
let (pf_tx, pf_rx) = bounded::<std::io::Result<Vec<u8>>>(1);
|
|
let (rc_tx, rc_rx) = bounded::<Vec<u8>>(1);
|
|
|
|
// Before any drop: the producer-side ends are live (a send into the
|
|
// depth-1 prefetch channel succeeds; the recycle receiver can still
|
|
// be fed).
|
|
assert!(
|
|
pf_tx.send(Ok(vec![1, 2, 3])).is_ok(),
|
|
"prefetch_tx must accept a send before any drop"
|
|
);
|
|
|
|
// Simulate the spawn-failure teardown: the move-closure owns prefetch_rx
|
|
// and recycle_tx, so dropping them mirrors `spawn` dropping the failed
|
|
// closure before producer_shell is joined.
|
|
drop(pf_rx);
|
|
drop(rc_tx);
|
|
|
|
// Now the producer-side handles observe disconnection via Err results —
|
|
// a blocked producer send/recv returns Err and the producer exits, so
|
|
// the subsequent join() completes without hanging.
|
|
assert!(
|
|
pf_tx.send(Ok(vec![4, 5, 6])).is_err(),
|
|
"prefetch_tx send must fail after prefetch_rx drop (producer would exit)"
|
|
);
|
|
assert!(
|
|
rc_rx.recv().is_err(),
|
|
"recycle_rx recv must fail after recycle_tx drop"
|
|
);
|
|
}
|
|
}
|