364 lines
11 KiB
Rust
364 lines
11 KiB
Rust
//! Close signal race condition tests
|
|
//! Tests that closing an announcer properly signals listeners even during active listen operations.
|
|
|
|
use announcement::{Announcement, ListenError};
|
|
#[cfg(feature = "std")]
|
|
use std::sync::{
|
|
atomic::{AtomicBool, Ordering},
|
|
Arc, Barrier,
|
|
};
|
|
#[cfg(feature = "std")]
|
|
use std::thread;
|
|
use std::time::Duration;
|
|
|
|
/// Test: Race condition test: close during blocking listen
|
|
///
|
|
/// Verifies: Verifies no race condition in close during blocking listen
|
|
/// Category: race
|
|
#[tokio::test]
|
|
async fn test_close_during_active_listen() {
|
|
for iteration in 0..1_000 {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
// Spawn task that will listen
|
|
let listen_handle = tokio::spawn(async move { listener.listen().await });
|
|
|
|
// Close at varying times
|
|
match iteration % 5 {
|
|
0 => {
|
|
// Close immediately
|
|
announcer.close();
|
|
}
|
|
1 => {
|
|
// Short delay
|
|
tokio::time::sleep(Duration::from_micros(1)).await;
|
|
announcer.close();
|
|
}
|
|
2 => {
|
|
// Medium delay
|
|
tokio::time::sleep(Duration::from_micros(10)).await;
|
|
announcer.close();
|
|
}
|
|
3 => {
|
|
// Drop instead of explicit close
|
|
drop(announcer);
|
|
}
|
|
_ => {
|
|
// Yield and close
|
|
tokio::task::yield_now().await;
|
|
announcer.close();
|
|
}
|
|
}
|
|
|
|
// Listener should see Closed error
|
|
let result = listen_handle.await.unwrap();
|
|
assert_eq!(result, Err(ListenError::Closed), "Iteration {}", iteration);
|
|
}
|
|
}
|
|
|
|
#[cfg(feature = "std")]
|
|
/// Test: Race condition test: close during blocking listen
|
|
///
|
|
/// Verifies: Verifies no race condition in close during blocking listen
|
|
/// Category: race
|
|
#[test]
|
|
fn test_close_during_blocking_listen() {
|
|
for iteration in 0..500 {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
// Thread that will block
|
|
let handle = thread::spawn(move || listener.listen_blocking());
|
|
|
|
// Close at varying times
|
|
match iteration % 4 {
|
|
0 => {
|
|
// Immediate
|
|
announcer.close();
|
|
}
|
|
1 => {
|
|
// Short delay
|
|
thread::sleep(Duration::from_micros(100));
|
|
announcer.close();
|
|
}
|
|
2 => {
|
|
// Drop
|
|
thread::sleep(Duration::from_micros(50));
|
|
drop(announcer);
|
|
}
|
|
_ => {
|
|
// Yield and close
|
|
thread::yield_now();
|
|
announcer.close();
|
|
}
|
|
}
|
|
|
|
let result = handle.join().unwrap();
|
|
assert_eq!(result, Err(ListenError::Closed));
|
|
}
|
|
}
|
|
|
|
/// Test: Race condition test: close during timeout blocking
|
|
///
|
|
/// Verifies: Verifies no race condition in close during timeout blocking
|
|
/// Category: race
|
|
#[tokio::test]
|
|
async fn test_close_with_concurrent_listeners() {
|
|
for _ in 0..500 {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
|
|
let mut handles = Vec::new();
|
|
|
|
// Spawn 10 listeners
|
|
for _ in 0..10 {
|
|
let listener = announcement.listener();
|
|
handles.push(tokio::spawn(async move { listener.listen().await }));
|
|
}
|
|
|
|
// Give them time to set up
|
|
tokio::time::sleep(Duration::from_micros(10)).await;
|
|
|
|
// Close
|
|
announcer.close();
|
|
|
|
// All should see Closed
|
|
for handle in handles {
|
|
let result = handle.await.unwrap();
|
|
assert_eq!(result, Err(ListenError::Closed));
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Test: Race condition test: close during timeout blocking
|
|
///
|
|
/// Verifies: Verifies no race condition in close during timeout blocking
|
|
/// Category: race
|
|
#[tokio::test]
|
|
async fn test_close_in_toctou_window() {
|
|
for iteration in 0..2_000 {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let announcement_clone = announcement.clone();
|
|
|
|
// Thread that closes at random time
|
|
let close_handle = tokio::spawn(async move {
|
|
// Vary timing to hit different points
|
|
match iteration % 7 {
|
|
0 => {}
|
|
1 => tokio::task::yield_now().await,
|
|
2 => tokio::time::sleep(Duration::from_nanos(1)).await,
|
|
3 => tokio::time::sleep(Duration::from_nanos(10)).await,
|
|
4 => tokio::time::sleep(Duration::from_nanos(100)).await,
|
|
5 => tokio::time::sleep(Duration::from_micros(1)).await,
|
|
_ => tokio::time::sleep(Duration::from_micros(10)).await,
|
|
}
|
|
announcer.close();
|
|
});
|
|
|
|
// Task that listens
|
|
let listen_handle = tokio::spawn(async move {
|
|
let listener = announcement_clone.listener();
|
|
listener.listen().await
|
|
});
|
|
|
|
close_handle.await.unwrap();
|
|
let result = listen_handle.await.unwrap();
|
|
|
|
// Must see either Closed error or timeout (but not hang)
|
|
assert_eq!(result, Err(ListenError::Closed));
|
|
}
|
|
}
|
|
|
|
#[cfg(feature = "std")]
|
|
/// Test: Race condition test: close during timeout blocking
|
|
///
|
|
/// Verifies: Verifies no race condition in close during timeout blocking
|
|
/// Category: race
|
|
#[test]
|
|
fn test_close_during_timeout_blocking() {
|
|
for iteration in 0..500 {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
let handle =
|
|
thread::spawn(move || listener.listen_timeout_blocking(Duration::from_secs(2)));
|
|
|
|
// Close at varying times
|
|
if iteration % 2 == 0 {
|
|
thread::sleep(Duration::from_micros(50));
|
|
}
|
|
announcer.close();
|
|
|
|
let result = handle.join().unwrap();
|
|
assert_eq!(result, Err(ListenError::Closed));
|
|
}
|
|
}
|
|
|
|
/// Test: Race condition test: close with barrier sync
|
|
///
|
|
/// Verifies: Verifies no race condition in close with barrier sync
|
|
/// Category: race
|
|
#[tokio::test]
|
|
async fn test_rapid_close_listen_cycles() {
|
|
for _ in 0..1_000 {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
let listen_handle = tokio::spawn(async move { listener.listen().await });
|
|
|
|
// Close immediately
|
|
announcer.close();
|
|
|
|
let result = listen_handle.await.unwrap();
|
|
assert_eq!(result, Err(ListenError::Closed));
|
|
}
|
|
}
|
|
|
|
#[cfg(feature = "std")]
|
|
/// Test: Race condition test: close with barrier sync
|
|
///
|
|
/// Verifies: Verifies no race condition in close with barrier sync
|
|
/// Category: race
|
|
#[test]
|
|
fn test_close_with_barrier_sync() {
|
|
for _ in 0..500 {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let barrier = Arc::new(Barrier::new(11)); // 10 listeners + closer
|
|
|
|
let mut handles = Vec::new();
|
|
|
|
// 10 threads that will block
|
|
for _ in 0..10 {
|
|
let listener = announcement.listener();
|
|
let barrier_clone = barrier.clone();
|
|
|
|
handles.push(thread::spawn(move || {
|
|
barrier_clone.wait(); // Sync start
|
|
listener.listen_blocking()
|
|
}));
|
|
}
|
|
|
|
// Closer thread
|
|
let closer = thread::spawn(move || {
|
|
barrier.wait(); // Sync with listeners
|
|
// Close immediately after barrier
|
|
announcer.close();
|
|
});
|
|
|
|
closer.join().unwrap();
|
|
|
|
// All listeners should see Closed
|
|
for handle in handles {
|
|
let result = handle.join().unwrap();
|
|
assert_eq!(result, Err(ListenError::Closed));
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Test: Stress test: close propagation stress
|
|
///
|
|
/// Verifies: Verifies system handles close propagation stress correctly under stress
|
|
/// Category: race
|
|
#[tokio::test]
|
|
async fn test_close_detection_is_closed() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
assert!(!announcement.is_closed());
|
|
assert!(!listener.is_closed());
|
|
|
|
// Close in background
|
|
tokio::spawn(async move {
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
announcer.close();
|
|
});
|
|
|
|
// Poll until closed
|
|
let start = tokio::time::Instant::now();
|
|
loop {
|
|
if listener.is_closed() {
|
|
break;
|
|
}
|
|
if start.elapsed() > Duration::from_secs(1) {
|
|
panic!("is_closed() never became true");
|
|
}
|
|
tokio::time::sleep(Duration::from_micros(100)).await;
|
|
}
|
|
|
|
// Now listen() should return Closed immediately
|
|
let result = listener.listen().await;
|
|
assert_eq!(result, Err(ListenError::Closed));
|
|
}
|
|
|
|
#[cfg(feature = "std")]
|
|
/// Test: Stress test: close propagation stress
|
|
///
|
|
/// Verifies: Verifies system handles close propagation stress correctly under stress
|
|
/// Category: race
|
|
#[test]
|
|
fn test_close_propagation_stress() {
|
|
let running = Arc::new(AtomicBool::new(true));
|
|
|
|
let mut handles = Vec::new();
|
|
|
|
// Run for 5 seconds
|
|
for _ in 0..10 {
|
|
let running_clone = running.clone();
|
|
|
|
handles.push(thread::spawn(move || {
|
|
let mut count = 0;
|
|
|
|
while running_clone.load(Ordering::Relaxed) {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
// Spawn blocking thread
|
|
let h = thread::spawn(move || listener.listen_blocking());
|
|
|
|
// Close quickly
|
|
thread::yield_now();
|
|
announcer.close();
|
|
|
|
let result = h.join().unwrap();
|
|
assert_eq!(result, Err(ListenError::Closed));
|
|
|
|
count += 1;
|
|
}
|
|
|
|
count
|
|
}));
|
|
}
|
|
|
|
thread::sleep(Duration::from_secs(5));
|
|
running.store(false, Ordering::Relaxed);
|
|
|
|
let total: usize = handles.into_iter().map(|h| h.join().unwrap()).sum();
|
|
println!("Close propagation stress: {} successful cycles", total);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_close_with_try_listen_checks() {
|
|
for _ in 0..1_000 {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
// Task doing rapid try_listen
|
|
let check_handle = tokio::spawn(async move {
|
|
for _ in 0..100 {
|
|
let _ = listener.try_listen();
|
|
tokio::task::yield_now().await;
|
|
}
|
|
listener
|
|
});
|
|
|
|
// Close
|
|
tokio::task::yield_now().await;
|
|
announcer.close();
|
|
|
|
// Get listener back and verify it sees closed
|
|
let listener = check_handle.await.unwrap();
|
|
assert!(listener.is_closed());
|
|
assert_eq!(listener.try_listen(), None);
|
|
}
|
|
}
|