Files
announcement/tests/deadlock_prevention_tests.rs
swaits e8e55b7676 chore: initial release v0.1.0
Runtime-agnostic oneshot broadcast channel for Rust.
2025-10-13 19:36:37 -06:00

764 lines
21 KiB
Rust

//! Deadlock prevention tests for the announcement crate.
//! Verifies that the crate cannot deadlock in any scenario.
use announcement::Announcement;
#[cfg(feature = "std")]
use announcement::ListenError;
use std::sync::{Arc, Barrier, Mutex};
use std::thread;
use std::time::{Duration, Instant};
#[cfg(feature = "std")]
/// Test: Deadlock prevention: listen timeout actually times out
///
/// Verifies: Verifies listen timeout actually times out does not cause deadlock
/// Category: deadlock
#[test]
fn test_listen_timeout_actually_times_out() {
let (_announcer, announcement) = Announcement::<i32>::new();
let listener = announcement.listener();
// Don't announce, just let it timeout
let start = Instant::now();
let result = listener.listen_timeout_blocking(Duration::from_millis(100));
let elapsed = start.elapsed();
assert_eq!(result, Err(ListenError::Timeout));
assert!(elapsed >= Duration::from_millis(100));
// Increased tolerance for coverage instrumentation overhead (tarpaulin adds significant delay)
assert!(elapsed < Duration::from_secs(1)); // Very generous tolerance
}
#[cfg(feature = "std")]
/// Test: Deadlock prevention: announcer dropped listeners detect closed
///
/// Verifies: Verifies announcer dropped listeners detect closed does not cause deadlock
/// Category: deadlock
#[test]
fn test_announcer_dropped_listeners_detect_closed() {
let announcement = {
let (announcer, announcement) = Announcement::<String>::new();
drop(announcer); // Drop announcer immediately - sets closed flag
announcement
};
let listener = announcement.listener();
// With closed detection, this returns Closed immediately, not Timeout
let result = listener.listen_timeout_blocking(Duration::from_millis(100));
assert_eq!(result, Err(ListenError::Closed));
assert!(announcement.is_closed());
}
/// Test: Deadlock prevention: no deadlock nested announcements
///
/// Verifies: Verifies no deadlock nested announcements cannot cause deadlock
/// Category: deadlock
#[test]
fn test_no_deadlock_nested_announcements() {
// A -> B -> C chain
let (ann_a, announcement_a) = Announcement::<i32>::new();
let (ann_b, announcement_b) = Announcement::<i32>::new();
let (ann_c, announcement_c) = Announcement::<i32>::new();
let listener_a = announcement_a.listener();
let listener_b = announcement_b.listener();
let listener_c = announcement_c.listener();
// Thread 1: A -> B
let h1 = thread::spawn(move || {
ann_a.announce(1).unwrap();
loop {
if let Some(val) = listener_b.try_listen() {
return val;
}
thread::yield_now();
}
});
// Thread 2: B -> C
let h2 = thread::spawn(move || {
ann_b.announce(2).unwrap();
loop {
if let Some(val) = listener_c.try_listen() {
return val;
}
thread::yield_now();
}
});
// Thread 3: C -> complete
let h3 = thread::spawn(move || {
ann_c.announce(3).unwrap();
loop {
if let Some(val) = listener_a.try_listen() {
return val;
}
thread::yield_now();
}
});
// All should complete without deadlock
assert_eq!(h1.join().unwrap(), 2);
assert_eq!(h2.join().unwrap(), 3);
assert_eq!(h3.join().unwrap(), 1);
}
/// Test: Deadlock prevention: no deadlock circular deps
///
/// Verifies: Verifies no deadlock circular deps cannot cause deadlock
/// Category: deadlock
#[test]
fn test_no_deadlock_circular_deps() {
let (ann1, announcement1) = Announcement::<&str>::new();
let (ann2, announcement2) = Announcement::<&str>::new();
let listener1_for_2 = announcement1.listener();
let listener2_for_1 = announcement2.listener();
// These would deadlock with improper locking
let h1 = thread::spawn(move || {
ann1.announce("from1").unwrap();
let start = Instant::now();
loop {
if let Some(val) = listener2_for_1.try_listen() {
return val;
}
if start.elapsed() > Duration::from_secs(1) {
return "timeout1";
}
thread::yield_now();
}
});
let h2 = thread::spawn(move || {
ann2.announce("from2").unwrap();
let start = Instant::now();
loop {
if let Some(val) = listener1_for_2.try_listen() {
return val;
}
if start.elapsed() > Duration::from_secs(1) {
return "timeout2";
}
thread::yield_now();
}
});
let r1 = h1.join().unwrap();
let r2 = h2.join().unwrap();
assert_eq!(r1, "from2");
assert_eq!(r2, "from1");
}
/// Test: Deadlock prevention: no deadlock multiple mutexes
///
/// Verifies: Verifies no deadlock multiple mutexes cannot cause deadlock
/// Category: deadlock
#[test]
fn test_no_deadlock_multiple_mutexes() {
let (announcer, announcement) = Announcement::new();
let mutex1 = Arc::new(Mutex::new(0));
let mutex2 = Arc::new(Mutex::new(0));
let m1_clone = mutex1.clone();
let m2_clone = mutex2.clone();
let listener1 = announcement.listener();
let h1 = thread::spawn(move || {
let _lock1 = m1_clone.lock().unwrap();
thread::sleep(Duration::from_millis(10));
let _lock2 = m2_clone.lock().unwrap();
loop {
if let Some(val) = listener1.try_listen() {
return val;
}
thread::yield_now();
}
});
let m1_clone2 = mutex1.clone();
let m2_clone2 = mutex2.clone();
let listener2 = announcement.listener();
let h2 = thread::spawn(move || {
let _lock2 = m2_clone2.lock().unwrap();
thread::sleep(Duration::from_millis(10));
let _lock1 = m1_clone2.lock().unwrap();
loop {
if let Some(val) = listener2.try_listen() {
return val;
}
thread::yield_now();
}
});
// This might deadlock with the mutexes, but announcement should still work
thread::sleep(Duration::from_millis(50));
announcer.announce(42).unwrap();
// Set a timeout for the test
let start = Instant::now();
while start.elapsed() < Duration::from_secs(2) {
if h1.is_finished() && h2.is_finished() {
break;
}
thread::sleep(Duration::from_millis(10));
}
// The mutex operations might deadlock, but announcement itself doesn't
// This test shows announcement doesn't contribute to deadlocks
}
#[cfg(feature = "std")]
/// Test: Deadlock prevention: no deadlock blocking ops
///
/// Verifies: Verifies no deadlock blocking ops cannot cause deadlock
/// Category: deadlock
#[test]
fn test_no_deadlock_blocking_ops() {
let (announcer, announcement) = Announcement::new();
let mut handles = vec![];
// Multiple blocking listeners
for _ in 0..10 {
let listener = announcement.listener();
handles.push(thread::spawn(move || {
listener.listen_timeout_blocking(Duration::from_millis(500))
}));
}
// Let them block for a bit
thread::sleep(Duration::from_millis(100));
// Announce
announcer.announce(42).unwrap();
// All should complete
for handle in handles {
let result = handle.join().unwrap();
assert_eq!(result, Ok(42));
}
}
/// Test: Deadlock prevention: no deadlock with panic
///
/// Verifies: Verifies no deadlock with panic cannot cause deadlock
/// Category: deadlock
#[test]
fn test_no_deadlock_with_panic() {
let (announcer, announcement) = Announcement::<i32>::new();
let listener = announcement.listener();
// Thread that panics
let h = thread::spawn(move || {
let _l = announcement.listener();
panic!("Intentional panic");
});
// Wait for panic
let _ = h.join();
// Should still work
announcer.announce(42).unwrap();
assert_eq!(listener.try_listen(), Some(42));
}
/// Test: Deadlock prevention: no deadlock poisoned mutex value
///
/// Verifies: Verifies no deadlock poisoned mutex value cannot cause deadlock
/// Category: deadlock
#[test]
fn test_no_deadlock_poisoned_mutex_value() {
let (announcer, announcement) = Announcement::new();
let mutex = Arc::new(Mutex::new(42));
let mutex_clone = mutex.clone();
// Poison the mutex
let _ = thread::spawn(move || {
let _guard = mutex_clone.lock().unwrap();
panic!("Poison the mutex");
})
.join();
// Announce the poisoned mutex
announcer.announce(mutex.clone()).unwrap();
let listener = announcement.listener();
let received = listener.try_listen().unwrap();
// Can still receive, even if mutex is poisoned
assert!(received.lock().is_err()); // It's poisoned
}
/// Test: Deadlock prevention: no deadlock recursive listeners
///
/// Verifies: Verifies no deadlock recursive listeners cannot cause deadlock
/// Category: deadlock
#[test]
fn test_no_deadlock_recursive_listeners() {
fn create_recursive_listeners(announcement: &Announcement<i32>, depth: usize) {
if depth == 0 {
return;
}
let _listener = announcement.listener();
create_recursive_listeners(announcement, depth - 1);
}
let (announcer, announcement) = Announcement::new();
// Create deeply nested listeners
create_recursive_listeners(&announcement, 1000);
// Should still work
announcer.announce(42).unwrap();
let final_listener = announcement.listener();
assert_eq!(final_listener.try_listen(), Some(42));
}
/// Test: Deadlock prevention: no deadlock barrier sync
///
/// Verifies: Verifies no deadlock barrier sync cannot cause deadlock
/// Category: deadlock
#[test]
fn test_no_deadlock_barrier_sync() {
let (announcer, announcement) = Announcement::new();
let barrier = Arc::new(Barrier::new(11));
let mut handles = vec![];
for _ in 0..10 {
let listener = announcement.listener();
let barrier = barrier.clone();
handles.push(thread::spawn(move || {
// Wait at barrier
barrier.wait();
// Try to receive
let start = Instant::now();
loop {
if let Some(val) = listener.try_listen() {
return val;
}
if start.elapsed() > Duration::from_secs(1) {
panic!("Timeout waiting for value");
}
thread::yield_now();
}
}));
}
// Main thread also waits
barrier.wait();
// Then announces
announcer.announce(42).unwrap();
// All should complete
for handle in handles {
assert_eq!(handle.join().unwrap(), 42);
}
}
/// Test: Deadlock prevention: no deadlock channel chains
///
/// Verifies: Verifies no deadlock channel chains cannot cause deadlock
/// Category: deadlock
#[test]
fn test_no_deadlock_channel_chains() {
// Create channels separately to avoid ownership issues
let mut announcers = Vec::new();
let mut announcements = Vec::new();
for _ in 0..10 {
let (announcer, announcement) = Announcement::<i32>::new();
announcers.push(Some(announcer));
announcements.push(announcement);
}
// Chain announcements
let mut handles = vec![];
for i in 0..9 {
let listener = announcements[i].listener();
let next_announcer = announcers[i + 1].take().unwrap();
handles.push(thread::spawn(move || loop {
if let Some(val) = listener.try_listen() {
next_announcer.announce(val + 1).unwrap();
break;
}
thread::yield_now();
}));
}
// Start the chain
announcers[0].take().unwrap().announce(1).unwrap();
// Wait for chain to complete
thread::sleep(Duration::from_millis(100));
// Check final value
let final_listener = announcements[9].listener();
let start = Instant::now();
loop {
if let Some(val) = final_listener.try_listen() {
assert_eq!(val, 10); // 1 + 9 increments
break;
}
if start.elapsed() > Duration::from_secs(1) {
panic!("Chain didn't complete");
}
thread::yield_now();
}
for handle in handles {
handle.join().unwrap();
}
}
#[cfg(feature = "std")]
/// Test: Deadlock prevention: no deadlock simultaneous timeouts
///
/// Verifies: Verifies no deadlock simultaneous timeouts cannot cause deadlock
/// Category: deadlock
#[test]
fn test_no_deadlock_simultaneous_timeouts() {
let (_announcer, announcement) = Announcement::<i32>::new();
let mut handles = vec![];
// Many threads timing out simultaneously
for _ in 0..50 {
let listener = announcement.listener();
handles.push(thread::spawn(move || {
listener.listen_timeout_blocking(Duration::from_millis(100))
}));
}
// All should timeout without deadlock
for handle in handles {
let result = handle.join().unwrap();
assert_eq!(result, Err(ListenError::Timeout));
}
}
#[cfg(feature = "std")]
/// Test: Deadlock prevention: no deadlock mixed blocking
///
/// Verifies: Verifies no deadlock mixed blocking cannot cause deadlock
/// Category: deadlock
#[test]
fn test_no_deadlock_mixed_blocking() {
let (announcer, announcement) = Announcement::new();
let mut handles = vec![];
// Blocking listeners
for _ in 0..5 {
let listener = announcement.listener();
handles.push(thread::spawn(move || {
listener.listen_timeout_blocking(Duration::from_secs(1))
}));
}
// Non-blocking listeners
for _ in 0..5 {
let listener = announcement.listener();
handles.push(thread::spawn(move || {
let start = Instant::now();
loop {
if let Some(val) = listener.try_listen() {
return Ok(val);
}
if start.elapsed() > Duration::from_secs(1) {
return Err(ListenError::Timeout);
}
thread::yield_now();
}
}));
}
thread::sleep(Duration::from_millis(100));
announcer.announce(42).unwrap();
// All should complete
for handle in handles {
let result = handle.join().unwrap();
assert_eq!(result, Ok(42));
}
}
/// Test: Deadlock prevention: no deadlock rapid churn
///
/// Verifies: Verifies no deadlock rapid churn cannot cause deadlock
/// Category: deadlock
#[test]
fn test_no_deadlock_rapid_churn() {
let duration = Duration::from_secs(2);
let start = Instant::now();
while start.elapsed() < duration {
let (announcer, announcement) = Announcement::<i32>::new();
let mut handles = vec![];
// Rapid creation
for _ in 0..10 {
let listener = announcement.listener();
handles.push(thread::spawn(move || listener.try_listen()));
}
// Quick announce
announcer.announce(42).unwrap();
// Rapid destruction
for handle in handles {
handle.join().unwrap();
}
}
}
#[cfg(feature = "std")]
/// Test: Deadlock prevention: no deadlock error propagation
///
/// Verifies: Verifies no deadlock error propagation cannot cause deadlock
/// Category: deadlock
#[test]
fn test_no_deadlock_error_propagation() {
let (_announcer, announcement) = Announcement::<Result<i32, String>>::new();
let mut handles = vec![];
for i in 0..10 {
let listener = announcement.listener();
handles.push(thread::spawn(move || {
match listener.listen_timeout_blocking(Duration::from_millis(100)) {
Ok(_val) => format!("Thread {} got value", i),
Err(ListenError::Timeout) => format!("Thread {} timed out", i),
Err(ListenError::Closed) => format!("Thread {} got closed", i),
}
}));
}
// All timeout, but no deadlock
for handle in handles {
let result = handle.join().unwrap();
assert!(result.contains("timed out"));
}
}
/// Test: Deadlock prevention: no deadlock signal pattern
///
/// Verifies: Verifies no deadlock signal pattern cannot cause deadlock
/// Category: deadlock
#[test]
fn test_no_deadlock_signal_pattern() {
let (shutdown_ann, shutdown) = Announcement::<()>::new();
let (ready_ann, ready) = Announcement::<()>::new();
let shutdown_listener = shutdown.listener();
let ready_listener = ready.listener();
// Worker thread
let worker = thread::spawn(move || {
// Signal ready
ready_ann.announce(()).unwrap();
// Work until shutdown
let start = Instant::now();
loop {
if shutdown_listener.try_listen().is_some() {
return "Shutdown received";
}
if start.elapsed() > Duration::from_secs(2) {
return "Timeout";
}
thread::yield_now();
}
});
// Wait for ready
let start = Instant::now();
loop {
if ready_listener.try_listen().is_some() {
break;
}
if start.elapsed() > Duration::from_secs(1) {
panic!("Ready signal timeout");
}
thread::yield_now();
}
// Send shutdown
shutdown_ann.announce(()).unwrap();
let result = worker.join().unwrap();
assert_eq!(result, "Shutdown received");
}
/// Test: Deadlock prevention: no deadlock cross thread deps
///
/// Verifies: Verifies no deadlock cross thread deps cannot cause deadlock
/// Category: deadlock
#[test]
fn test_no_deadlock_cross_thread_deps() {
let (ann1, announcement1) = Announcement::new();
let (ann2, announcement2) = Announcement::new();
let (ann3, announcement3) = Announcement::new();
let l1 = announcement1.listener();
let l2 = announcement2.listener();
let l3 = announcement3.listener();
// Complex dependency graph
let h1 = thread::spawn(move || {
ann1.announce(1).unwrap();
// Wait for 3
loop {
if let Some(val) = l3.try_listen() {
return val;
}
thread::yield_now();
}
});
let h2 = thread::spawn(move || {
// Wait for 1
loop {
if l1.try_listen().is_some() {
break;
}
thread::yield_now();
}
ann2.announce(2).unwrap();
2
});
let h3 = thread::spawn(move || {
// Wait for 2
loop {
if l2.try_listen().is_some() {
break;
}
thread::yield_now();
}
ann3.announce(3).unwrap();
3
});
// All should complete without deadlock
assert_eq!(h1.join().unwrap(), 3);
assert_eq!(h2.join().unwrap(), 2);
assert_eq!(h3.join().unwrap(), 3);
}
#[cfg(feature = "std")]
/// Test: Deadlock prevention: no deadlock timeout cascade
///
/// Verifies: Verifies no deadlock timeout cascade cannot cause deadlock
/// Category: deadlock
#[test]
fn test_no_deadlock_timeout_cascade() {
// Create chains separately
let mut announcers = Vec::new();
let mut announcements = Vec::new();
for _ in 0..5 {
let (announcer, announcement) = Announcement::<i32>::new();
announcers.push(Some(announcer));
announcements.push(announcement);
}
let mut handles = vec![];
for i in 0..4 {
let listener = announcements[i].listener();
let announcer = announcers[i + 1].take().unwrap();
handles.push(thread::spawn(move || {
match listener.listen_timeout_blocking(Duration::from_millis(200)) {
Ok(val) => {
announcer.announce(val + 1).ok();
Ok(val)
}
Err(e) => Err(e),
}
}));
}
// Don't announce to first, causing cascade
// First announcer dropped at end of function -> Closed
// Cascade means others either timeout or get closed
for handle in handles {
let result = handle.join().unwrap();
// Accept either Timeout or Closed - both are valid depending on timing
assert!(
result == Err(ListenError::Timeout) || result == Err(ListenError::Closed),
"Expected Timeout or Closed, got {:?}",
result
);
}
}
/// Test: Deadlock prevention: no deadlock summary
///
/// Verifies: Verifies no deadlock summary cannot cause deadlock
/// Category: deadlock
#[test]
fn test_no_deadlock_summary() {
// This test verifies that common deadlock scenarios are prevented
// 1. No internal locks that could deadlock
let (ann, announcement) = Announcement::<i32>::new();
let l1 = announcement.listener();
let l2 = announcement.listener();
// Concurrent operations don't deadlock
let h1 = thread::spawn(move || l1.try_listen());
let h2 = thread::spawn(move || l2.try_listen());
ann.announce(42).unwrap();
h1.join().unwrap();
h2.join().unwrap();
// 2. No blocking infinite waits (with timeout)
#[cfg(feature = "std")]
{
let (_ann, announcement) = Announcement::<i32>::new();
let listener = announcement.listener();
let result = listener.listen_timeout_blocking(Duration::from_millis(10));
assert_eq!(result, Err(ListenError::Timeout));
}
// 3. Clean panic handling
let (ann, announcement) = Announcement::<i32>::new();
let _listener = announcement.listener();
let _ = std::panic::catch_unwind(|| {
panic!("Test panic");
});
// Still works after panic
ann.announce(42).unwrap();
println!("All deadlock prevention tests passed!");
}