From 34be8a6088b3fcc294bbf74f3c49b4dfe1be0276 Mon Sep 17 00:00:00 2001 From: Yasunobu <42543015+P4suta@users.noreply.github.com> Date: Thu, 23 Jul 2026 23:00:59 +0900 Subject: [PATCH] fix(xz): deterministic async XZ decoder wakeups, no worker deadlock [DEV-124] The async XZ path drove the worker-backed decoder through the blocking Codec::process, whose wait_event() parks the executor thread on the worker's event channel. Under a slow, one-byte-at-a-time async source the executor could park with no registered waker and the worker could finish between the owner's try_recv and its park, losing the wakeup and hanging. Add a non-blocking Codec::poll_process (defaulting to process) and route the async adapter through it. In the XZ codec, refactor process() into a shared drive(.., waker: Option<&Waker>): with None it keeps the byte-for-byte blocking behavior; with Some it registers the waker before the deciding try_recv and returns Poll::Pending on Empty. A cloneable EventSink wakes the async owner after every NeedInput/Output send, and decode_worker wakes the retained waker Arc only after all senders drop, so a parked owner's next try_recv observes Disconnected instead of Empty. Add a park/unpark std executor plus a process-abort watchdog to the async regression suite and a focused async_xz_never_deadlocks_under_slow_source test (50 iterations over the one-byte source); route the three at-risk existing tests through the same guarded executor. Co-Authored-By: Claude Opus 4.8 (1M context) --- libarchive_oxide-core/src/protocol.rs | 22 +++ libarchive_oxide/src/async_filter.rs | 18 +- libarchive_oxide/src/filter/xz.rs | 215 +++++++++++++++++----- libarchive_oxide/src/pipeline_codec.rs | 32 ++++ libarchive_oxide/tests/async_stream_v2.rs | 156 ++++++++++++---- 5 files changed, 356 insertions(+), 87 deletions(-) diff --git a/libarchive_oxide-core/src/protocol.rs b/libarchive_oxide-core/src/protocol.rs index 50c01ea..711c694 100644 --- a/libarchive_oxide-core/src/protocol.rs +++ b/libarchive_oxide-core/src/protocol.rs @@ -86,6 +86,28 @@ pub trait Codec { output: &mut [u8], end: EndOfInput, ) -> Result; + + /// Non-blocking, executor-safe drive used by async adapters. + /// + /// Unlike [`process`](Codec::process) this must never block the calling + /// thread. `Ok(None)` means "would block": the codec has arranged for + /// `waker` to be woken once it can make progress, and the caller should + /// return `Poll::Pending` and poll again after the wake. The default + /// delegates to [`process`](Codec::process), which is correct for every + /// codec that never blocks a caller — in-process pull decoders and codecs + /// that signal backpressure through [`CodecStatus`] rather than a blocking + /// wait. Only a codec that would otherwise block a thread (e.g. one backed + /// by a worker thread and a blocking channel) needs to override this. + fn poll_process( + &mut self, + input: &[u8], + output: &mut [u8], + end: EndOfInput, + waker: &core::task::Waker, + ) -> Result, ArchiveError> { + let _ = waker; + self.process(input, output, end).map(Some) + } } /// A borrowed entry-data window. diff --git a/libarchive_oxide/src/async_filter.rs b/libarchive_oxide/src/async_filter.rs index e712f3f..4817a4c 100644 --- a/libarchive_oxide/src/async_filter.rs +++ b/libarchive_oxide/src/async_filter.rs @@ -247,11 +247,21 @@ impl AsyncRead for AsyncCodecReader { }; let step = match this .codec - .process(&this.compressed[this.start..this.end], output, end) - .and_then(|step| step.validate(input_length, output.len())) + .poll_process( + &this.compressed[this.start..this.end], + output, + end, + cx.waker(), + ) + .transpose() { - Ok(step) => step, - Err(error) => return Poll::Ready(Err(archive_codec_error(error))), + Some(result) => { + match result.and_then(|step| step.validate(input_length, output.len())) { + Ok(step) => step, + Err(error) => return Poll::Ready(Err(archive_codec_error(error))), + } + }, + None => return Poll::Pending, }; this.start += step.consumed; if step.produced != 0 { diff --git a/libarchive_oxide/src/filter/xz.rs b/libarchive_oxide/src/filter/xz.rs index 484bd69..20e872a 100644 --- a/libarchive_oxide/src/filter/xz.rs +++ b/libarchive_oxide/src/filter/xz.rs @@ -6,8 +6,9 @@ use std::fmt; use std::io::{self, Read, Write}; -use std::sync::Mutex; use std::sync::mpsc::{self, Receiver, SyncSender, TryRecvError, TrySendError}; +use std::sync::{Arc, Mutex}; +use std::task::Waker; use std::thread; use libarchive_oxide_core::{ @@ -31,9 +32,42 @@ enum WorkerEvent { Output(Vec), } +/// Cloneable worker-event sender that wakes the async owner after every send. +/// +/// The worker thread and its input pipe both hold clones. `send` first delivers +/// the event over the blocking channel, then wakes (and clears) any async waker +/// registered by the owner, so a parked executor observes both `NeedInput` and +/// `Output` progress. `wake_owner` performs just the wake, used on worker exit. +#[derive(Clone)] +struct EventSink { + sender: SyncSender, + waker: Arc>>, +} + +impl EventSink { + fn send(&self, event: WorkerEvent) -> Result<(), mpsc::SendError> { + let result = self.sender.send(event); + self.wake_owner(); + result + } + + fn wake_owner(&self) { + wake_cell(&self.waker); + } +} + +/// Wakes and clears the waker parked in `cell`, tolerating lock poisoning. +fn wake_cell(cell: &Mutex>) { + if let Ok(mut guard) = cell.lock() { + if let Some(waker) = guard.take() { + waker.wake(); + } + } +} + struct InputPipe { receiver: Receiver, - events: SyncSender, + events: EventSink, current: Vec, position: usize, ended: bool, @@ -455,6 +489,8 @@ pub(crate) struct XzDecoder { sender: Option>, // Preserve the public readers' historical Sync and RefUnwindSafe auto traits. events: Mutex>, + // Shared slot the worker wakes; `Arc>` keeps Sync + RefUnwindSafe. + waker_cell: Arc>>, worker: Mutex>>>, validator: XzValidator, pending_input: Vec, @@ -469,9 +505,14 @@ impl XzDecoder { pub(crate) fn new(limits: Limits) -> Result { let (sender, input) = mpsc::sync_channel(2); let (event_sender, events) = mpsc::sync_channel(2); + let waker_cell = Arc::new(Mutex::new(None)); + let sink = EventSink { + sender: event_sender, + waker: Arc::clone(&waker_cell), + }; let worker = thread::Builder::new() .name("libarchive-oxide-xz".into()) - .spawn(move || decode_worker(input, &event_sender)) + .spawn(move || decode_worker(input, sink)) .map_err(|error| { ArchiveError::new(ErrorKind::Capability) .with_format("xz") @@ -480,6 +521,7 @@ impl XzDecoder { Ok(Self { sender: Some(sender), events: Mutex::new(events), + waker_cell, worker: Mutex::new(Some(worker)), validator: XzValidator::new(limits), pending_input: Vec::with_capacity(MAX_STAGED), @@ -617,6 +659,38 @@ impl XzDecoder { Ok(0) } } + + /// Parks `waker` in the shared slot so the worker can wake it. Uses + /// [`Waker::will_wake`] to skip a redundant clone when re-registering the + /// same executor waker across polls. + fn register_waker(&self, waker: &Waker) { + if let Ok(mut guard) = self.waker_cell.lock() { + let refresh = guard + .as_ref() + .is_none_or(|existing| !existing.will_wake(waker)); + if refresh { + *guard = Some(waker.clone()); + } + } + } + + /// Waits for the next worker event, blocking on the sync path (`waker` is + /// `None`) and non-blocking on the async path (`Some`). In the async case + /// the waker is registered *before* the deciding `try_recv`, so a would-be + /// blocking read reports `Ok(None)` ("pending") after arranging a wake. + fn pump( + &mut self, + output: &mut [u8], + waker: Option<&Waker>, + ) -> Result, ArchiveError> { + match waker { + None => self.wait_event(output).map(Some), + Some(waker) => { + self.register_waker(waker); + self.poll_event(output) + }, + } + } } impl fmt::Debug for XzDecoder { @@ -636,24 +710,32 @@ impl fmt::Debug for XzDecoder { } } -impl Codec for XzDecoder { +impl XzDecoder { + /// Shared engine for the sync and async codec entry points. + /// + /// `waker` is `None` on the blocking [`Codec::process`] path (the returned + /// step is always `Some`, byte-for-byte identical to the historical + /// behavior) and `Some` on the async [`Codec::poll_process`] path, where + /// `Ok(None)` means "would block": a wake has been arranged and the caller + /// should return `Poll::Pending`. #[allow(clippy::too_many_lines)] // Progress, backpressure, and terminal ordering stay together. - fn process( + fn drive( &mut self, input: &[u8], output: &mut [u8], end: EndOfInput, - ) -> Result { + waker: Option<&Waker>, + ) -> Result, ArchiveError> { if let Some(error) = &self.failure { return Err(error.clone()); } if self.done { if input.is_empty() { - return Ok(CodecStep { + return Ok(Some(CodecStep { consumed: 0, produced: 0, status: CodecStatus::Done, - }); + })); } return Err(self.fail(malformed("data follows the completed XZ stream"))); } @@ -662,11 +744,11 @@ impl Codec for XzDecoder { let mut produced = self.drain_output(output); loop { if self.done { - return Ok(CodecStep { + return Ok(Some(CodecStep { consumed, produced, status: CodecStatus::Done, - }); + })); } while produced < output.len() { let Some(read) = self.poll_event(&mut output[produced..])? else { @@ -678,11 +760,11 @@ impl Codec for XzDecoder { } } if self.done { - return Ok(CodecStep { + return Ok(Some(CodecStep { consumed, produced, status: CodecStatus::Done, - }); + })); } let flushed = self.flush_input()?; @@ -722,18 +804,21 @@ impl Codec for XzDecoder { } } if self.done { - return Ok(CodecStep { + return Ok(Some(CodecStep { consumed, produced, status: CodecStatus::Done, - }); + })); } if effective_end && produced == 0 && self.pending_output.is_empty() && !output.is_empty() { - produced += self.wait_event(&mut output[produced..])?; + let Some(read) = self.pump(&mut output[produced..], waker)? else { + return Ok(None); + }; + produced += read; continue; } if produced != 0 || consumed != 0 { @@ -747,28 +832,34 @@ impl Codec for XzDecoder { } else { CodecStatus::NeedInput }; - return Ok(CodecStep { + return Ok(Some(CodecStep { consumed, produced, status, - }); + })); } if effective_end { if output.is_empty() && !self.pending_output.is_empty() { - return Ok(CodecStep { + return Ok(Some(CodecStep { consumed, produced, status: CodecStatus::NeedOutput, - }); + })); } - produced += self.wait_event(&mut output[produced..])?; + let Some(read) = self.pump(&mut output[produced..], waker)? else { + return Ok(None); + }; + produced += read; continue; } if !input.is_empty() || !self.pending_input.is_empty() { - produced += self.wait_event(&mut output[produced..])?; + let Some(read) = self.pump(&mut output[produced..], waker)? else { + return Ok(None); + }; + produced += read; continue; } - return Ok(CodecStep { + return Ok(Some(CodecStep { consumed: 0, produced, status: if output.is_empty() && !self.pending_output.is_empty() { @@ -776,37 +867,69 @@ impl Codec for XzDecoder { } else { CodecStatus::NeedInput }, - }); + })); } } } -fn decode_worker( - input: Receiver, - events: &SyncSender, -) -> io::Result<()> { - let pipe = InputPipe { - receiver: input, - events: events.clone(), - current: Vec::new(), - position: 0, - ended: false, - }; - let mut decoder = lzma_rust2::XzReader::new(pipe, true); - let mut output = vec![0; BUFFER]; - loop { - match decoder.read(&mut output) { - Ok(0) => return Ok(()), - Ok(read) => events - .send(WorkerEvent::Output(output[..read].to_vec())) - .map_err(|_| { - io::Error::new(io::ErrorKind::BrokenPipe, "XZ codec owner was dropped") - })?, - Err(error) => return Err(error), - } +impl Codec for XzDecoder { + // The `None` waker drives the blocking path, which never yields `Ok(None)`; + // the `expect` documents that invariant on the one impossible branch. + #[allow(clippy::expect_used)] + fn process( + &mut self, + input: &[u8], + output: &mut [u8], + end: EndOfInput, + ) -> Result { + self.drive(input, output, end, None) + .map(|step| step.expect("blocking XZ drive always yields a step")) + } + + fn poll_process( + &mut self, + input: &[u8], + output: &mut [u8], + end: EndOfInput, + waker: &Waker, + ) -> Result, ArchiveError> { + self.drive(input, output, end, Some(waker)) } } +fn decode_worker(input: Receiver, events: EventSink) -> io::Result<()> { + // Retain ONLY the waker Arc. The decode loop runs in an inner scope that + // owns every `EventSink` (this one and the pipe's clone); when the scope + // ends all event senders drop and the event channel disconnects. Only then + // do we wake, so a parked owner's next `try_recv` observes `Disconnected` + // (terminal progress) rather than `Empty` (a lost wakeup / deadlock). + let waker_cell = Arc::clone(&events.waker); + let result = (move || -> io::Result<()> { + let pipe = InputPipe { + receiver: input, + events: events.clone(), + current: Vec::new(), + position: 0, + ended: false, + }; + let mut decoder = lzma_rust2::XzReader::new(pipe, true); + let mut output = vec![0; BUFFER]; + loop { + match decoder.read(&mut output) { + Ok(0) => return Ok(()), + Ok(read) => events + .send(WorkerEvent::Output(output[..read].to_vec())) + .map_err(|_| { + io::Error::new(io::ErrorKind::BrokenPipe, "XZ codec owner was dropped") + })?, + Err(error) => return Err(error), + } + } + })(); + wake_cell(&waker_cell); + result +} + fn validate_block_memory(header: &[u8], codec_memory: Option) -> Result<(), ArchiveError> { let Some(limit) = codec_memory else { return Ok(()); diff --git a/libarchive_oxide/src/pipeline_codec.rs b/libarchive_oxide/src/pipeline_codec.rs index c7946c5..059f5fb 100644 --- a/libarchive_oxide/src/pipeline_codec.rs +++ b/libarchive_oxide/src/pipeline_codec.rs @@ -4,6 +4,9 @@ //! Private static dispatch for caller-driven outer codecs. +#[cfg(feature = "async")] +use std::task::Waker; + use libarchive_oxide_core::filter::FilterId; use libarchive_oxide_core::{ArchiveError, Codec, CodecStep, EndOfInput, ErrorKind, Limits}; @@ -140,6 +143,35 @@ impl PipelineCodec { Self::Lz4(codec) => codec.process(input, output, end), } } + + /// Non-blocking mirror of [`process`](Self::process) for async adapters. + /// + /// Every variant inherits the blocking-delegating default except `Xz`, + /// which overrides [`Codec::poll_process`] to avoid parking the executor + /// thread on its worker channel. + #[cfg(feature = "async")] + pub(crate) fn poll_process( + &mut self, + input: &[u8], + output: &mut [u8], + end: EndOfInput, + waker: &Waker, + ) -> Result, ArchiveError> { + match self { + #[cfg(feature = "native-codecs")] + Self::Gzip(codec) => codec.poll_process(input, output, end, waker), + #[cfg(not(feature = "native-codecs"))] + Self::Gzip(codec) => codec.poll_process(input, output, end, waker), + #[cfg(feature = "bzip2")] + Self::Bzip2(codec) => codec.poll_process(input, output, end, waker), + #[cfg(feature = "zstd")] + Self::Zstd(codec) => codec.poll_process(input, output, end, waker), + #[cfg(feature = "xz")] + Self::Xz(codec) => codec.poll_process(input, output, end, waker), + #[cfg(feature = "lz4")] + Self::Lz4(codec) => codec.poll_process(input, output, end, waker), + } + } } #[cfg(all(feature = "zstd", feature = "native-codecs"))] diff --git a/libarchive_oxide/tests/async_stream_v2.rs b/libarchive_oxide/tests/async_stream_v2.rs index f50667b..f02b054 100644 --- a/libarchive_oxide/tests/async_stream_v2.rs +++ b/libarchive_oxide/tests/async_stream_v2.rs @@ -6,10 +6,15 @@ #![cfg(feature = "async")] #![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)] +use std::future::Future; use std::io::{Cursor, Write}; use std::panic::RefUnwindSafe; use std::pin::Pin; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::task::{Context, Poll}; +use std::thread; +use std::time::{Duration, Instant}; use futures_io::AsyncRead; use futures_lite::future::block_on; @@ -159,6 +164,80 @@ impl AsyncRead for AsyncOneByte { } } +/// Minimal std-only executor that polls `future` to completion, parking the +/// thread between polls behind a *real* `Waker` built from thread unpark. +/// +/// Unlike `futures_lite::block_on`, it has no internal timer: if a codec +/// returns `Poll::Pending` without ever waking its waker, this parks forever. +/// That is exactly the deadlock we regression-test, and the watchdog turns the +/// hang into a fast, deterministic failure. +fn park_block_on(future: F) -> F::Output { + use std::task::Wake; + + struct ThreadWaker(thread::Thread); + impl Wake for ThreadWaker { + fn wake(self: Arc) { + self.0.unpark(); + } + fn wake_by_ref(self: &Arc) { + self.0.unpark(); + } + } + + let waker = std::task::Waker::from(Arc::new(ThreadWaker(thread::current()))); + let mut context = Context::from_waker(&waker); + let mut future = Box::pin(future); + loop { + match future.as_mut().poll(&mut context) { + Poll::Ready(output) => return output, + Poll::Pending => thread::park(), + } + } +} + +/// Arms a watchdog thread that aborts the process if `label` has not signaled +/// completion within ~30s, so a deadlocked poll fails loudly instead of hanging +/// the test runner indefinitely. +fn arm_watchdog(label: &'static str) -> Arc { + let done = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&done); + thread::spawn(move || { + let deadline = Instant::now() + Duration::from_secs(30); + while Instant::now() < deadline { + if flag.load(Ordering::SeqCst) { + return; + } + thread::sleep(Duration::from_millis(50)); + } + eprintln!("watchdog: `{label}` did not complete within 30s; aborting the deadlocked test"); + std::process::abort(); + }); + done +} + +/// Drives `future` on the park/unpark executor under the abort watchdog. +fn guarded(label: &'static str, future: F) -> F::Output { + let done = arm_watchdog(label); + let output = park_block_on(future); + done.store(true, Ordering::SeqCst); + output +} + +#[test] +fn async_xz_never_deadlocks_under_slow_source() { + let expected = collect_sync(fixture()); + let archive = sync_filtered_fixture(FilterId::Xz); + guarded("async_xz_never_deadlocks_under_slow_source", async { + for iteration in 0..50 { + assert_eq!( + collect_futures(archive.clone()).await, + expected, + "xz iteration {iteration}" + ); + } + }); +} + fn sync_filtered_fixture(filter: FilterId) -> Vec { let mut writer = ArchiveWriter::with_filter(Vec::new(), FormatId::Tar, Some(filter), Limits::default()) @@ -171,7 +250,7 @@ fn sync_filtered_fixture(filter: FilterId) -> Vec { #[test] fn futures_reader_and_writer_match_sync_contract() { - block_on(async { + guarded("futures_reader_and_writer_match_sync_contract", async { let expected = collect_sync(fixture()); assert_eq!(collect_futures(fixture()).await, expected); @@ -227,48 +306,51 @@ fn futures_reader_and_writer_match_sync_contract() { #[test] fn async_reader_concatenates_every_filter_and_rejects_trailing_data() { - block_on(async { - let plain = fixture(); - let expected = collect_sync(plain.clone()); - let split = plain.len() / 2; - for filter in [ - FilterId::Gzip, - FilterId::Bzip2, - FilterId::Zstd, - FilterId::Xz, - FilterId::Lz4, - ] { - let mut members = compress(&plain[..split], filter).unwrap(); - members.extend_from_slice(&compress(&plain[split..], filter).unwrap()); - assert_eq!(collect_futures(members).await, expected, "{filter:?}"); - - let mut trailing = compress(&plain, filter).unwrap(); - trailing.push(0); - let mut reader = AsyncArchiveReader::new(AsyncOneByte { - bytes: trailing, - position: 0, - }); - loop { - match reader.next_event().await { - Ok(ReaderEvent::Done) => panic!("{filter:?} accepted trailing data"), - Ok(_) => {}, - Err(error) => { - assert_eq!( - error.io_error().map(std::io::Error::kind), - Some(std::io::ErrorKind::InvalidData), - "{filter:?}: {error}" - ); - break; - }, + guarded( + "async_reader_concatenates_every_filter_and_rejects_trailing_data", + async { + let plain = fixture(); + let expected = collect_sync(plain.clone()); + let split = plain.len() / 2; + for filter in [ + FilterId::Gzip, + FilterId::Bzip2, + FilterId::Zstd, + FilterId::Xz, + FilterId::Lz4, + ] { + let mut members = compress(&plain[..split], filter).unwrap(); + members.extend_from_slice(&compress(&plain[split..], filter).unwrap()); + assert_eq!(collect_futures(members).await, expected, "{filter:?}"); + + let mut trailing = compress(&plain, filter).unwrap(); + trailing.push(0); + let mut reader = AsyncArchiveReader::new(AsyncOneByte { + bytes: trailing, + position: 0, + }); + loop { + match reader.next_event().await { + Ok(ReaderEvent::Done) => panic!("{filter:?} accepted trailing data"), + Ok(_) => {}, + Err(error) => { + assert_eq!( + error.io_error().map(std::io::Error::kind), + Some(std::io::ErrorKind::InvalidData), + "{filter:?}: {error}" + ); + break; + }, + } } } - } - }); + }, + ); } #[test] fn async_reader_composes_four_nested_filters() { - block_on(async { + guarded("async_reader_composes_four_nested_filters", async { let plain = fixture(); let expected = collect_sync(plain.clone()); let mut nested = plain;