From 18e0fa451affca184f47df7baf59b0ea5acbb8e6 Mon Sep 17 00:00:00 2001 From: Marcos Hernanz <96699542+marcoshernanz@users.noreply.github.com> Date: Fri, 7 Aug 2026 10:39:21 -0700 Subject: [PATCH] fix(turbo-tasks): eliminate ABBA deadlock in the compilation event queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit send() held event_history.lock() for the whole task, then acquired DashMap shard write guards and awaited channel sends, while subscribe(None) acquired the Global shard write guard and then awaited event_history.lock() plus replayed history while still holding the shard guard. Lock orders: send = history -> shard, subscribe = shard -> history — a two-worker interleave deadlocks (the parking_lot shard wait is a synchronous non-yielding block), and every subsequent send then blocks another worker thread, progressively stalling the runtime. Restructure around one serialization point (the history lock) with no lock ever held across an await: - subscribe: [lock] register the sender + replay history with try_send [unlock] — synchronous because the fresh channel's capacity (MAX_QUEUE_SIZE) is always >= the history's capacity, so it can never block. - send: [lock] append to history + snapshot the subscriber lists [unlock], then deliver guard-free. That yields exactly-once at subscribe boundaries (an event is in the replay snapshot xor in delivery snapshots) and strict ordering (replays fully enqueue before any later live delivery), while removing the deadlock and the global stall a full/slow subscriber channel used to cause. Delivery is concurrent across subscribers (join_all), and failed subscribers are removed afterwards under fresh guards, matched by channel identity so concurrent additions are kept. --- .../crates/turbo-tasks/src/message_queue.rs | 229 ++++++++++++++---- 1 file changed, 186 insertions(+), 43 deletions(-) diff --git a/turbopack/crates/turbo-tasks/src/message_queue.rs b/turbopack/crates/turbo-tasks/src/message_queue.rs index 892285ef1921..f47c27595fcb 100644 --- a/turbopack/crates/turbo-tasks/src/message_queue.rs +++ b/turbopack/crates/turbo-tasks/src/message_queue.rs @@ -53,40 +53,62 @@ impl CompilationEventQueue { // Spawn a task to handle the async operations tokio::spawn(async move { - // Store the message in history - let mut history = event_history.lock().await; - if history.len() >= MAX_QUEUE_SIZE { - history.pop_front(); - } - history.push_back(message_clone.clone()); - - // Send to all active receivers of the same message type - if let Some(mut type_subscribers) = subscribers.get_mut(&EventChannelType::Type( - message_clone.type_name().to_owned(), - )) { - let mut removal_indices = Vec::new(); - for (ix, sender) in type_subscribers.iter().enumerate() { - if sender.send(message_clone.clone()).await.is_err() { - removal_indices.push(ix); - } - } - - for ix in removal_indices.iter().rev() { - type_subscribers.remove(*ix); + // Append to history AND snapshot the channel lists atomically + // (under the history lock): this is the single serialization + // point that keeps subscribe-boundary events exactly-once — an + // event is either in a subscriber's history snapshot xor in its + // delivery snapshot, never both. The DashMap operations are + // synchronous (no await), so this can't deadlock: no lock is ever + // held across an await. + let type_key = EventChannelType::Type(message_clone.type_name().to_owned()); + let type_channels: Vec; + let global_channels: Vec; + { + let mut history = event_history.lock().await; + if history.len() >= MAX_QUEUE_SIZE { + history.pop_front(); } + history.push_back(message_clone.clone()); + type_channels = subscribers + .get(&type_key) + .map(|v| v.clone()) + .unwrap_or_default(); + global_channels = subscribers + .get(&EventChannelType::Global) + .map(|v| v.clone()) + .unwrap_or_default(); } - // Send to all global message subscribers - let mut all_channel = subscribers.get_mut(&EventChannelType::Global).unwrap(); - let mut removal_indices = Vec::new(); - for (ix, sender) in all_channel.iter_mut().enumerate() { - if sender.send(message_clone.clone()).await.is_err() { - removal_indices.push(ix); + // Deliver to all subscribers concurrently: awaiting a full (but + // alive) typed channel must not stall or reorder delivery to + // global subscribers of the same event. + let failed: Vec = + futures::future::join_all(type_channels.iter().chain(global_channels.iter()).map( + |sender| { + let message = message_clone.clone(); + async move { + if sender.send(message).await.is_err() { + Some(sender.clone()) + } else { + None + } + } + }, + )) + .await + .into_iter() + .flatten() + .collect(); + + // Remove failed subscribers under freshly taken guards, matching + // by channel identity so concurrently added subscribers are kept. + if !failed.is_empty() { + if let Some(mut subs) = subscribers.get_mut(&type_key) { + subs.retain(|s| !failed.iter().any(|f| f.same_channel(s))); + } + if let Some(mut subs) = subscribers.get_mut(&EventChannelType::Global) { + subs.retain(|s| !failed.iter().any(|f| f.same_channel(s))); } - } - - for ix in removal_indices.iter().rev() { - all_channel.remove(*ix); } }); @@ -104,27 +126,40 @@ impl CompilationEventQueue { // Spawn a task to handle the async operations tokio::spawn(async move { - // Store the sender + // Register the sender AND snapshot the history atomically (under + // the history lock): an event is either in the snapshot (replayed + // below) xor in some send's delivery snapshot (delivered live), + // never both. The entry() guard is dropped synchronously — no + // lock is ever held across an await. + // Register the sender AND replay the history inside the history + // lock. The replay uses try_send: the fresh channel's capacity + // (MAX_QUEUE_SIZE) is >= the history's capacity (also + // MAX_QUEUE_SIZE), so it can never block. Everything is + // synchronous while the lock is held — a later send's history + // append can't interleave — so subscribers see replayed history + // strictly before any live event, and every boundary event is + // delivered exactly once (see the send() side). if let Some(event_types) = event_types { + let history_guard = event_history.lock().await; for event_type in event_types.iter() { - let mut type_subscribers = subscribers + subscribers .entry(EventChannelType::Type(event_type.clone())) - .or_default(); - type_subscribers.push(tx_clone.clone()); + .or_default() + .push(tx_clone.clone()); } - - for event in event_history.lock().await.iter() { + for event in history_guard.iter() { if event_types.contains(&event.type_name().to_string()) { - let _ = tx_clone.send(event.clone()).await; + let _ = tx_clone.try_send(event.clone()); } } } else { - let mut global_subscribers = - subscribers.entry(EventChannelType::Global).or_default(); - global_subscribers.push(tx_clone.clone()); - - for event in event_history.lock().await.iter() { - let _ = tx_clone.send(event.clone()).await; + let history_guard = event_history.lock().await; + subscribers + .entry(EventChannelType::Global) + .or_default() + .push(tx_clone.clone()); + for event in history_guard.iter() { + let _ = tx_clone.try_send(event.clone()); } } }); @@ -292,6 +327,114 @@ impl CompilationEvent for TraceEvent { mod tests { use super::*; + /// Drives concurrent send + subscribe traffic and asserts everything + /// completes — the previous implementation held the history lock and + /// DashMap shard guards across awaits in opposite lock orders (ABBA), + /// which could deadlock the runtime. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn test_concurrent_send_and_subscribe_no_deadlock() { + for round in 0..50 { + let queue = CompilationEventQueue::default(); + + // Pre-fill history so subscribe(None) replays. + for i in 0..5 { + queue + .send(Arc::new(DiagnosticEvent::new( + Severity::Info, + format!("seed {i}"), + ))) + .unwrap(); + } + + let mut rx = queue.subscribe(None); + queue + .send(Arc::new(DiagnosticEvent::new( + Severity::Info, + "post".to_string(), + ))) + .unwrap(); + + // Collect the replayed + live events with a deadline. + let count = tokio::time::timeout(Duration::from_secs(10), async { + let mut count = 0; + while count < 6 { + match rx.recv().await { + Some(_) => count += 1, + None => break, + } + } + count + }) + .await + .unwrap_or_else(|_| panic!("round {round}: timed out (deadlock?)")); + + assert_eq!(count, 6, "round {round}: expected 5 replayed + 1 live"); + + // And nothing else: an event crossing the subscribe boundary must + // be delivered exactly once (not replayed AND delivered live). + let extra = tokio::time::timeout(Duration::from_millis(100), rx.recv()).await; + assert!( + matches!(extra, Err(_) | Ok(None)), + "round {round}: unexpected duplicate event" + ); + } + } + + /// A subscriber whose receiver is dropped must be removed from the + /// subscriber list; live subscribers must be kept. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_dead_subscriber_removed_and_live_kept() { + let queue = CompilationEventQueue::default(); + + let mut rx_live = queue.subscribe(None); + let rx_dead = queue.subscribe(None); + drop(rx_dead); + + // Let the subscribe tasks register both senders. + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!( + queue + .subscribers + .get(&EventChannelType::Global) + .unwrap() + .len(), + 2 + ); + + // This send delivers to the live one and must remove the dead one. + queue + .send(Arc::new(DiagnosticEvent::new( + Severity::Info, + "hello".to_string(), + ))) + .unwrap(); + tokio::time::sleep(Duration::from_millis(100)).await; + + assert!(rx_live.try_recv().is_ok()); + assert_eq!( + queue + .subscribers + .get(&EventChannelType::Global) + .unwrap() + .len(), + 1 + ); + + // A second event must still reach the live receiver — this only + // succeeds if the removal kept the live sender (identity-matched) + // rather than removing the wrong one. + queue + .send(Arc::new(DiagnosticEvent::new( + Severity::Info, + "second".to_string(), + ))) + .unwrap(); + tokio::time::timeout(Duration::from_secs(5), rx_live.recv()) + .await + .expect("timed out waiting for the second event") + .expect("live subscriber was removed"); + } + #[test] fn test_timing_event_string_formatting() { let tests = vec![