bjoernager/rust
bjoernager/rust
1
Fork 0
Code Activity

Move std::sync unit tests to integration tests

Browse source 
This removes two minor OnceLock tests which test private methods. The
rest of the tests should be more than enough to catch mistakes in those
private methods. Also makes ReentrantLock::try_lock public. And finally
it makes the mpmc tests actually run.
This commit is contained in:
bjorn3 2025-01-17 11:04:40 +00:00
parent 332fb7e6f1
commit b8ae372e48
22 changed files with 101 additions and 99 deletions
Download patch file Download diff file Expand all files Collapse all files

3
library/std/src/sync/barrier.rs
View file

@ -1,6 +1,3 @@
#[cfg(test)]
mod tests;
use crate::fmt;
// FIXME(nonpoison_mutex,nonpoison_condvar): switch to nonpoison versions once they are available
use crate::sync::{Condvar, Mutex};

35
library/std/src/sync/barrier/tests.rs
View file

@ -1,35 +0,0 @@
use crate::sync::mpsc::{TryRecvError, channel};
use crate::sync::{Arc, Barrier};
use crate::thread;
#[test]
#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads
fn test_barrier() {
const N: usize = 10;
let barrier = Arc::new(Barrier::new(N));
let (tx, rx) = channel();
for _ in 0..N - 1 {
let c = barrier.clone();
let tx = tx.clone();
thread::spawn(move || {
tx.send(c.wait().is_leader()).unwrap();
});
}
// At this point, all spawned threads should be blocked,
// so we shouldn't get anything from the port
assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
let mut leader_found = barrier.wait().is_leader();
// Now, the barrier is cleared and we should get data.
for _ in 0..N - 1 {
if rx.recv().unwrap() {
assert!(!leader_found);
leader_found = true;
}
}
assert!(leader_found);
}

3
library/std/src/sync/lazy_lock.rs
View file

@ -350,6 +350,3 @@ unsafe impl<T: Sync + Send, F: Send> Sync for LazyLock<T, F> {}
impl<T: RefUnwindSafe + UnwindSafe, F: UnwindSafe> RefUnwindSafe for LazyLock<T, F> {}
#[stable(feature = "lazy_cell", since = "1.80.0")]
impl<T: UnwindSafe, F: UnwindSafe> UnwindSafe for LazyLock<T, F> {}
#[cfg(test)]
mod tests;

167
library/std/src/sync/lazy_lock/tests.rs
View file

@ -1,167 +0,0 @@
use crate::cell::LazyCell;
use crate::sync::atomic::AtomicUsize;
use crate::sync::atomic::Ordering::SeqCst;
use crate::sync::{LazyLock, Mutex, OnceLock};
use crate::{panic, thread};
fn spawn_and_wait<R: Send + 'static>(f: impl FnOnce() -> R + Send + 'static) -> R {
thread::spawn(f).join().unwrap()
}
#[test]
fn lazy_default() {
static CALLED: AtomicUsize = AtomicUsize::new(0);
struct Foo(u8);
impl Default for Foo {
fn default() -> Self {
CALLED.fetch_add(1, SeqCst);
Foo(42)
}
}
let lazy: LazyCell<Mutex<Foo>> = <_>::default();
assert_eq!(CALLED.load(SeqCst), 0);
assert_eq!(lazy.lock().unwrap().0, 42);
assert_eq!(CALLED.load(SeqCst), 1);
lazy.lock().unwrap().0 = 21;
assert_eq!(lazy.lock().unwrap().0, 21);
assert_eq!(CALLED.load(SeqCst), 1);
}
#[test]
#[cfg_attr(not(panic = "unwind"), ignore = "test requires unwinding support")]
fn lazy_poisoning() {
let x: LazyCell<String> = LazyCell::new(|| panic!("kaboom"));
for _ in 0..2 {
let res = panic::catch_unwind(panic::AssertUnwindSafe(|| x.len()));
assert!(res.is_err());
}
}
#[test]
#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads
fn sync_lazy_new() {
static CALLED: AtomicUsize = AtomicUsize::new(0);
static SYNC_LAZY: LazyLock<i32> = LazyLock::new(|| {
CALLED.fetch_add(1, SeqCst);
92
});
assert_eq!(CALLED.load(SeqCst), 0);
spawn_and_wait(|| {
let y = *SYNC_LAZY - 30;
assert_eq!(y, 62);
assert_eq!(CALLED.load(SeqCst), 1);
});
let y = *SYNC_LAZY - 30;
assert_eq!(y, 62);
assert_eq!(CALLED.load(SeqCst), 1);
}
#[test]
fn sync_lazy_default() {
static CALLED: AtomicUsize = AtomicUsize::new(0);
struct Foo(u8);
impl Default for Foo {
fn default() -> Self {
CALLED.fetch_add(1, SeqCst);
Foo(42)
}
}
let lazy: LazyLock<Mutex<Foo>> = <_>::default();
assert_eq!(CALLED.load(SeqCst), 0);
assert_eq!(lazy.lock().unwrap().0, 42);
assert_eq!(CALLED.load(SeqCst), 1);
lazy.lock().unwrap().0 = 21;
assert_eq!(lazy.lock().unwrap().0, 21);
assert_eq!(CALLED.load(SeqCst), 1);
}
#[test]
#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads
fn static_sync_lazy() {
static XS: LazyLock<Vec<i32>> = LazyLock::new(|| {
let mut xs = Vec::new();
xs.push(1);
xs.push(2);
xs.push(3);
xs
});
spawn_and_wait(|| {
assert_eq!(&*XS, &vec![1, 2, 3]);
});
assert_eq!(&*XS, &vec![1, 2, 3]);
}
#[test]
fn static_sync_lazy_via_fn() {
fn xs() -> &'static Vec<i32> {
static XS: OnceLock<Vec<i32>> = OnceLock::new();
XS.get_or_init(|| {
let mut xs = Vec::new();
xs.push(1);
xs.push(2);
xs.push(3);
xs
})
}
assert_eq!(xs(), &vec![1, 2, 3]);
}
#[test]
#[cfg_attr(not(panic = "unwind"), ignore = "test requires unwinding support")]
fn sync_lazy_poisoning() {
let x: LazyLock<String> = LazyLock::new(|| panic!("kaboom"));
for _ in 0..2 {
let res = panic::catch_unwind(|| x.len());
assert!(res.is_err());
}
}
// Check that we can infer `T` from closure's type.
#[test]
fn lazy_type_inference() {
let _ = LazyCell::new(|| ());
}
#[test]
fn is_sync_send() {
fn assert_traits<T: Send + Sync>() {}
assert_traits::<LazyLock<String>>();
}
#[test]
#[should_panic = "has previously been poisoned"]
fn lazy_force_mut_panic() {
let mut lazy = LazyLock::<String>::new(|| panic!());
crate::panic::catch_unwind(crate::panic::AssertUnwindSafe(|| {
let _ = LazyLock::force_mut(&mut lazy);
}))
.unwrap_err();
let _ = &*lazy;
}
#[test]
fn lazy_force_mut() {
let s = "abc".to_owned();
let mut lazy = LazyLock::new(move || s);
LazyLock::force_mut(&mut lazy);
let p = LazyLock::force_mut(&mut lazy);
p.clear();
LazyLock::force_mut(&mut lazy);
}

728
library/std/src/sync/mpmc/tests.rs
View file

@ -1,728 +0,0 @@
use super::*;
use crate::{env, thread};
pub fn stress_factor() -> usize {
match env::var("RUST_TEST_STRESS") {
Ok(val) => val.parse().unwrap(),
Err(..) => 1,
}
}
#[test]
fn smoke() {
let (tx, rx) = channel::<i32>();
tx.send(1).unwrap();
assert_eq!(rx.recv().unwrap(), 1);
}
#[test]
fn drop_full() {
let (tx, _rx) = channel::<Box<isize>>();
tx.send(Box::new(1)).unwrap();
}
#[test]
fn drop_full_shared() {
let (tx, _rx) = channel::<Box<isize>>();
drop(tx.clone());
drop(tx.clone());
tx.send(Box::new(1)).unwrap();
}
#[test]
fn smoke_shared() {
let (tx, rx) = channel::<i32>();
tx.send(1).unwrap();
assert_eq!(rx.recv().unwrap(), 1);
let tx = tx.clone();
tx.send(1).unwrap();
assert_eq!(rx.recv().unwrap(), 1);
}
#[test]
fn smoke_threads() {
let (tx, rx) = channel::<i32>();
let t1 = thread::spawn(move || {
for i in 0..2 {
tx.send(i).unwrap();
}
});
let t2 = thread::spawn(move || {
assert_eq!(rx.recv().unwrap(), 0);
assert_eq!(rx.recv().unwrap(), 1);
});
t1.join().unwrap();
t2.join().unwrap();
}
#[test]
fn smoke_port_gone() {
let (tx, rx) = channel::<i32>();
drop(rx);
assert!(tx.send(1).is_err());
}
#[test]
fn smoke_shared_port_gone() {
let (tx, rx) = channel::<i32>();
drop(rx);
assert!(tx.send(1).is_err())
}
#[test]
fn smoke_shared_port_gone2() {
let (tx, rx) = channel::<i32>();
drop(rx);
let tx2 = tx.clone();
drop(tx);
assert!(tx2.send(1).is_err());
}
#[test]
fn port_gone_concurrent() {
let (tx, rx) = channel::<i32>();
let _t = thread::spawn(move || {
rx.recv().unwrap();
});
while tx.send(1).is_ok() {}
}
#[test]
fn port_gone_concurrent_shared() {
let (tx, rx) = channel::<i32>();
let tx2 = tx.clone();
let _t = thread::spawn(move || {
rx.recv().unwrap();
});
while tx.send(1).is_ok() && tx2.send(1).is_ok() {}
}
#[test]
fn smoke_chan_gone() {
let (tx, rx) = channel::<i32>();
drop(tx);
assert!(rx.recv().is_err());
}
#[test]
fn smoke_chan_gone_shared() {
let (tx, rx) = channel::<()>();
let tx2 = tx.clone();
drop(tx);
drop(tx2);
assert!(rx.recv().is_err());
}
#[test]
fn chan_gone_concurrent() {
let (tx, rx) = channel::<i32>();
let _t = thread::spawn(move || {
tx.send(1).unwrap();
tx.send(1).unwrap();
});
while rx.recv().is_ok() {}
}
#[test]
fn stress() {
let count = if cfg!(miri) { 100 } else { 10000 };
let (tx, rx) = channel::<i32>();
let t = thread::spawn(move || {
for _ in 0..count {
tx.send(1).unwrap();
}
});
for _ in 0..count {
assert_eq!(rx.recv().unwrap(), 1);
}
t.join().ok().expect("thread panicked");
}
#[test]
fn stress_shared() {
const AMT: u32 = if cfg!(miri) { 100 } else { 10000 };
const NTHREADS: u32 = 8;
let (tx, rx) = channel::<i32>();
let t = thread::spawn(move || {
for _ in 0..AMT * NTHREADS {
assert_eq!(rx.recv().unwrap(), 1);
}
match rx.try_recv() {
Ok(..) => panic!(),
_ => {}
}
});
for _ in 0..NTHREADS {
let tx = tx.clone();
thread::spawn(move || {
for _ in 0..AMT {
tx.send(1).unwrap();
}
});
}
drop(tx);
t.join().ok().expect("thread panicked");
}
#[test]
fn send_from_outside_runtime() {
let (tx1, rx1) = channel::<()>();
let (tx2, rx2) = channel::<i32>();
let t1 = thread::spawn(move || {
tx1.send(()).unwrap();
for _ in 0..40 {
assert_eq!(rx2.recv().unwrap(), 1);
}
});
rx1.recv().unwrap();
let t2 = thread::spawn(move || {
for _ in 0..40 {
tx2.send(1).unwrap();
}
});
t1.join().ok().expect("thread panicked");
t2.join().ok().expect("thread panicked");
}
#[test]
fn recv_from_outside_runtime() {
let (tx, rx) = channel::<i32>();
let t = thread::spawn(move || {
for _ in 0..40 {
assert_eq!(rx.recv().unwrap(), 1);
}
});
for _ in 0..40 {
tx.send(1).unwrap();
}
t.join().ok().expect("thread panicked");
}
#[test]
fn no_runtime() {
let (tx1, rx1) = channel::<i32>();
let (tx2, rx2) = channel::<i32>();
let t1 = thread::spawn(move || {
assert_eq!(rx1.recv().unwrap(), 1);
tx2.send(2).unwrap();
});
let t2 = thread::spawn(move || {
tx1.send(1).unwrap();
assert_eq!(rx2.recv().unwrap(), 2);
});
t1.join().ok().expect("thread panicked");
t2.join().ok().expect("thread panicked");
}
#[test]
fn oneshot_single_thread_close_port_first() {
// Simple test of closing without sending
let (_tx, rx) = channel::<i32>();
drop(rx);
}
#[test]
fn oneshot_single_thread_close_chan_first() {
// Simple test of closing without sending
let (tx, _rx) = channel::<i32>();
drop(tx);
}
#[test]
fn oneshot_single_thread_send_port_close() {
// Testing that the sender cleans up the payload if receiver is closed
let (tx, rx) = channel::<Box<i32>>();
drop(rx);
assert!(tx.send(Box::new(0)).is_err());
}
#[test]
fn oneshot_single_thread_recv_chan_close() {
// Receiving on a closed chan will panic
let res = thread::spawn(move || {
let (tx, rx) = channel::<i32>();
drop(tx);
rx.recv().unwrap();
})
.join();
// What is our res?
assert!(res.is_err());
}
#[test]
fn oneshot_single_thread_send_then_recv() {
let (tx, rx) = channel::<Box<i32>>();
tx.send(Box::new(10)).unwrap();
assert!(*rx.recv().unwrap() == 10);
}
#[test]
fn oneshot_single_thread_try_send_open() {
let (tx, rx) = channel::<i32>();
assert!(tx.send(10).is_ok());
assert!(rx.recv().unwrap() == 10);
}
#[test]
fn oneshot_single_thread_try_send_closed() {
let (tx, rx) = channel::<i32>();
drop(rx);
assert!(tx.send(10).is_err());
}
#[test]
fn oneshot_single_thread_try_recv_open() {
let (tx, rx) = channel::<i32>();
tx.send(10).unwrap();
assert!(rx.recv() == Ok(10));
}
#[test]
fn oneshot_single_thread_try_recv_closed() {
let (tx, rx) = channel::<i32>();
drop(tx);
assert!(rx.recv().is_err());
}
#[test]
fn oneshot_single_thread_peek_data() {
let (tx, rx) = channel::<i32>();
assert_eq!(rx.try_recv(), Err(TryRecvError::Empty));
tx.send(10).unwrap();
assert_eq!(rx.try_recv(), Ok(10));
}
#[test]
fn oneshot_single_thread_peek_close() {
let (tx, rx) = channel::<i32>();
drop(tx);
assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected));
assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected));
}
#[test]
fn oneshot_single_thread_peek_open() {
let (_tx, rx) = channel::<i32>();
assert_eq!(rx.try_recv(), Err(TryRecvError::Empty));
}
#[test]
fn oneshot_multi_task_recv_then_send() {
let (tx, rx) = channel::<Box<i32>>();
let _t = thread::spawn(move || {
assert!(*rx.recv().unwrap() == 10);
});
tx.send(Box::new(10)).unwrap();
}
#[test]
fn oneshot_multi_task_recv_then_close() {
let (tx, rx) = channel::<Box<i32>>();
let _t = thread::spawn(move || {
drop(tx);
});
let res = thread::spawn(move || {
assert!(*rx.recv().unwrap() == 10);
})
.join();
assert!(res.is_err());
}
#[test]
fn oneshot_multi_thread_close_stress() {
for _ in 0..stress_factor() {
let (tx, rx) = channel::<i32>();
let _t = thread::spawn(move || {
drop(rx);
});
drop(tx);
}
}
#[test]
fn oneshot_multi_thread_send_close_stress() {
for _ in 0..stress_factor() {
let (tx, rx) = channel::<i32>();
let _t = thread::spawn(move || {
drop(rx);
});
let _ = thread::spawn(move || {
tx.send(1).unwrap();
})
.join();
}
}
#[test]
fn oneshot_multi_thread_recv_close_stress() {
for _ in 0..stress_factor() {
let (tx, rx) = channel::<i32>();
thread::spawn(move || {
let res = thread::spawn(move || {
rx.recv().unwrap();
})
.join();
assert!(res.is_err());
});
let _t = thread::spawn(move || {
thread::spawn(move || {
drop(tx);
});
});
}
}
#[test]
fn oneshot_multi_thread_send_recv_stress() {
for _ in 0..stress_factor() {
let (tx, rx) = channel::<Box<isize>>();
let _t = thread::spawn(move || {
tx.send(Box::new(10)).unwrap();
});
assert!(*rx.recv().unwrap() == 10);
}
}
#[test]
fn stream_send_recv_stress() {
for _ in 0..stress_factor() {
let (tx, rx) = channel();
send(tx, 0);
recv(rx, 0);
fn send(tx: Sender<Box<i32>>, i: i32) {
if i == 10 {
return;
}
thread::spawn(move || {
tx.send(Box::new(i)).unwrap();
send(tx, i + 1);
});
}
fn recv(rx: Receiver<Box<i32>>, i: i32) {
if i == 10 {
return;
}
thread::spawn(move || {
assert!(*rx.recv().unwrap() == i);
recv(rx, i + 1);
});
}
}
}
#[test]
fn oneshot_single_thread_recv_timeout() {
let (tx, rx) = channel();
tx.send(()).unwrap();
assert_eq!(rx.recv_timeout(Duration::from_millis(1)), Ok(()));
assert_eq!(rx.recv_timeout(Duration::from_millis(1)), Err(RecvTimeoutError::Timeout));
tx.send(()).unwrap();
assert_eq!(rx.recv_timeout(Duration::from_millis(1)), Ok(()));
}
#[test]
fn stress_recv_timeout_two_threads() {
let (tx, rx) = channel();
let stress = stress_factor() + 100;
let timeout = Duration::from_millis(100);
thread::spawn(move || {
for i in 0..stress {
if i % 2 == 0 {
thread::sleep(timeout * 2);
}
tx.send(1usize).unwrap();
}
});
let mut recv_count = 0;
loop {
match rx.recv_timeout(timeout) {
Ok(n) => {
assert_eq!(n, 1usize);
recv_count += 1;
}
Err(RecvTimeoutError::Timeout) => continue,
Err(RecvTimeoutError::Disconnected) => break,
}
}
assert_eq!(recv_count, stress);
}
#[test]
fn recv_timeout_upgrade() {
let (tx, rx) = channel::<()>();
let timeout = Duration::from_millis(1);
let _tx_clone = tx.clone();
let start = Instant::now();
assert_eq!(rx.recv_timeout(timeout), Err(RecvTimeoutError::Timeout));
assert!(Instant::now() >= start + timeout);
}
#[test]
fn stress_recv_timeout_shared() {
let (tx, rx) = channel();
let stress = stress_factor() + 100;
for i in 0..stress {
let tx = tx.clone();
thread::spawn(move || {
thread::sleep(Duration::from_millis(i as u64 * 10));
tx.send(1usize).unwrap();
});
}
drop(tx);
let mut recv_count = 0;
loop {
match rx.recv_timeout(Duration::from_millis(10)) {
Ok(n) => {
assert_eq!(n, 1usize);
recv_count += 1;
}
Err(RecvTimeoutError::Timeout) => continue,
Err(RecvTimeoutError::Disconnected) => break,
}
}
assert_eq!(recv_count, stress);
}
#[test]
fn very_long_recv_timeout_wont_panic() {
let (tx, rx) = channel::<()>();
let join_handle = thread::spawn(move || rx.recv_timeout(Duration::from_secs(u64::MAX)));
thread::sleep(Duration::from_secs(1));
assert!(tx.send(()).is_ok());
assert_eq!(join_handle.join().unwrap(), Ok(()));
}
#[test]
fn recv_a_lot() {
let count = if cfg!(miri) { 1000 } else { 10000 };
// Regression test that we don't run out of stack in scheduler context
let (tx, rx) = channel();
for _ in 0..count {
tx.send(()).unwrap();
}
for _ in 0..count {
rx.recv().unwrap();
}
}
#[test]
fn shared_recv_timeout() {
let (tx, rx) = channel();
let total = 5;
for _ in 0..total {
let tx = tx.clone();
thread::spawn(move || {
tx.send(()).unwrap();
});
}
for _ in 0..total {
rx.recv().unwrap();
}
assert_eq!(rx.recv_timeout(Duration::from_millis(1)), Err(RecvTimeoutError::Timeout));
tx.send(()).unwrap();
assert_eq!(rx.recv_timeout(Duration::from_millis(1)), Ok(()));
}
#[test]
fn shared_chan_stress() {
let (tx, rx) = channel();
let total = stress_factor() + 100;
for _ in 0..total {
let tx = tx.clone();
thread::spawn(move || {
tx.send(()).unwrap();
});
}
for _ in 0..total {
rx.recv().unwrap();
}
}
#[test]
fn test_nested_recv_iter() {
let (tx, rx) = channel::<i32>();
let (total_tx, total_rx) = channel::<i32>();
let _t = thread::spawn(move || {
let mut acc = 0;
for x in rx.iter() {
acc += x;
}
total_tx.send(acc).unwrap();
});
tx.send(3).unwrap();
tx.send(1).unwrap();
tx.send(2).unwrap();
drop(tx);
assert_eq!(total_rx.recv().unwrap(), 6);
}
#[test]
fn test_recv_iter_break() {
let (tx, rx) = channel::<i32>();
let (count_tx, count_rx) = channel();
let _t = thread::spawn(move || {
let mut count = 0;
for x in rx.iter() {
if count >= 3 {
break;
} else {
count += x;
}
}
count_tx.send(count).unwrap();
});
tx.send(2).unwrap();
tx.send(2).unwrap();
tx.send(2).unwrap();
let _ = tx.send(2);
drop(tx);
assert_eq!(count_rx.recv().unwrap(), 4);
}
#[test]
fn test_recv_try_iter() {
let (request_tx, request_rx) = channel();
let (response_tx, response_rx) = channel();
// Request `x`s until we have `6`.
let t = thread::spawn(move || {
let mut count = 0;
loop {
for x in response_rx.try_iter() {
count += x;
if count == 6 {
return count;
}
}
request_tx.send(()).unwrap();
}
});
for _ in request_rx.iter() {
if response_tx.send(2).is_err() {
break;
}
}
assert_eq!(t.join().unwrap(), 6);
}
#[test]
fn test_recv_into_iter_owned() {
let mut iter = {
let (tx, rx) = channel::<i32>();
tx.send(1).unwrap();
tx.send(2).unwrap();
rx.into_iter()
};
assert_eq!(iter.next().unwrap(), 1);
assert_eq!(iter.next().unwrap(), 2);
assert_eq!(iter.next().is_none(), true);
}
#[test]
fn test_recv_into_iter_borrowed() {
let (tx, rx) = channel::<i32>();
tx.send(1).unwrap();
tx.send(2).unwrap();
drop(tx);
let mut iter = (&rx).into_iter();
assert_eq!(iter.next().unwrap(), 1);
assert_eq!(iter.next().unwrap(), 2);
assert_eq!(iter.next().is_none(), true);
}
#[test]
fn try_recv_states() {
let (tx1, rx1) = channel::<i32>();
let (tx2, rx2) = channel::<()>();
let (tx3, rx3) = channel::<()>();
let _t = thread::spawn(move || {
rx2.recv().unwrap();
tx1.send(1).unwrap();
tx3.send(()).unwrap();
rx2.recv().unwrap();
drop(tx1);
tx3.send(()).unwrap();
});
assert_eq!(rx1.try_recv(), Err(TryRecvError::Empty));
tx2.send(()).unwrap();
rx3.recv().unwrap();
assert_eq!(rx1.try_recv(), Ok(1));
assert_eq!(rx1.try_recv(), Err(TryRecvError::Empty));
tx2.send(()).unwrap();
rx3.recv().unwrap();
assert_eq!(rx1.try_recv(), Err(TryRecvError::Disconnected));
}
// This bug used to end up in a livelock inside of the Receiver destructor
// because the internal state of the Shared packet was corrupted
#[test]
fn destroy_upgraded_shared_port_when_sender_still_active() {
let (tx, rx) = channel();
let (tx2, rx2) = channel();
let _t = thread::spawn(move || {
rx.recv().unwrap(); // wait on a oneshot
drop(rx); // destroy a shared
tx2.send(()).unwrap();
});
// make sure the other thread has gone to sleep
for _ in 0..5000 {
thread::yield_now();
}
// upgrade to a shared chan and send a message
let t = tx.clone();
drop(tx);
t.send(()).unwrap();
// wait for the child thread to exit before we exit
rx2.recv().unwrap();
}
#[test]
fn issue_32114() {
let (tx, _) = channel();
let _ = tx.send(123);
assert_eq!(tx.send(123), Err(SendError(123)));
}
#[test]
fn issue_39364() {
let (tx, rx) = channel::<()>();
let t = thread::spawn(move || {
thread::sleep(Duration::from_millis(300));
let _ = tx.clone();
// Don't drop; hand back to caller.
tx
});
let _ = rx.recv_timeout(Duration::from_millis(500));
let _tx = t.join().unwrap(); // delay dropping until end of test
let _ = rx.recv_timeout(Duration::from_millis(500));
}

13
library/std/src/sync/mpsc/mod.rs → library/std/src/sync/mpsc.rs
View file

@ -137,12 +137,6 @@
#![stable(feature = "rust1", since = "1.0.0")]
#[cfg(all(test, not(any(target_os = "emscripten", target_os = "wasi"))))]
mod tests;
#[cfg(all(test, not(any(target_os = "emscripten", target_os = "wasi"))))]
mod sync_tests;
// MPSC channels are built as a wrapper around MPMC channels, which
// were ported from the `crossbeam-channel` crate. MPMC channels are
// not exposed publicly, but if you are curious about the implementation,
@ -737,9 +731,10 @@ impl<T> SyncSender<T> {
// Attempts to send for a value on this receiver, returning an error if the
// corresponding channel has hung up, or if it waits more than `timeout`.
//
// This method is currently private and only used for tests.
#[allow(unused)]
fn send_timeout(&self, t: T, timeout: Duration) -> Result<(), mpmc::SendTimeoutError<T>> {
// This method is currently only used for tests.
#[unstable(issue = "none", feature = "std_internals")]
#[doc(hidden)]
pub fn send_timeout(&self, t: T, timeout: Duration) -> Result<(), mpmc::SendTimeoutError<T>> {
self.inner.send_timeout(t, timeout)
}
}

669
library/std/src/sync/mpsc/sync_tests.rs
View file

@ -1,669 +0,0 @@
use super::*;
use crate::rc::Rc;
use crate::sync::mpmc::SendTimeoutError;
use crate::{env, thread};
pub fn stress_factor() -> usize {
match env::var("RUST_TEST_STRESS") {
Ok(val) => val.parse().unwrap(),
Err(..) => 1,
}
}
#[test]
fn smoke() {
let (tx, rx) = sync_channel::<i32>(1);
tx.send(1).unwrap();
assert_eq!(rx.recv().unwrap(), 1);
}
#[test]
fn drop_full() {
let (tx, _rx) = sync_channel::<Box<isize>>(1);
tx.send(Box::new(1)).unwrap();
}
#[test]
fn smoke_shared() {
let (tx, rx) = sync_channel::<i32>(1);
tx.send(1).unwrap();
assert_eq!(rx.recv().unwrap(), 1);
let tx = tx.clone();
tx.send(1).unwrap();
assert_eq!(rx.recv().unwrap(), 1);
}
#[test]
fn recv_timeout() {
let (tx, rx) = sync_channel::<i32>(1);
assert_eq!(rx.recv_timeout(Duration::from_millis(1)), Err(RecvTimeoutError::Timeout));
tx.send(1).unwrap();
assert_eq!(rx.recv_timeout(Duration::from_millis(1)), Ok(1));
}
#[test]
fn send_timeout() {
let (tx, _rx) = sync_channel::<i32>(1);
assert_eq!(tx.send_timeout(1, Duration::from_millis(1)), Ok(()));
assert_eq!(tx.send_timeout(1, Duration::from_millis(1)), Err(SendTimeoutError::Timeout(1)));
}
#[test]
fn smoke_threads() {
let (tx, rx) = sync_channel::<i32>(0);
let _t = thread::spawn(move || {
tx.send(1).unwrap();
});
assert_eq!(rx.recv().unwrap(), 1);
}
#[test]
fn smoke_port_gone() {
let (tx, rx) = sync_channel::<i32>(0);
drop(rx);
assert!(tx.send(1).is_err());
}
#[test]
fn smoke_shared_port_gone2() {
let (tx, rx) = sync_channel::<i32>(0);
drop(rx);
let tx2 = tx.clone();
drop(tx);
assert!(tx2.send(1).is_err());
}
#[test]
fn port_gone_concurrent() {
let (tx, rx) = sync_channel::<i32>(0);
let _t = thread::spawn(move || {
rx.recv().unwrap();
});
while tx.send(1).is_ok() {}
}
#[test]
fn port_gone_concurrent_shared() {
let (tx, rx) = sync_channel::<i32>(0);
let tx2 = tx.clone();
let _t = thread::spawn(move || {
rx.recv().unwrap();
});
while tx.send(1).is_ok() && tx2.send(1).is_ok() {}
}
#[test]
fn smoke_chan_gone() {
let (tx, rx) = sync_channel::<i32>(0);
drop(tx);
assert!(rx.recv().is_err());
}
#[test]
fn smoke_chan_gone_shared() {
let (tx, rx) = sync_channel::<()>(0);
let tx2 = tx.clone();
drop(tx);
drop(tx2);
assert!(rx.recv().is_err());
}
#[test]
fn chan_gone_concurrent() {
let (tx, rx) = sync_channel::<i32>(0);
thread::spawn(move || {
tx.send(1).unwrap();
tx.send(1).unwrap();
});
while rx.recv().is_ok() {}
}
#[test]
fn stress() {
let count = if cfg!(miri) { 100 } else { 10000 };
let (tx, rx) = sync_channel::<i32>(0);
thread::spawn(move || {
for _ in 0..count {
tx.send(1).unwrap();
}
});
for _ in 0..count {
assert_eq!(rx.recv().unwrap(), 1);
}
}
#[test]
fn stress_recv_timeout_two_threads() {
let count = if cfg!(miri) { 100 } else { 10000 };
let (tx, rx) = sync_channel::<i32>(0);
thread::spawn(move || {
for _ in 0..count {
tx.send(1).unwrap();
}
});
let mut recv_count = 0;
loop {
match rx.recv_timeout(Duration::from_millis(1)) {
Ok(v) => {
assert_eq!(v, 1);
recv_count += 1;
}
Err(RecvTimeoutError::Timeout) => continue,
Err(RecvTimeoutError::Disconnected) => break,
}
}
assert_eq!(recv_count, count);
}
#[test]
fn stress_recv_timeout_shared() {
const AMT: u32 = if cfg!(miri) { 100 } else { 1000 };
const NTHREADS: u32 = 8;
let (tx, rx) = sync_channel::<i32>(0);
let (dtx, drx) = sync_channel::<()>(0);
thread::spawn(move || {
let mut recv_count = 0;
loop {
match rx.recv_timeout(Duration::from_millis(10)) {
Ok(v) => {
assert_eq!(v, 1);
recv_count += 1;
}
Err(RecvTimeoutError::Timeout) => continue,
Err(RecvTimeoutError::Disconnected) => break,
}
}
assert_eq!(recv_count, AMT * NTHREADS);
assert!(rx.try_recv().is_err());
dtx.send(()).unwrap();
});
for _ in 0..NTHREADS {
let tx = tx.clone();
thread::spawn(move || {
for _ in 0..AMT {
tx.send(1).unwrap();
}
});
}
drop(tx);
drx.recv().unwrap();
}
#[test]
fn stress_shared() {
const AMT: u32 = if cfg!(miri) { 100 } else { 1000 };
const NTHREADS: u32 = 8;
let (tx, rx) = sync_channel::<i32>(0);
let (dtx, drx) = sync_channel::<()>(0);
thread::spawn(move || {
for _ in 0..AMT * NTHREADS {
assert_eq!(rx.recv().unwrap(), 1);
}
match rx.try_recv() {
Ok(..) => panic!(),
_ => {}
}
dtx.send(()).unwrap();
});
for _ in 0..NTHREADS {
let tx = tx.clone();
thread::spawn(move || {
for _ in 0..AMT {
tx.send(1).unwrap();
}
});
}
drop(tx);
drx.recv().unwrap();
}
#[test]
fn oneshot_single_thread_close_port_first() {
// Simple test of closing without sending
let (_tx, rx) = sync_channel::<i32>(0);
drop(rx);
}
#[test]
fn oneshot_single_thread_close_chan_first() {
// Simple test of closing without sending
let (tx, _rx) = sync_channel::<i32>(0);
drop(tx);
}
#[test]
fn oneshot_single_thread_send_port_close() {
// Testing that the sender cleans up the payload if receiver is closed
let (tx, rx) = sync_channel::<Box<i32>>(0);
drop(rx);
assert!(tx.send(Box::new(0)).is_err());
}
#[test]
fn oneshot_single_thread_recv_chan_close() {
// Receiving on a closed chan will panic
let res = thread::spawn(move || {
let (tx, rx) = sync_channel::<i32>(0);
drop(tx);
rx.recv().unwrap();
})
.join();
// What is our res?
assert!(res.is_err());
}
#[test]
fn oneshot_single_thread_send_then_recv() {
let (tx, rx) = sync_channel::<Box<i32>>(1);
tx.send(Box::new(10)).unwrap();
assert!(*rx.recv().unwrap() == 10);
}
#[test]
fn oneshot_single_thread_try_send_open() {
let (tx, rx) = sync_channel::<i32>(1);
assert_eq!(tx.try_send(10), Ok(()));
assert!(rx.recv().unwrap() == 10);
}
#[test]
fn oneshot_single_thread_try_send_closed() {
let (tx, rx) = sync_channel::<i32>(0);
drop(rx);
assert_eq!(tx.try_send(10), Err(TrySendError::Disconnected(10)));
}
#[test]
fn oneshot_single_thread_try_send_closed2() {
let (tx, _rx) = sync_channel::<i32>(0);
assert_eq!(tx.try_send(10), Err(TrySendError::Full(10)));
}
#[test]
fn oneshot_single_thread_try_recv_open() {
let (tx, rx) = sync_channel::<i32>(1);
tx.send(10).unwrap();
assert!(rx.recv() == Ok(10));
}
#[test]
fn oneshot_single_thread_try_recv_closed() {
let (tx, rx) = sync_channel::<i32>(0);
drop(tx);
assert!(rx.recv().is_err());
}
#[test]
fn oneshot_single_thread_try_recv_closed_with_data() {
let (tx, rx) = sync_channel::<i32>(1);
tx.send(10).unwrap();
drop(tx);
assert_eq!(rx.try_recv(), Ok(10));
assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected));
}
#[test]
fn oneshot_single_thread_peek_data() {
let (tx, rx) = sync_channel::<i32>(1);
assert_eq!(rx.try_recv(), Err(TryRecvError::Empty));
tx.send(10).unwrap();
assert_eq!(rx.try_recv(), Ok(10));
}
#[test]
fn oneshot_single_thread_peek_close() {
let (tx, rx) = sync_channel::<i32>(0);
drop(tx);
assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected));
assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected));
}
#[test]
fn oneshot_single_thread_peek_open() {
let (_tx, rx) = sync_channel::<i32>(0);
assert_eq!(rx.try_recv(), Err(TryRecvError::Empty));
}
#[test]
fn oneshot_multi_task_recv_then_send() {
let (tx, rx) = sync_channel::<Box<i32>>(0);
let _t = thread::spawn(move || {
assert!(*rx.recv().unwrap() == 10);
});
tx.send(Box::new(10)).unwrap();
}
#[test]
fn oneshot_multi_task_recv_then_close() {
let (tx, rx) = sync_channel::<Box<i32>>(0);
let _t = thread::spawn(move || {
drop(tx);
});
let res = thread::spawn(move || {
assert!(*rx.recv().unwrap() == 10);
})
.join();
assert!(res.is_err());
}
#[test]
fn oneshot_multi_thread_close_stress() {
for _ in 0..stress_factor() {
let (tx, rx) = sync_channel::<i32>(0);
let _t = thread::spawn(move || {
drop(rx);
});
drop(tx);
}
}
#[test]
fn oneshot_multi_thread_send_close_stress() {
for _ in 0..stress_factor() {
let (tx, rx) = sync_channel::<i32>(0);
let _t = thread::spawn(move || {
drop(rx);
});
let _ = thread::spawn(move || {
tx.send(1).unwrap();
})
.join();
}
}
#[test]
fn oneshot_multi_thread_recv_close_stress() {
for _ in 0..stress_factor() {
let (tx, rx) = sync_channel::<i32>(0);
let _t = thread::spawn(move || {
let res = thread::spawn(move || {
rx.recv().unwrap();
})
.join();
assert!(res.is_err());
});
let _t = thread::spawn(move || {
thread::spawn(move || {
drop(tx);
});
});
}
}
#[test]
fn oneshot_multi_thread_send_recv_stress() {
for _ in 0..stress_factor() {
let (tx, rx) = sync_channel::<Box<i32>>(0);
let _t = thread::spawn(move || {
tx.send(Box::new(10)).unwrap();
});
assert!(*rx.recv().unwrap() == 10);
}
}
#[test]
fn stream_send_recv_stress() {
for _ in 0..stress_factor() {
let (tx, rx) = sync_channel::<Box<i32>>(0);
send(tx, 0);
recv(rx, 0);
fn send(tx: SyncSender<Box<i32>>, i: i32) {
if i == 10 {
return;
}
thread::spawn(move || {
tx.send(Box::new(i)).unwrap();
send(tx, i + 1);
});
}
fn recv(rx: Receiver<Box<i32>>, i: i32) {
if i == 10 {
return;
}
thread::spawn(move || {
assert!(*rx.recv().unwrap() == i);
recv(rx, i + 1);
});
}
}
}
#[test]
fn recv_a_lot() {
let count = if cfg!(miri) { 1000 } else { 10000 };
// Regression test that we don't run out of stack in scheduler context
let (tx, rx) = sync_channel(count);
for _ in 0..count {
tx.send(()).unwrap();
}
for _ in 0..count {
rx.recv().unwrap();
}
}
#[test]
fn shared_chan_stress() {
let (tx, rx) = sync_channel(0);
let total = stress_factor() + 100;
for _ in 0..total {
let tx = tx.clone();
thread::spawn(move || {
tx.send(()).unwrap();
});
}
for _ in 0..total {
rx.recv().unwrap();
}
}
#[test]
fn test_nested_recv_iter() {
let (tx, rx) = sync_channel::<i32>(0);
let (total_tx, total_rx) = sync_channel::<i32>(0);
let _t = thread::spawn(move || {
let mut acc = 0;
for x in rx.iter() {
acc += x;
}
total_tx.send(acc).unwrap();
});
tx.send(3).unwrap();
tx.send(1).unwrap();
tx.send(2).unwrap();
drop(tx);
assert_eq!(total_rx.recv().unwrap(), 6);
}
#[test]
fn test_recv_iter_break() {
let (tx, rx) = sync_channel::<i32>(0);
let (count_tx, count_rx) = sync_channel(0);
let _t = thread::spawn(move || {
let mut count = 0;
for x in rx.iter() {
if count >= 3 {
break;
} else {
count += x;
}
}
count_tx.send(count).unwrap();
});
tx.send(2).unwrap();
tx.send(2).unwrap();
tx.send(2).unwrap();
let _ = tx.try_send(2);
drop(tx);
assert_eq!(count_rx.recv().unwrap(), 4);
}
#[test]
fn try_recv_states() {
let (tx1, rx1) = sync_channel::<i32>(1);
let (tx2, rx2) = sync_channel::<()>(1);
let (tx3, rx3) = sync_channel::<()>(1);
let _t = thread::spawn(move || {
rx2.recv().unwrap();
tx1.send(1).unwrap();
tx3.send(()).unwrap();
rx2.recv().unwrap();
drop(tx1);
tx3.send(()).unwrap();
});
assert_eq!(rx1.try_recv(), Err(TryRecvError::Empty));
tx2.send(()).unwrap();
rx3.recv().unwrap();
assert_eq!(rx1.try_recv(), Ok(1));
assert_eq!(rx1.try_recv(), Err(TryRecvError::Empty));
tx2.send(()).unwrap();
rx3.recv().unwrap();
assert_eq!(rx1.try_recv(), Err(TryRecvError::Disconnected));
}
// This bug used to end up in a livelock inside of the Receiver destructor
// because the internal state of the Shared packet was corrupted
#[test]
fn destroy_upgraded_shared_port_when_sender_still_active() {
let (tx, rx) = sync_channel::<()>(0);
let (tx2, rx2) = sync_channel::<()>(0);
let _t = thread::spawn(move || {
rx.recv().unwrap(); // wait on a oneshot
drop(rx); // destroy a shared
tx2.send(()).unwrap();
});
// make sure the other thread has gone to sleep
for _ in 0..5000 {
thread::yield_now();
}
// upgrade to a shared chan and send a message
let t = tx.clone();
drop(tx);
t.send(()).unwrap();
// wait for the child thread to exit before we exit
rx2.recv().unwrap();
}
#[test]
fn send1() {
let (tx, rx) = sync_channel::<i32>(0);
let _t = thread::spawn(move || {
rx.recv().unwrap();
});
assert_eq!(tx.send(1), Ok(()));
}
#[test]
fn send2() {
let (tx, rx) = sync_channel::<i32>(0);
let _t = thread::spawn(move || {
drop(rx);
});
assert!(tx.send(1).is_err());
}
#[test]
fn send3() {
let (tx, rx) = sync_channel::<i32>(1);
assert_eq!(tx.send(1), Ok(()));
let _t = thread::spawn(move || {
drop(rx);
});
assert!(tx.send(1).is_err());
}
#[test]
fn send4() {
let (tx, rx) = sync_channel::<i32>(0);
let tx2 = tx.clone();
let (done, donerx) = channel();
let done2 = done.clone();
let _t = thread::spawn(move || {
assert!(tx.send(1).is_err());
done.send(()).unwrap();
});
let _t = thread::spawn(move || {
assert!(tx2.send(2).is_err());
done2.send(()).unwrap();
});
drop(rx);
donerx.recv().unwrap();
donerx.recv().unwrap();
}
#[test]
fn try_send1() {
let (tx, _rx) = sync_channel::<i32>(0);
assert_eq!(tx.try_send(1), Err(TrySendError::Full(1)));
}
#[test]
fn try_send2() {
let (tx, _rx) = sync_channel::<i32>(1);
assert_eq!(tx.try_send(1), Ok(()));
assert_eq!(tx.try_send(1), Err(TrySendError::Full(1)));
}
#[test]
fn try_send3() {
let (tx, rx) = sync_channel::<i32>(1);
assert_eq!(tx.try_send(1), Ok(()));
drop(rx);
assert_eq!(tx.try_send(1), Err(TrySendError::Disconnected(1)));
}
#[test]
fn issue_15761() {
fn repro() {
let (tx1, rx1) = sync_channel::<()>(3);
let (tx2, rx2) = sync_channel::<()>(3);
let _t = thread::spawn(move || {
rx1.recv().unwrap();
tx2.try_send(()).unwrap();
});
tx1.try_send(()).unwrap();
rx2.recv().unwrap();
}
for _ in 0..100 {
repro()
}
}
#[test]
fn drop_unreceived() {
let (tx, rx) = sync_channel::<Rc<()>>(1);
let msg = Rc::new(());
let weak = Rc::downgrade(&msg);
assert!(tx.send(msg).is_ok());
drop(rx);
// Messages should be dropped immediately when the last receiver is destroyed.
assert!(weak.upgrade().is_none());
drop(tx);
}

721
library/std/src/sync/mpsc/tests.rs
View file

@ -1,721 +0,0 @@
use super::*;
use crate::{env, thread};
pub fn stress_factor() -> usize {
match env::var("RUST_TEST_STRESS") {
Ok(val) => val.parse().unwrap(),
Err(..) => 1,
}
}
#[test]
fn smoke() {
let (tx, rx) = channel::<i32>();
tx.send(1).unwrap();
assert_eq!(rx.recv().unwrap(), 1);
}
#[test]
fn drop_full() {
let (tx, _rx) = channel::<Box<isize>>();
tx.send(Box::new(1)).unwrap();
}
#[test]
fn drop_full_shared() {
let (tx, _rx) = channel::<Box<isize>>();
drop(tx.clone());
drop(tx.clone());
tx.send(Box::new(1)).unwrap();
}
#[test]
fn smoke_shared() {
let (tx, rx) = channel::<i32>();
tx.send(1).unwrap();
assert_eq!(rx.recv().unwrap(), 1);
let tx = tx.clone();
tx.send(1).unwrap();
assert_eq!(rx.recv().unwrap(), 1);
}
#[test]
fn smoke_threads() {
let (tx, rx) = channel::<i32>();
let _t = thread::spawn(move || {
tx.send(1).unwrap();
});
assert_eq!(rx.recv().unwrap(), 1);
}
#[test]
fn smoke_port_gone() {
let (tx, rx) = channel::<i32>();
drop(rx);
assert!(tx.send(1).is_err());
}
#[test]
fn smoke_shared_port_gone() {
let (tx, rx) = channel::<i32>();
drop(rx);
assert!(tx.send(1).is_err())
}
#[test]
fn smoke_shared_port_gone2() {
let (tx, rx) = channel::<i32>();
drop(rx);
let tx2 = tx.clone();
drop(tx);
assert!(tx2.send(1).is_err());
}
#[test]
fn port_gone_concurrent() {
let (tx, rx) = channel::<i32>();
let _t = thread::spawn(move || {
rx.recv().unwrap();
});
while tx.send(1).is_ok() {}
}
#[test]
fn port_gone_concurrent_shared() {
let (tx, rx) = channel::<i32>();
let tx2 = tx.clone();
let _t = thread::spawn(move || {
rx.recv().unwrap();
});
while tx.send(1).is_ok() && tx2.send(1).is_ok() {}
}
#[test]
fn smoke_chan_gone() {
let (tx, rx) = channel::<i32>();
drop(tx);
assert!(rx.recv().is_err());
}
#[test]
fn smoke_chan_gone_shared() {
let (tx, rx) = channel::<()>();
let tx2 = tx.clone();
drop(tx);
drop(tx2);
assert!(rx.recv().is_err());
}
#[test]
fn chan_gone_concurrent() {
let (tx, rx) = channel::<i32>();
let _t = thread::spawn(move || {
tx.send(1).unwrap();
tx.send(1).unwrap();
});
while rx.recv().is_ok() {}
}
#[test]
fn stress() {
let count = if cfg!(miri) { 100 } else { 10000 };
let (tx, rx) = channel::<i32>();
let t = thread::spawn(move || {
for _ in 0..count {
tx.send(1).unwrap();
}
});
for _ in 0..count {
assert_eq!(rx.recv().unwrap(), 1);
}
t.join().ok().expect("thread panicked");
}
#[test]
fn stress_shared() {
const AMT: u32 = if cfg!(miri) { 100 } else { 10000 };
const NTHREADS: u32 = 8;
let (tx, rx) = channel::<i32>();
let t = thread::spawn(move || {
for _ in 0..AMT * NTHREADS {
assert_eq!(rx.recv().unwrap(), 1);
}
match rx.try_recv() {
Ok(..) => panic!(),
_ => {}
}
});
for _ in 0..NTHREADS {
let tx = tx.clone();
thread::spawn(move || {
for _ in 0..AMT {
tx.send(1).unwrap();
}
});
}
drop(tx);
t.join().ok().expect("thread panicked");
}
#[test]
fn send_from_outside_runtime() {
let (tx1, rx1) = channel::<()>();
let (tx2, rx2) = channel::<i32>();
let t1 = thread::spawn(move || {
tx1.send(()).unwrap();
for _ in 0..40 {
assert_eq!(rx2.recv().unwrap(), 1);
}
});
rx1.recv().unwrap();
let t2 = thread::spawn(move || {
for _ in 0..40 {
tx2.send(1).unwrap();
}
});
t1.join().ok().expect("thread panicked");
t2.join().ok().expect("thread panicked");
}
#[test]
fn recv_from_outside_runtime() {
let (tx, rx) = channel::<i32>();
let t = thread::spawn(move || {
for _ in 0..40 {
assert_eq!(rx.recv().unwrap(), 1);
}
});
for _ in 0..40 {
tx.send(1).unwrap();
}
t.join().ok().expect("thread panicked");
}
#[test]
fn no_runtime() {
let (tx1, rx1) = channel::<i32>();
let (tx2, rx2) = channel::<i32>();
let t1 = thread::spawn(move || {
assert_eq!(rx1.recv().unwrap(), 1);
tx2.send(2).unwrap();
});
let t2 = thread::spawn(move || {
tx1.send(1).unwrap();
assert_eq!(rx2.recv().unwrap(), 2);
});
t1.join().ok().expect("thread panicked");
t2.join().ok().expect("thread panicked");
}
#[test]
fn oneshot_single_thread_close_port_first() {
// Simple test of closing without sending
let (_tx, rx) = channel::<i32>();
drop(rx);
}
#[test]
fn oneshot_single_thread_close_chan_first() {
// Simple test of closing without sending
let (tx, _rx) = channel::<i32>();
drop(tx);
}
#[test]
fn oneshot_single_thread_send_port_close() {
// Testing that the sender cleans up the payload if receiver is closed
let (tx, rx) = channel::<Box<i32>>();
drop(rx);
assert!(tx.send(Box::new(0)).is_err());
}
#[test]
fn oneshot_single_thread_recv_chan_close() {
// Receiving on a closed chan will panic
let res = thread::spawn(move || {
let (tx, rx) = channel::<i32>();
drop(tx);
rx.recv().unwrap();
})
.join();
// What is our res?
assert!(res.is_err());
}
#[test]
fn oneshot_single_thread_send_then_recv() {
let (tx, rx) = channel::<Box<i32>>();
tx.send(Box::new(10)).unwrap();
assert!(*rx.recv().unwrap() == 10);
}
#[test]
fn oneshot_single_thread_try_send_open() {
let (tx, rx) = channel::<i32>();
assert!(tx.send(10).is_ok());
assert!(rx.recv().unwrap() == 10);
}
#[test]
fn oneshot_single_thread_try_send_closed() {
let (tx, rx) = channel::<i32>();
drop(rx);
assert!(tx.send(10).is_err());
}
#[test]
fn oneshot_single_thread_try_recv_open() {
let (tx, rx) = channel::<i32>();
tx.send(10).unwrap();
assert!(rx.recv() == Ok(10));
}
#[test]
fn oneshot_single_thread_try_recv_closed() {
let (tx, rx) = channel::<i32>();
drop(tx);
assert!(rx.recv().is_err());
}
#[test]
fn oneshot_single_thread_peek_data() {
let (tx, rx) = channel::<i32>();
assert_eq!(rx.try_recv(), Err(TryRecvError::Empty));
tx.send(10).unwrap();
assert_eq!(rx.try_recv(), Ok(10));
}
#[test]
fn oneshot_single_thread_peek_close() {
let (tx, rx) = channel::<i32>();
drop(tx);
assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected));
assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected));
}
#[test]
fn oneshot_single_thread_peek_open() {
let (_tx, rx) = channel::<i32>();
assert_eq!(rx.try_recv(), Err(TryRecvError::Empty));
}
#[test]
fn oneshot_multi_task_recv_then_send() {
let (tx, rx) = channel::<Box<i32>>();
let _t = thread::spawn(move || {
assert!(*rx.recv().unwrap() == 10);
});
tx.send(Box::new(10)).unwrap();
}
#[test]
fn oneshot_multi_task_recv_then_close() {
let (tx, rx) = channel::<Box<i32>>();
let _t = thread::spawn(move || {
drop(tx);
});
let res = thread::spawn(move || {
assert!(*rx.recv().unwrap() == 10);
})
.join();
assert!(res.is_err());
}
#[test]
fn oneshot_multi_thread_close_stress() {
for _ in 0..stress_factor() {
let (tx, rx) = channel::<i32>();
let _t = thread::spawn(move || {
drop(rx);
});
drop(tx);
}
}
#[test]
fn oneshot_multi_thread_send_close_stress() {
for _ in 0..stress_factor() {
let (tx, rx) = channel::<i32>();
let _t = thread::spawn(move || {
drop(rx);
});
let _ = thread::spawn(move || {
tx.send(1).unwrap();
})
.join();
}
}
#[test]
fn oneshot_multi_thread_recv_close_stress() {
for _ in 0..stress_factor() {
let (tx, rx) = channel::<i32>();
thread::spawn(move || {
let res = thread::spawn(move || {
rx.recv().unwrap();
})
.join();
assert!(res.is_err());
});
let _t = thread::spawn(move || {
thread::spawn(move || {
drop(tx);
});
});
}
}
#[test]
fn oneshot_multi_thread_send_recv_stress() {
for _ in 0..stress_factor() {
let (tx, rx) = channel::<Box<isize>>();
let _t = thread::spawn(move || {
tx.send(Box::new(10)).unwrap();
});
assert!(*rx.recv().unwrap() == 10);
}
}
#[test]
fn stream_send_recv_stress() {
for _ in 0..stress_factor() {
let (tx, rx) = channel();
send(tx, 0);
recv(rx, 0);
fn send(tx: Sender<Box<i32>>, i: i32) {
if i == 10 {
return;
}
thread::spawn(move || {
tx.send(Box::new(i)).unwrap();
send(tx, i + 1);
});
}
fn recv(rx: Receiver<Box<i32>>, i: i32) {
if i == 10 {
return;
}
thread::spawn(move || {
assert!(*rx.recv().unwrap() == i);
recv(rx, i + 1);
});
}
}
}
#[test]
fn oneshot_single_thread_recv_timeout() {
let (tx, rx) = channel();
tx.send(()).unwrap();
assert_eq!(rx.recv_timeout(Duration::from_millis(1)), Ok(()));
assert_eq!(rx.recv_timeout(Duration::from_millis(1)), Err(RecvTimeoutError::Timeout));
tx.send(()).unwrap();
assert_eq!(rx.recv_timeout(Duration::from_millis(1)), Ok(()));
}
#[test]
fn stress_recv_timeout_two_threads() {
let (tx, rx) = channel();
let stress = stress_factor() + 100;
let timeout = Duration::from_millis(100);
thread::spawn(move || {
for i in 0..stress {
if i % 2 == 0 {
thread::sleep(timeout * 2);
}
tx.send(1usize).unwrap();
}
});
let mut recv_count = 0;
loop {
match rx.recv_timeout(timeout) {
Ok(n) => {
assert_eq!(n, 1usize);
recv_count += 1;
}
Err(RecvTimeoutError::Timeout) => continue,
Err(RecvTimeoutError::Disconnected) => break,
}
}
assert_eq!(recv_count, stress);
}
#[test]
fn recv_timeout_upgrade() {
let (tx, rx) = channel::<()>();
let timeout = Duration::from_millis(1);
let _tx_clone = tx.clone();
let start = Instant::now();
assert_eq!(rx.recv_timeout(timeout), Err(RecvTimeoutError::Timeout));
assert!(Instant::now() >= start + timeout);
}
#[test]
fn stress_recv_timeout_shared() {
let (tx, rx) = channel();
let stress = stress_factor() + 100;
for i in 0..stress {
let tx = tx.clone();
thread::spawn(move || {
thread::sleep(Duration::from_millis(i as u64 * 10));
tx.send(1usize).unwrap();
});
}
drop(tx);
let mut recv_count = 0;
loop {
match rx.recv_timeout(Duration::from_millis(10)) {
Ok(n) => {
assert_eq!(n, 1usize);
recv_count += 1;
}
Err(RecvTimeoutError::Timeout) => continue,
Err(RecvTimeoutError::Disconnected) => break,
}
}
assert_eq!(recv_count, stress);
}
#[test]
fn very_long_recv_timeout_wont_panic() {
let (tx, rx) = channel::<()>();
let join_handle = thread::spawn(move || rx.recv_timeout(Duration::from_secs(u64::MAX)));
thread::sleep(Duration::from_secs(1));
assert!(tx.send(()).is_ok());
assert_eq!(join_handle.join().unwrap(), Ok(()));
}
#[test]
fn recv_a_lot() {
let count = if cfg!(miri) { 1000 } else { 10000 };
// Regression test that we don't run out of stack in scheduler context
let (tx, rx) = channel();
for _ in 0..count {
tx.send(()).unwrap();
}
for _ in 0..count {
rx.recv().unwrap();
}
}
#[test]
fn shared_recv_timeout() {
let (tx, rx) = channel();
let total = 5;
for _ in 0..total {
let tx = tx.clone();
thread::spawn(move || {
tx.send(()).unwrap();
});
}
for _ in 0..total {
rx.recv().unwrap();
}
assert_eq!(rx.recv_timeout(Duration::from_millis(1)), Err(RecvTimeoutError::Timeout));
tx.send(()).unwrap();
assert_eq!(rx.recv_timeout(Duration::from_millis(1)), Ok(()));
}
#[test]
fn shared_chan_stress() {
let (tx, rx) = channel();
let total = stress_factor() + 100;
for _ in 0..total {
let tx = tx.clone();
thread::spawn(move || {
tx.send(()).unwrap();
});
}
for _ in 0..total {
rx.recv().unwrap();
}
}
#[test]
fn test_nested_recv_iter() {
let (tx, rx) = channel::<i32>();
let (total_tx, total_rx) = channel::<i32>();
let _t = thread::spawn(move || {
let mut acc = 0;
for x in rx.iter() {
acc += x;
}
total_tx.send(acc).unwrap();
});
tx.send(3).unwrap();
tx.send(1).unwrap();
tx.send(2).unwrap();
drop(tx);
assert_eq!(total_rx.recv().unwrap(), 6);
}
#[test]
fn test_recv_iter_break() {
let (tx, rx) = channel::<i32>();
let (count_tx, count_rx) = channel();
let _t = thread::spawn(move || {
let mut count = 0;
for x in rx.iter() {
if count >= 3 {
break;
} else {
count += x;
}
}
count_tx.send(count).unwrap();
});
tx.send(2).unwrap();
tx.send(2).unwrap();
tx.send(2).unwrap();
let _ = tx.send(2);
drop(tx);
assert_eq!(count_rx.recv().unwrap(), 4);
}
#[test]
fn test_recv_try_iter() {
let (request_tx, request_rx) = channel();
let (response_tx, response_rx) = channel();
// Request `x`s until we have `6`.
let t = thread::spawn(move || {
let mut count = 0;
loop {
for x in response_rx.try_iter() {
count += x;
if count == 6 {
return count;
}
}
request_tx.send(()).unwrap();
}
});
for _ in request_rx.iter() {
if response_tx.send(2).is_err() {
break;
}
}
assert_eq!(t.join().unwrap(), 6);
}
#[test]
fn test_recv_into_iter_owned() {
let mut iter = {
let (tx, rx) = channel::<i32>();
tx.send(1).unwrap();
tx.send(2).unwrap();
rx.into_iter()
};
assert_eq!(iter.next().unwrap(), 1);
assert_eq!(iter.next().unwrap(), 2);
assert_eq!(iter.next().is_none(), true);
}
#[test]
fn test_recv_into_iter_borrowed() {
let (tx, rx) = channel::<i32>();
tx.send(1).unwrap();
tx.send(2).unwrap();
drop(tx);
let mut iter = (&rx).into_iter();
assert_eq!(iter.next().unwrap(), 1);
assert_eq!(iter.next().unwrap(), 2);
assert_eq!(iter.next().is_none(), true);
}
#[test]
fn try_recv_states() {
let (tx1, rx1) = channel::<i32>();
let (tx2, rx2) = channel::<()>();
let (tx3, rx3) = channel::<()>();
let _t = thread::spawn(move || {
rx2.recv().unwrap();
tx1.send(1).unwrap();
tx3.send(()).unwrap();
rx2.recv().unwrap();
drop(tx1);
tx3.send(()).unwrap();
});
assert_eq!(rx1.try_recv(), Err(TryRecvError::Empty));
tx2.send(()).unwrap();
rx3.recv().unwrap();
assert_eq!(rx1.try_recv(), Ok(1));
assert_eq!(rx1.try_recv(), Err(TryRecvError::Empty));
tx2.send(()).unwrap();
rx3.recv().unwrap();
assert_eq!(rx1.try_recv(), Err(TryRecvError::Disconnected));
}
// This bug used to end up in a livelock inside of the Receiver destructor
// because the internal state of the Shared packet was corrupted
#[test]
fn destroy_upgraded_shared_port_when_sender_still_active() {
let (tx, rx) = channel();
let (tx2, rx2) = channel();
let _t = thread::spawn(move || {
rx.recv().unwrap(); // wait on a oneshot
drop(rx); // destroy a shared
tx2.send(()).unwrap();
});
// make sure the other thread has gone to sleep
for _ in 0..5000 {
thread::yield_now();
}
// upgrade to a shared chan and send a message
let t = tx.clone();
drop(tx);
t.send(()).unwrap();
// wait for the child thread to exit before we exit
rx2.recv().unwrap();
}
#[test]
fn issue_32114() {
let (tx, _) = channel();
let _ = tx.send(123);
assert_eq!(tx.send(123), Err(SendError(123)));
}
#[test]
fn issue_39364() {
let (tx, rx) = channel::<()>();
let t = thread::spawn(move || {
thread::sleep(Duration::from_millis(300));
let _ = tx.clone();
// Don't drop; hand back to caller.
tx
});
let _ = rx.recv_timeout(Duration::from_millis(500));
let _tx = t.join().unwrap(); // delay dropping until end of test
let _ = rx.recv_timeout(Duration::from_millis(500));
}

3
library/std/src/sync/once_lock.rs
View file

@ -676,6 +676,3 @@ unsafe impl<#[may_dangle] T> Drop for OnceLock<T> {
}
}
}
#[cfg(test)]
mod tests;

200
library/std/src/sync/once_lock/tests.rs
View file

@ -1,200 +0,0 @@
use crate::sync::OnceLock;
use crate::sync::atomic::AtomicUsize;
use crate::sync::atomic::Ordering::SeqCst;
use crate::sync::mpsc::channel;
use crate::{panic, thread};
fn spawn_and_wait<R: Send + 'static>(f: impl FnOnce() -> R + Send + 'static) -> R {
thread::spawn(f).join().unwrap()
}
#[test]
#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads
fn sync_once_cell() {
static ONCE_CELL: OnceLock<i32> = OnceLock::new();
assert!(ONCE_CELL.get().is_none());
spawn_and_wait(|| {
ONCE_CELL.get_or_init(|| 92);
assert_eq!(ONCE_CELL.get(), Some(&92));
});
ONCE_CELL.get_or_init(|| panic!("Kaboom!"));
assert_eq!(ONCE_CELL.get(), Some(&92));
}
#[test]
fn sync_once_cell_get_mut() {
let mut c = OnceLock::new();
assert!(c.get_mut().is_none());
c.set(90).unwrap();
*c.get_mut().unwrap() += 2;
assert_eq!(c.get_mut(), Some(&mut 92));
}
#[test]
fn sync_once_cell_get_unchecked() {
let c = OnceLock::new();
c.set(92).unwrap();
unsafe {
assert_eq!(c.get_unchecked(), &92);
}
}
#[test]
#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads
fn sync_once_cell_drop() {
static DROP_CNT: AtomicUsize = AtomicUsize::new(0);
struct Dropper;
impl Drop for Dropper {
fn drop(&mut self) {
DROP_CNT.fetch_add(1, SeqCst);
}
}
let x = OnceLock::new();
spawn_and_wait(move || {
x.get_or_init(|| Dropper);
assert_eq!(DROP_CNT.load(SeqCst), 0);
drop(x);
});
assert_eq!(DROP_CNT.load(SeqCst), 1);
}
#[test]
fn sync_once_cell_drop_empty() {
let x = OnceLock::<String>::new();
drop(x);
}
#[test]
fn clone() {
let s = OnceLock::new();
let c = s.clone();
assert!(c.get().is_none());
s.set("hello".to_string()).unwrap();
let c = s.clone();
assert_eq!(c.get().map(String::as_str), Some("hello"));
}
#[test]
#[cfg_attr(not(panic = "unwind"), ignore = "test requires unwinding support")]
fn get_or_try_init() {
let cell: OnceLock<String> = OnceLock::new();
assert!(cell.get().is_none());
let res = panic::catch_unwind(|| cell.get_or_try_init(|| -> Result<_, ()> { panic!() }));
assert!(res.is_err());
assert!(!cell.is_initialized());
assert!(cell.get().is_none());
assert_eq!(cell.get_or_try_init(|| Err(())), Err(()));
assert_eq!(cell.get_or_try_init(|| Ok::<_, ()>("hello".to_string())), Ok(&"hello".to_string()));
assert_eq!(cell.get(), Some(&"hello".to_string()));
}
#[test]
fn from_impl() {
assert_eq!(OnceLock::from("value").get(), Some(&"value"));
assert_ne!(OnceLock::from("foo").get(), Some(&"bar"));
}
#[test]
fn partialeq_impl() {
assert!(OnceLock::from("value") == OnceLock::from("value"));
assert!(OnceLock::from("foo") != OnceLock::from("bar"));
assert!(OnceLock::<String>::new() == OnceLock::new());
assert!(OnceLock::<String>::new() != OnceLock::from("value".to_owned()));
}
#[test]
fn into_inner() {
let cell: OnceLock<String> = OnceLock::new();
assert_eq!(cell.into_inner(), None);
let cell = OnceLock::new();
cell.set("hello".to_string()).unwrap();
assert_eq!(cell.into_inner(), Some("hello".to_string()));
}
#[test]
fn is_sync_send() {
fn assert_traits<T: Send + Sync>() {}
assert_traits::<OnceLock<String>>();
}
#[test]
fn eval_once_macro() {
macro_rules! eval_once {
(|| -> $ty:ty {
$($body:tt)*
}) => {{
static ONCE_CELL: OnceLock<$ty> = OnceLock::new();
fn init() -> $ty {
$($body)*
}
ONCE_CELL.get_or_init(init)
}};
}
let fib: &'static Vec<i32> = eval_once! {
|| -> Vec<i32> {
let mut res = vec![1, 1];
for i in 0..10 {
let next = res[i] + res[i + 1];
res.push(next);
}
res
}
};
assert_eq!(fib[5], 8)
}
#[test]
#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads
fn sync_once_cell_does_not_leak_partially_constructed_boxes() {
static ONCE_CELL: OnceLock<String> = OnceLock::new();
let n_readers = 10;
let n_writers = 3;
const MSG: &str = "Hello, World";
let (tx, rx) = channel();
for _ in 0..n_readers {
let tx = tx.clone();
thread::spawn(move || {
loop {
if let Some(msg) = ONCE_CELL.get() {
tx.send(msg).unwrap();
break;
}
#[cfg(target_env = "sgx")]
crate::thread::yield_now();
}
});
}
for _ in 0..n_writers {
thread::spawn(move || {
let _ = ONCE_CELL.set(MSG.to_owned());
});
}
for _ in 0..n_readers {
let msg = rx.recv().unwrap();
assert_eq!(msg, MSG);
}
}
#[test]
fn dropck() {
let cell = OnceLock::new();
{
let s = String::new();
cell.set(&s).unwrap();
}
}

3
library/std/src/sync/poison/condvar.rs
View file

@ -1,6 +1,3 @@
#[cfg(test)]
mod tests;
use crate::fmt;
use crate::sync::poison::{self, LockResult, MutexGuard, PoisonError, mutex};
use crate::sys::sync as sys;

190
library/std/src/sync/poison/condvar/tests.rs
View file

@ -1,190 +0,0 @@
use crate::sync::atomic::{AtomicBool, Ordering};
use crate::sync::mpsc::channel;
use crate::sync::{Arc, Condvar, Mutex};
use crate::thread;
use crate::time::Duration;
#[test]
fn smoke() {
let c = Condvar::new();
c.notify_one();
c.notify_all();
}
#[test]
#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads
fn notify_one() {
let m = Arc::new(Mutex::new(()));
let m2 = m.clone();
let c = Arc::new(Condvar::new());
let c2 = c.clone();
let g = m.lock().unwrap();
let _t = thread::spawn(move || {
let _g = m2.lock().unwrap();
c2.notify_one();
});
let g = c.wait(g).unwrap();
drop(g);
}
#[test]
#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads
fn notify_all() {
const N: usize = 10;
let data = Arc::new((Mutex::new(0), Condvar::new()));
let (tx, rx) = channel();
for _ in 0..N {
let data = data.clone();
let tx = tx.clone();
thread::spawn(move || {
let &(ref lock, ref cond) = &*data;
let mut cnt = lock.lock().unwrap();
*cnt += 1;
if *cnt == N {
tx.send(()).unwrap();
}
while *cnt != 0 {
cnt = cond.wait(cnt).unwrap();
}
tx.send(()).unwrap();
});
}
drop(tx);
let &(ref lock, ref cond) = &*data;
rx.recv().unwrap();
let mut cnt = lock.lock().unwrap();
*cnt = 0;
cond.notify_all();
drop(cnt);
for _ in 0..N {
rx.recv().unwrap();
}
}
#[test]
#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads
fn wait_while() {
let pair = Arc::new((Mutex::new(false), Condvar::new()));
let pair2 = pair.clone();
// Inside of our lock, spawn a new thread, and then wait for it to start.
thread::spawn(move || {
let &(ref lock, ref cvar) = &*pair2;
let mut started = lock.lock().unwrap();
*started = true;
// We notify the condvar that the value has changed.
cvar.notify_one();
});
// Wait for the thread to start up.
let &(ref lock, ref cvar) = &*pair;
let guard = cvar.wait_while(lock.lock().unwrap(), |started| !*started);
assert!(*guard.unwrap());
}
#[test]
#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // condvar wait not supported
fn wait_timeout_wait() {
let m = Arc::new(Mutex::new(()));
let c = Arc::new(Condvar::new());
loop {
let g = m.lock().unwrap();
let (_g, no_timeout) = c.wait_timeout(g, Duration::from_millis(1)).unwrap();
// spurious wakeups mean this isn't necessarily true
// so execute test again, if not timeout
if !no_timeout.timed_out() {
continue;
}
break;
}
}
#[test]
#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // condvar wait not supported
fn wait_timeout_while_wait() {
let m = Arc::new(Mutex::new(()));
let c = Arc::new(Condvar::new());
let g = m.lock().unwrap();
let (_g, wait) = c.wait_timeout_while(g, Duration::from_millis(1), |_| true).unwrap();
// no spurious wakeups. ensure it timed-out
assert!(wait.timed_out());
}
#[test]
#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // condvar wait not supported
fn wait_timeout_while_instant_satisfy() {
let m = Arc::new(Mutex::new(()));
let c = Arc::new(Condvar::new());
let g = m.lock().unwrap();
let (_g, wait) = c.wait_timeout_while(g, Duration::from_millis(0), |_| false).unwrap();
// ensure it didn't time-out even if we were not given any time.
assert!(!wait.timed_out());
}
#[test]
#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads
fn wait_timeout_while_wake() {
let pair = Arc::new((Mutex::new(false), Condvar::new()));
let pair_copy = pair.clone();
let &(ref m, ref c) = &*pair;
let g = m.lock().unwrap();
let _t = thread::spawn(move || {
let &(ref lock, ref cvar) = &*pair_copy;
let mut started = lock.lock().unwrap();
thread::sleep(Duration::from_millis(1));
*started = true;
cvar.notify_one();
});
let (g2, wait) = c
.wait_timeout_while(g, Duration::from_millis(u64::MAX), |&mut notified| !notified)
.unwrap();
// ensure it didn't time-out even if we were not given any time.
assert!(!wait.timed_out());
assert!(*g2);
}
#[test]
#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads
fn wait_timeout_wake() {
let m = Arc::new(Mutex::new(()));
let c = Arc::new(Condvar::new());
loop {
let g = m.lock().unwrap();
let c2 = c.clone();
let m2 = m.clone();
let notified = Arc::new(AtomicBool::new(false));
let notified_copy = notified.clone();
let t = thread::spawn(move || {
let _g = m2.lock().unwrap();
thread::sleep(Duration::from_millis(1));
notified_copy.store(true, Ordering::Relaxed);
c2.notify_one();
});
let (g, timeout_res) = c.wait_timeout(g, Duration::from_millis(u64::MAX)).unwrap();
assert!(!timeout_res.timed_out());
// spurious wakeups mean this isn't necessarily true
// so execute test again, if not notified
if !notified.load(Ordering::Relaxed) {
t.join().unwrap();
continue;
}
drop(g);
t.join().unwrap();
break;
}
}

3
library/std/src/sync/poison/mutex.rs
View file

@ -1,6 +1,3 @@
#[cfg(all(test, not(any(target_os = "emscripten", target_os = "wasi"))))]
mod tests;
use crate::cell::UnsafeCell;
use crate::fmt;
use crate::marker::PhantomData;

442
library/std/src/sync/poison/mutex/tests.rs
View file

@ -1,442 +0,0 @@
use crate::fmt::Debug;
use crate::ops::FnMut;
use crate::panic::{self, AssertUnwindSafe};
use crate::sync::atomic::{AtomicUsize, Ordering};
use crate::sync::mpsc::channel;
use crate::sync::{Arc, Condvar, MappedMutexGuard, Mutex, MutexGuard, TryLockError};
use crate::{hint, mem, thread};
struct Packet<T>(Arc<(Mutex<T>, Condvar)>);
#[derive(Eq, PartialEq, Debug)]
struct NonCopy(i32);
#[derive(Eq, PartialEq, Debug)]
struct NonCopyNeedsDrop(i32);
impl Drop for NonCopyNeedsDrop {
fn drop(&mut self) {
hint::black_box(());
}
}
#[test]
fn test_needs_drop() {
assert!(!mem::needs_drop::<NonCopy>());
assert!(mem::needs_drop::<NonCopyNeedsDrop>());
}
#[derive(Clone, Eq, PartialEq, Debug)]
struct Cloneable(i32);
#[test]
fn smoke() {
let m = Mutex::new(());
drop(m.lock().unwrap());
drop(m.lock().unwrap());
}
#[test]
fn lots_and_lots() {
const J: u32 = 1000;
const K: u32 = 3;
let m = Arc::new(Mutex::new(0));
fn inc(m: &Mutex<u32>) {
for _ in 0..J {
*m.lock().unwrap() += 1;
}
}
let (tx, rx) = channel();
for _ in 0..K {
let tx2 = tx.clone();
let m2 = m.clone();
thread::spawn(move || {
inc(&m2);
tx2.send(()).unwrap();
});
let tx2 = tx.clone();
let m2 = m.clone();
thread::spawn(move || {
inc(&m2);
tx2.send(()).unwrap();
});
}
drop(tx);
for _ in 0..2 * K {
rx.recv().unwrap();
}
assert_eq!(*m.lock().unwrap(), J * K * 2);
}
#[test]
fn try_lock() {
let m = Mutex::new(());
*m.try_lock().unwrap() = ();
}
fn new_poisoned_mutex<T>(value: T) -> Mutex<T> {
let mutex = Mutex::new(value);
let catch_unwind_result = panic::catch_unwind(AssertUnwindSafe(|| {
let _guard = mutex.lock().unwrap();
panic!("test panic to poison mutex");
}));
assert!(catch_unwind_result.is_err());
assert!(mutex.is_poisoned());
mutex
}
#[test]
fn test_into_inner() {
let m = Mutex::new(NonCopy(10));
assert_eq!(m.into_inner().unwrap(), NonCopy(10));
}
#[test]
fn test_into_inner_drop() {
struct Foo(Arc<AtomicUsize>);
impl Drop for Foo {
fn drop(&mut self) {
self.0.fetch_add(1, Ordering::SeqCst);
}
}
let num_drops = Arc::new(AtomicUsize::new(0));
let m = Mutex::new(Foo(num_drops.clone()));
assert_eq!(num_drops.load(Ordering::SeqCst), 0);
{
let _inner = m.into_inner().unwrap();
assert_eq!(num_drops.load(Ordering::SeqCst), 0);
}
assert_eq!(num_drops.load(Ordering::SeqCst), 1);
}
#[test]
fn test_into_inner_poison() {
let m = new_poisoned_mutex(NonCopy(10));
match m.into_inner() {
Err(e) => assert_eq!(e.into_inner(), NonCopy(10)),
Ok(x) => panic!("into_inner of poisoned Mutex is Ok: {x:?}"),
}
}
#[test]
fn test_get_cloned() {
let m = Mutex::new(Cloneable(10));
assert_eq!(m.get_cloned().unwrap(), Cloneable(10));
}
#[test]
fn test_get_cloned_poison() {
let m = new_poisoned_mutex(Cloneable(10));
match m.get_cloned() {
Err(e) => assert_eq!(e.into_inner(), ()),
Ok(x) => panic!("get of poisoned Mutex is Ok: {x:?}"),
}
}
#[test]
fn test_get_mut() {
let mut m = Mutex::new(NonCopy(10));
*m.get_mut().unwrap() = NonCopy(20);
assert_eq!(m.into_inner().unwrap(), NonCopy(20));
}
#[test]
fn test_get_mut_poison() {
let mut m = new_poisoned_mutex(NonCopy(10));
match m.get_mut() {
Err(e) => assert_eq!(*e.into_inner(), NonCopy(10)),
Ok(x) => panic!("get_mut of poisoned Mutex is Ok: {x:?}"),
}
}
#[test]
fn test_set() {
fn inner<T>(mut init: impl FnMut() -> T, mut value: impl FnMut() -> T)
where
T: Debug + Eq,
{
let m = Mutex::new(init());
assert_eq!(*m.lock().unwrap(), init());
m.set(value()).unwrap();
assert_eq!(*m.lock().unwrap(), value());
}
inner(|| NonCopy(10), || NonCopy(20));
inner(|| NonCopyNeedsDrop(10), || NonCopyNeedsDrop(20));
}
#[test]
fn test_set_poison() {
fn inner<T>(mut init: impl FnMut() -> T, mut value: impl FnMut() -> T)
where
T: Debug + Eq,
{
let m = new_poisoned_mutex(init());
match m.set(value()) {
Err(e) => {
assert_eq!(e.into_inner(), value());
assert_eq!(m.into_inner().unwrap_err().into_inner(), init());
}
Ok(x) => panic!("set of poisoned Mutex is Ok: {x:?}"),
}
}
inner(|| NonCopy(10), || NonCopy(20));
inner(|| NonCopyNeedsDrop(10), || NonCopyNeedsDrop(20));
}
#[test]
fn test_replace() {
fn inner<T>(mut init: impl FnMut() -> T, mut value: impl FnMut() -> T)
where
T: Debug + Eq,
{
let m = Mutex::new(init());
assert_eq!(*m.lock().unwrap(), init());
assert_eq!(m.replace(value()).unwrap(), init());
assert_eq!(*m.lock().unwrap(), value());
}
inner(|| NonCopy(10), || NonCopy(20));
inner(|| NonCopyNeedsDrop(10), || NonCopyNeedsDrop(20));
}
#[test]
fn test_replace_poison() {
fn inner<T>(mut init: impl FnMut() -> T, mut value: impl FnMut() -> T)
where
T: Debug + Eq,
{
let m = new_poisoned_mutex(init());
match m.replace(value()) {
Err(e) => {
assert_eq!(e.into_inner(), value());
assert_eq!(m.into_inner().unwrap_err().into_inner(), init());
}
Ok(x) => panic!("replace of poisoned Mutex is Ok: {x:?}"),
}
}
inner(|| NonCopy(10), || NonCopy(20));
inner(|| NonCopyNeedsDrop(10), || NonCopyNeedsDrop(20));
}
#[test]
fn test_mutex_arc_condvar() {
let packet = Packet(Arc::new((Mutex::new(false), Condvar::new())));
let packet2 = Packet(packet.0.clone());
let (tx, rx) = channel();
let _t = thread::spawn(move || {
// wait until parent gets in
rx.recv().unwrap();
let &(ref lock, ref cvar) = &*packet2.0;
let mut lock = lock.lock().unwrap();
*lock = true;
cvar.notify_one();
});
let &(ref lock, ref cvar) = &*packet.0;
let mut lock = lock.lock().unwrap();
tx.send(()).unwrap();
assert!(!*lock);
while !*lock {
lock = cvar.wait(lock).unwrap();
}
}
#[test]
fn test_arc_condvar_poison() {
let packet = Packet(Arc::new((Mutex::new(1), Condvar::new())));
let packet2 = Packet(packet.0.clone());
let (tx, rx) = channel();
let _t = thread::spawn(move || -> () {
rx.recv().unwrap();
let &(ref lock, ref cvar) = &*packet2.0;
let _g = lock.lock().unwrap();
cvar.notify_one();
// Parent should fail when it wakes up.
panic!();
});
let &(ref lock, ref cvar) = &*packet.0;
let mut lock = lock.lock().unwrap();
tx.send(()).unwrap();
while *lock == 1 {
match cvar.wait(lock) {
Ok(l) => {
lock = l;
assert_eq!(*lock, 1);
}
Err(..) => break,
}
}
}
#[test]
fn test_mutex_arc_poison() {
let arc = Arc::new(Mutex::new(1));
assert!(!arc.is_poisoned());
let arc2 = arc.clone();
let _ = thread::spawn(move || {
let lock = arc2.lock().unwrap();
assert_eq!(*lock, 2); // deliberate assertion failure to poison the mutex
})
.join();
assert!(arc.lock().is_err());
assert!(arc.is_poisoned());
}
#[test]
fn test_mutex_arc_poison_mapped() {
let arc = Arc::new(Mutex::new(1));
assert!(!arc.is_poisoned());
let arc2 = arc.clone();
let _ = thread::spawn(move || {
let lock = arc2.lock().unwrap();
let lock = MutexGuard::map(lock, |val| val);
assert_eq!(*lock, 2); // deliberate assertion failure to poison the mutex
})
.join();
assert!(arc.lock().is_err());
assert!(arc.is_poisoned());
}
#[test]
fn test_mutex_arc_nested() {
// Tests nested mutexes and access
// to underlying data.
let arc = Arc::new(Mutex::new(1));
let arc2 = Arc::new(Mutex::new(arc));
let (tx, rx) = channel();
let _t = thread::spawn(move || {
let lock = arc2.lock().unwrap();
let lock2 = lock.lock().unwrap();
assert_eq!(*lock2, 1);
tx.send(()).unwrap();
});
rx.recv().unwrap();
}
#[test]
fn test_mutex_arc_access_in_unwind() {
let arc = Arc::new(Mutex::new(1));
let arc2 = arc.clone();
let _ = thread::spawn(move || -> () {
struct Unwinder {
i: Arc<Mutex<i32>>,
}
impl Drop for Unwinder {
fn drop(&mut self) {
*self.i.lock().unwrap() += 1;
}
}
let _u = Unwinder { i: arc2 };
panic!();
})
.join();
let lock = arc.lock().unwrap();
assert_eq!(*lock, 2);
}
#[test]
fn test_mutex_unsized() {
let mutex: &Mutex<[i32]> = &Mutex::new([1, 2, 3]);
{
let b = &mut *mutex.lock().unwrap();
b[0] = 4;
b[2] = 5;
}
let comp: &[i32] = &[4, 2, 5];
assert_eq!(&*mutex.lock().unwrap(), comp);
}
#[test]
fn test_mapping_mapped_guard() {
let arr = [0; 4];
let mut lock = Mutex::new(arr);
let guard = lock.lock().unwrap();
let guard = MutexGuard::map(guard, |arr| &mut arr[..2]);
let mut guard = MappedMutexGuard::map(guard, |slice| &mut slice[1..]);
assert_eq!(guard.len(), 1);
guard[0] = 42;
drop(guard);
assert_eq!(*lock.get_mut().unwrap(), [0, 42, 0, 0]);
}
#[test]
fn panic_while_mapping_unlocked_poison() {
let lock = Mutex::new(());
let _ = panic::catch_unwind(|| {
let guard = lock.lock().unwrap();
let _guard = MutexGuard::map::<(), _>(guard, |_| panic!());
});
match lock.try_lock() {
Ok(_) => panic!("panicking in a MutexGuard::map closure should poison the Mutex"),
Err(TryLockError::WouldBlock) => {
panic!("panicking in a MutexGuard::map closure should unlock the mutex")
}
Err(TryLockError::Poisoned(_)) => {}
}
let _ = panic::catch_unwind(|| {
let guard = lock.lock().unwrap();
let _guard = MutexGuard::try_map::<(), _>(guard, |_| panic!());
});
match lock.try_lock() {
Ok(_) => panic!("panicking in a MutexGuard::try_map closure should poison the Mutex"),
Err(TryLockError::WouldBlock) => {
panic!("panicking in a MutexGuard::try_map closure should unlock the mutex")
}
Err(TryLockError::Poisoned(_)) => {}
}
let _ = panic::catch_unwind(|| {
let guard = lock.lock().unwrap();
let guard = MutexGuard::map::<(), _>(guard, |val| val);
let _guard = MappedMutexGuard::map::<(), _>(guard, |_| panic!());
});
match lock.try_lock() {
Ok(_) => panic!("panicking in a MappedMutexGuard::map closure should poison the Mutex"),
Err(TryLockError::WouldBlock) => {
panic!("panicking in a MappedMutexGuard::map closure should unlock the mutex")
}
Err(TryLockError::Poisoned(_)) => {}
}
let _ = panic::catch_unwind(|| {
let guard = lock.lock().unwrap();
let guard = MutexGuard::map::<(), _>(guard, |val| val);
let _guard = MappedMutexGuard::try_map::<(), _>(guard, |_| panic!());
});
match lock.try_lock() {
Ok(_) => panic!("panicking in a MappedMutexGuard::try_map closure should poison the Mutex"),
Err(TryLockError::WouldBlock) => {
panic!("panicking in a MappedMutexGuard::try_map closure should unlock the mutex")
}
Err(TryLockError::Poisoned(_)) => {}
}
drop(lock);
}

3
library/std/src/sync/poison/once.rs
View file

@ -3,9 +3,6 @@
//! This primitive is meant to be used to run one-time initialization. An
//! example use case would be for initializing an FFI library.
#[cfg(all(test, not(any(target_os = "emscripten", target_os = "wasi"))))]
mod tests;
use crate::fmt;
use crate::panic::{RefUnwindSafe, UnwindSafe};
use crate::sys::sync as sys;

162
library/std/src/sync/poison/once/tests.rs
View file

@ -1,162 +0,0 @@
use super::Once;
use crate::sync::atomic::AtomicBool;
use crate::sync::atomic::Ordering::Relaxed;
use crate::sync::mpsc::channel;
use crate::time::Duration;
use crate::{panic, thread};
#[test]
fn smoke_once() {
static O: Once = Once::new();
let mut a = 0;
O.call_once(|| a += 1);
assert_eq!(a, 1);
O.call_once(|| a += 1);
assert_eq!(a, 1);
}
#[test]
fn stampede_once() {
static O: Once = Once::new();
static mut RUN: bool = false;
let (tx, rx) = channel();
for _ in 0..10 {
let tx = tx.clone();
thread::spawn(move || {
for _ in 0..4 {
thread::yield_now()
}
unsafe {
O.call_once(|| {
assert!(!RUN);
RUN = true;
});
assert!(RUN);
}
tx.send(()).unwrap();
});
}
unsafe {
O.call_once(|| {
assert!(!RUN);
RUN = true;
});
assert!(RUN);
}
for _ in 0..10 {
rx.recv().unwrap();
}
}
#[test]
fn poison_bad() {
static O: Once = Once::new();
// poison the once
let t = panic::catch_unwind(|| {
O.call_once(|| panic!());
});
assert!(t.is_err());
// poisoning propagates
let t = panic::catch_unwind(|| {
O.call_once(|| {});
});
assert!(t.is_err());
// we can subvert poisoning, however
let mut called = false;
O.call_once_force(|p| {
called = true;
assert!(p.is_poisoned())
});
assert!(called);
// once any success happens, we stop propagating the poison
O.call_once(|| {});
}
#[test]
fn wait_for_force_to_finish() {
static O: Once = Once::new();
// poison the once
let t = panic::catch_unwind(|| {
O.call_once(|| panic!());
});
assert!(t.is_err());
// make sure someone's waiting inside the once via a force
let (tx1, rx1) = channel();
let (tx2, rx2) = channel();
let t1 = thread::spawn(move || {
O.call_once_force(|p| {
assert!(p.is_poisoned());
tx1.send(()).unwrap();
rx2.recv().unwrap();
});
});
rx1.recv().unwrap();
// put another waiter on the once
let t2 = thread::spawn(|| {
let mut called = false;
O.call_once(|| {
called = true;
});
assert!(!called);
});
tx2.send(()).unwrap();
assert!(t1.join().is_ok());
assert!(t2.join().is_ok());
}
#[test]
fn wait() {
for _ in 0..50 {
let val = AtomicBool::new(false);
let once = Once::new();
thread::scope(|s| {
for _ in 0..4 {
s.spawn(|| {
once.wait();
assert!(val.load(Relaxed));
});
}
once.call_once(|| val.store(true, Relaxed));
});
}
}
#[test]
fn wait_on_poisoned() {
let once = Once::new();
panic::catch_unwind(|| once.call_once(|| panic!())).unwrap_err();
panic::catch_unwind(|| once.wait()).unwrap_err();
}
#[test]
fn wait_force_on_poisoned() {
let once = Once::new();
thread::scope(|s| {
panic::catch_unwind(|| once.call_once(|| panic!())).unwrap_err();
s.spawn(|| {
thread::sleep(Duration::from_millis(100));
once.call_once_force(|_| {});
});
once.wait_force();
})
}

3
library/std/src/sync/poison/rwlock.rs
View file

@ -1,6 +1,3 @@
#[cfg(all(test, not(any(target_os = "emscripten", target_os = "wasi"))))]
mod tests;
use crate::cell::UnsafeCell;
use crate::fmt;
use crate::marker::PhantomData;

729
library/std/src/sync/poison/rwlock/tests.rs
View file

@ -1,729 +0,0 @@
use rand::Rng;
use crate::fmt::Debug;
use crate::ops::FnMut;
use crate::panic::{self, AssertUnwindSafe};
use crate::sync::atomic::{AtomicUsize, Ordering};
use crate::sync::mpsc::channel;
use crate::sync::{
Arc, MappedRwLockReadGuard, MappedRwLockWriteGuard, RwLock, RwLockReadGuard, RwLockWriteGuard,
TryLockError,
};
use crate::{hint, mem, thread};
#[derive(Eq, PartialEq, Debug)]
struct NonCopy(i32);
#[derive(Eq, PartialEq, Debug)]
struct NonCopyNeedsDrop(i32);
impl Drop for NonCopyNeedsDrop {
fn drop(&mut self) {
hint::black_box(());
}
}
#[test]
fn test_needs_drop() {
assert!(!mem::needs_drop::<NonCopy>());
assert!(mem::needs_drop::<NonCopyNeedsDrop>());
}
#[derive(Clone, Eq, PartialEq, Debug)]
struct Cloneable(i32);
#[test]
fn smoke() {
let l = RwLock::new(());
drop(l.read().unwrap());
drop(l.write().unwrap());
drop((l.read().unwrap(), l.read().unwrap()));
drop(l.write().unwrap());
}
#[test]
// FIXME: On macOS we use a provenance-incorrect implementation and Miri
// catches that issue with a chance of around 1/1000.
// See <https://github.com/rust-lang/rust/issues/121950> for details.
#[cfg_attr(all(miri, target_os = "macos"), ignore)]
fn frob() {
const N: u32 = 10;
const M: usize = if cfg!(miri) { 100 } else { 1000 };
let r = Arc::new(RwLock::new(()));
let (tx, rx) = channel::<()>();
for _ in 0..N {
let tx = tx.clone();
let r = r.clone();
thread::spawn(move || {
let mut rng = crate::test_helpers::test_rng();
for _ in 0..M {
if rng.gen_bool(1.0 / (N as f64)) {
drop(r.write().unwrap());
} else {
drop(r.read().unwrap());
}
}
drop(tx);
});
}
drop(tx);
let _ = rx.recv();
}
#[test]
fn test_rw_arc_poison_wr() {
let arc = Arc::new(RwLock::new(1));
let arc2 = arc.clone();
let _: Result<(), _> = thread::spawn(move || {
let _lock = arc2.write().unwrap();
panic!();
})
.join();
assert!(arc.read().is_err());
}
#[test]
fn test_rw_arc_poison_mapped_w_r() {
let arc = Arc::new(RwLock::new(1));
let arc2 = arc.clone();
let _: Result<(), _> = thread::spawn(move || {
let lock = arc2.write().unwrap();
let _lock = RwLockWriteGuard::map(lock, |val| val);
panic!();
})
.join();
assert!(arc.read().is_err());
}
#[test]
fn test_rw_arc_poison_ww() {
let arc = Arc::new(RwLock::new(1));
assert!(!arc.is_poisoned());
let arc2 = arc.clone();
let _: Result<(), _> = thread::spawn(move || {
let _lock = arc2.write().unwrap();
panic!();
})
.join();
assert!(arc.write().is_err());
assert!(arc.is_poisoned());
}
#[test]
fn test_rw_arc_poison_mapped_w_w() {
let arc = Arc::new(RwLock::new(1));
let arc2 = arc.clone();
let _: Result<(), _> = thread::spawn(move || {
let lock = arc2.write().unwrap();
let _lock = RwLockWriteGuard::map(lock, |val| val);
panic!();
})
.join();
assert!(arc.write().is_err());
assert!(arc.is_poisoned());
}
#[test]
fn test_rw_arc_no_poison_rr() {
let arc = Arc::new(RwLock::new(1));
let arc2 = arc.clone();
let _: Result<(), _> = thread::spawn(move || {
let _lock = arc2.read().unwrap();
panic!();
})
.join();
let lock = arc.read().unwrap();
assert_eq!(*lock, 1);
}
#[test]
fn test_rw_arc_no_poison_mapped_r_r() {
let arc = Arc::new(RwLock::new(1));
let arc2 = arc.clone();
let _: Result<(), _> = thread::spawn(move || {
let lock = arc2.read().unwrap();
let _lock = RwLockReadGuard::map(lock, |val| val);
panic!();
})
.join();
let lock = arc.read().unwrap();
assert_eq!(*lock, 1);
}
#[test]
fn test_rw_arc_no_poison_rw() {
let arc = Arc::new(RwLock::new(1));
let arc2 = arc.clone();
let _: Result<(), _> = thread::spawn(move || {
let _lock = arc2.read().unwrap();
panic!()
})
.join();
let lock = arc.write().unwrap();
assert_eq!(*lock, 1);
}
#[test]
fn test_rw_arc_no_poison_mapped_r_w() {
let arc = Arc::new(RwLock::new(1));
let arc2 = arc.clone();
let _: Result<(), _> = thread::spawn(move || {
let lock = arc2.read().unwrap();
let _lock = RwLockReadGuard::map(lock, |val| val);
panic!();
})
.join();
let lock = arc.write().unwrap();
assert_eq!(*lock, 1);
}
#[test]
fn test_rw_arc() {
let arc = Arc::new(RwLock::new(0));
let arc2 = arc.clone();
let (tx, rx) = channel();
thread::spawn(move || {
let mut lock = arc2.write().unwrap();
for _ in 0..10 {
let tmp = *lock;
*lock = -1;
thread::yield_now();
*lock = tmp + 1;
}
tx.send(()).unwrap();
});
// Readers try to catch the writer in the act
let mut children = Vec::new();
for _ in 0..5 {
let arc3 = arc.clone();
children.push(thread::spawn(move || {
let lock = arc3.read().unwrap();
assert!(*lock >= 0);
}));
}
// Wait for children to pass their asserts
for r in children {
assert!(r.join().is_ok());
}
// Wait for writer to finish
rx.recv().unwrap();
let lock = arc.read().unwrap();
assert_eq!(*lock, 10);
}
#[test]
fn test_rw_arc_access_in_unwind() {
let arc = Arc::new(RwLock::new(1));
let arc2 = arc.clone();
let _ = thread::spawn(move || -> () {
struct Unwinder {
i: Arc<RwLock<isize>>,
}
impl Drop for Unwinder {
fn drop(&mut self) {
let mut lock = self.i.write().unwrap();
*lock += 1;
}
}
let _u = Unwinder { i: arc2 };
panic!();
})
.join();
let lock = arc.read().unwrap();
assert_eq!(*lock, 2);
}
#[test]
fn test_rwlock_unsized() {
let rw: &RwLock<[i32]> = &RwLock::new([1, 2, 3]);
{
let b = &mut *rw.write().unwrap();
b[0] = 4;
b[2] = 5;
}
let comp: &[i32] = &[4, 2, 5];
assert_eq!(&*rw.read().unwrap(), comp);
}
#[test]
fn test_rwlock_try_write() {
let lock = RwLock::new(0isize);
let read_guard = lock.read().unwrap();
let write_result = lock.try_write();
match write_result {
Err(TryLockError::WouldBlock) => (),
Ok(_) => assert!(false, "try_write should not succeed while read_guard is in scope"),
Err(_) => assert!(false, "unexpected error"),
}
drop(read_guard);
let mapped_read_guard = RwLockReadGuard::map(lock.read().unwrap(), |_| &());
let write_result = lock.try_write();
match write_result {
Err(TryLockError::WouldBlock) => (),
Ok(_) => assert!(false, "try_write should not succeed while mapped_read_guard is in scope"),
Err(_) => assert!(false, "unexpected error"),
}
drop(mapped_read_guard);
}
fn new_poisoned_rwlock<T>(value: T) -> RwLock<T> {
let lock = RwLock::new(value);
let catch_unwind_result = panic::catch_unwind(AssertUnwindSafe(|| {
let _guard = lock.write().unwrap();
panic!("test panic to poison RwLock");
}));
assert!(catch_unwind_result.is_err());
assert!(lock.is_poisoned());
lock
}
#[test]
fn test_into_inner() {
let m = RwLock::new(NonCopy(10));
assert_eq!(m.into_inner().unwrap(), NonCopy(10));
}
#[test]
fn test_into_inner_drop() {
struct Foo(Arc<AtomicUsize>);
impl Drop for Foo {
fn drop(&mut self) {
self.0.fetch_add(1, Ordering::SeqCst);
}
}
let num_drops = Arc::new(AtomicUsize::new(0));
let m = RwLock::new(Foo(num_drops.clone()));
assert_eq!(num_drops.load(Ordering::SeqCst), 0);
{
let _inner = m.into_inner().unwrap();
assert_eq!(num_drops.load(Ordering::SeqCst), 0);
}
assert_eq!(num_drops.load(Ordering::SeqCst), 1);
}
#[test]
fn test_into_inner_poison() {
let m = new_poisoned_rwlock(NonCopy(10));
match m.into_inner() {
Err(e) => assert_eq!(e.into_inner(), NonCopy(10)),
Ok(x) => panic!("into_inner of poisoned RwLock is Ok: {x:?}"),
}
}
#[test]
fn test_get_cloned() {
let m = RwLock::new(Cloneable(10));
assert_eq!(m.get_cloned().unwrap(), Cloneable(10));
}
#[test]
fn test_get_cloned_poison() {
let m = new_poisoned_rwlock(Cloneable(10));
match m.get_cloned() {
Err(e) => assert_eq!(e.into_inner(), ()),
Ok(x) => panic!("get of poisoned RwLock is Ok: {x:?}"),
}
}
#[test]
fn test_get_mut() {
let mut m = RwLock::new(NonCopy(10));
*m.get_mut().unwrap() = NonCopy(20);
assert_eq!(m.into_inner().unwrap(), NonCopy(20));
}
#[test]
fn test_get_mut_poison() {
let mut m = new_poisoned_rwlock(NonCopy(10));
match m.get_mut() {
Err(e) => assert_eq!(*e.into_inner(), NonCopy(10)),
Ok(x) => panic!("get_mut of poisoned RwLock is Ok: {x:?}"),
}
}
#[test]
fn test_set() {
fn inner<T>(mut init: impl FnMut() -> T, mut value: impl FnMut() -> T)
where
T: Debug + Eq,
{
let m = RwLock::new(init());
assert_eq!(*m.read().unwrap(), init());
m.set(value()).unwrap();
assert_eq!(*m.read().unwrap(), value());
}
inner(|| NonCopy(10), || NonCopy(20));
inner(|| NonCopyNeedsDrop(10), || NonCopyNeedsDrop(20));
}
#[test]
fn test_set_poison() {
fn inner<T>(mut init: impl FnMut() -> T, mut value: impl FnMut() -> T)
where
T: Debug + Eq,
{
let m = new_poisoned_rwlock(init());
match m.set(value()) {
Err(e) => {
assert_eq!(e.into_inner(), value());
assert_eq!(m.into_inner().unwrap_err().into_inner(), init());
}
Ok(x) => panic!("set of poisoned RwLock is Ok: {x:?}"),
}
}
inner(|| NonCopy(10), || NonCopy(20));
inner(|| NonCopyNeedsDrop(10), || NonCopyNeedsDrop(20));
}
#[test]
fn test_replace() {
fn inner<T>(mut init: impl FnMut() -> T, mut value: impl FnMut() -> T)
where
T: Debug + Eq,
{
let m = RwLock::new(init());
assert_eq!(*m.read().unwrap(), init());
assert_eq!(m.replace(value()).unwrap(), init());
assert_eq!(*m.read().unwrap(), value());
}
inner(|| NonCopy(10), || NonCopy(20));
inner(|| NonCopyNeedsDrop(10), || NonCopyNeedsDrop(20));
}
#[test]
fn test_replace_poison() {
fn inner<T>(mut init: impl FnMut() -> T, mut value: impl FnMut() -> T)
where
T: Debug + Eq,
{
let m = new_poisoned_rwlock(init());
match m.replace(value()) {
Err(e) => {
assert_eq!(e.into_inner(), value());
assert_eq!(m.into_inner().unwrap_err().into_inner(), init());
}
Ok(x) => panic!("replace of poisoned RwLock is Ok: {x:?}"),
}
}
inner(|| NonCopy(10), || NonCopy(20));
inner(|| NonCopyNeedsDrop(10), || NonCopyNeedsDrop(20));
}
#[test]
fn test_read_guard_covariance() {
fn do_stuff<'a>(_: RwLockReadGuard<'_, &'a i32>, _: &'a i32) {}
let j: i32 = 5;
let lock = RwLock::new(&j);
{
let i = 6;
do_stuff(lock.read().unwrap(), &i);
}
drop(lock);
}
#[test]
fn test_mapped_read_guard_covariance() {
fn do_stuff<'a>(_: MappedRwLockReadGuard<'_, &'a i32>, _: &'a i32) {}
let j: i32 = 5;
let lock = RwLock::new((&j, &j));
{
let i = 6;
let guard = lock.read().unwrap();
let guard = RwLockReadGuard::map(guard, |(val, _val)| val);
do_stuff(guard, &i);
}
drop(lock);
}
#[test]
fn test_mapping_mapped_guard() {
let arr = [0; 4];
let mut lock = RwLock::new(arr);
let guard = lock.write().unwrap();
let guard = RwLockWriteGuard::map(guard, |arr| &mut arr[..2]);
let mut guard = MappedRwLockWriteGuard::map(guard, |slice| &mut slice[1..]);
assert_eq!(guard.len(), 1);
guard[0] = 42;
drop(guard);
assert_eq!(*lock.get_mut().unwrap(), [0, 42, 0, 0]);
let guard = lock.read().unwrap();
let guard = RwLockReadGuard::map(guard, |arr| &arr[..2]);
let guard = MappedRwLockReadGuard::map(guard, |slice| &slice[1..]);
assert_eq!(*guard, [42]);
drop(guard);
assert_eq!(*lock.get_mut().unwrap(), [0, 42, 0, 0]);
}
#[test]
fn panic_while_mapping_read_unlocked_no_poison() {
let lock = RwLock::new(());
let _ = panic::catch_unwind(|| {
let guard = lock.read().unwrap();
let _guard = RwLockReadGuard::map::<(), _>(guard, |_| panic!());
});
match lock.try_write() {
Ok(_) => {}
Err(TryLockError::WouldBlock) => {
panic!("panicking in a RwLockReadGuard::map closure should release the read lock")
}
Err(TryLockError::Poisoned(_)) => {
panic!("panicking in a RwLockReadGuard::map closure should not poison the RwLock")
}
}
let _ = panic::catch_unwind(|| {
let guard = lock.read().unwrap();
let _guard = RwLockReadGuard::try_map::<(), _>(guard, |_| panic!());
});
match lock.try_write() {
Ok(_) => {}
Err(TryLockError::WouldBlock) => {
panic!("panicking in a RwLockReadGuard::try_map closure should release the read lock")
}
Err(TryLockError::Poisoned(_)) => {
panic!("panicking in a RwLockReadGuard::try_map closure should not poison the RwLock")
}
}
let _ = panic::catch_unwind(|| {
let guard = lock.read().unwrap();
let guard = RwLockReadGuard::map::<(), _>(guard, |val| val);
let _guard = MappedRwLockReadGuard::map::<(), _>(guard, |_| panic!());
});
match lock.try_write() {
Ok(_) => {}
Err(TryLockError::WouldBlock) => {
panic!("panicking in a MappedRwLockReadGuard::map closure should release the read lock")
}
Err(TryLockError::Poisoned(_)) => {
panic!("panicking in a MappedRwLockReadGuard::map closure should not poison the RwLock")
}
}
let _ = panic::catch_unwind(|| {
let guard = lock.read().unwrap();
let guard = RwLockReadGuard::map::<(), _>(guard, |val| val);
let _guard = MappedRwLockReadGuard::try_map::<(), _>(guard, |_| panic!());
});
match lock.try_write() {
Ok(_) => {}
Err(TryLockError::WouldBlock) => panic!(
"panicking in a MappedRwLockReadGuard::try_map closure should release the read lock"
),
Err(TryLockError::Poisoned(_)) => panic!(
"panicking in a MappedRwLockReadGuard::try_map closure should not poison the RwLock"
),
}
drop(lock);
}
#[test]
fn panic_while_mapping_write_unlocked_poison() {
let lock = RwLock::new(());
let _ = panic::catch_unwind(|| {
let guard = lock.write().unwrap();
let _guard = RwLockWriteGuard::map::<(), _>(guard, |_| panic!());
});
match lock.try_write() {
Ok(_) => panic!("panicking in a RwLockWriteGuard::map closure should poison the RwLock"),
Err(TryLockError::WouldBlock) => {
panic!("panicking in a RwLockWriteGuard::map closure should release the write lock")
}
Err(TryLockError::Poisoned(_)) => {}
}
let _ = panic::catch_unwind(|| {
let guard = lock.write().unwrap();
let _guard = RwLockWriteGuard::try_map::<(), _>(guard, |_| panic!());
});
match lock.try_write() {
Ok(_) => {
panic!("panicking in a RwLockWriteGuard::try_map closure should poison the RwLock")
}
Err(TryLockError::WouldBlock) => {
panic!("panicking in a RwLockWriteGuard::try_map closure should release the write lock")
}
Err(TryLockError::Poisoned(_)) => {}
}
let _ = panic::catch_unwind(|| {
let guard = lock.write().unwrap();
let guard = RwLockWriteGuard::map::<(), _>(guard, |val| val);
let _guard = MappedRwLockWriteGuard::map::<(), _>(guard, |_| panic!());
});
match lock.try_write() {
Ok(_) => {
panic!("panicking in a MappedRwLockWriteGuard::map closure should poison the RwLock")
}
Err(TryLockError::WouldBlock) => panic!(
"panicking in a MappedRwLockWriteGuard::map closure should release the write lock"
),
Err(TryLockError::Poisoned(_)) => {}
}
let _ = panic::catch_unwind(|| {
let guard = lock.write().unwrap();
let guard = RwLockWriteGuard::map::<(), _>(guard, |val| val);
let _guard = MappedRwLockWriteGuard::try_map::<(), _>(guard, |_| panic!());
});
match lock.try_write() {
Ok(_) => panic!(
"panicking in a MappedRwLockWriteGuard::try_map closure should poison the RwLock"
),
Err(TryLockError::WouldBlock) => panic!(
"panicking in a MappedRwLockWriteGuard::try_map closure should release the write lock"
),
Err(TryLockError::Poisoned(_)) => {}
}
drop(lock);
}
#[test]
fn test_downgrade_basic() {
let r = RwLock::new(());
let write_guard = r.write().unwrap();
let _read_guard = RwLockWriteGuard::downgrade(write_guard);
}
#[test]
// FIXME: On macOS we use a provenance-incorrect implementation and Miri catches that issue.
// See <https://github.com/rust-lang/rust/issues/121950> for details.
#[cfg_attr(all(miri, target_os = "macos"), ignore)]
fn test_downgrade_observe() {
// Taken from the test `test_rwlock_downgrade` from:
// https://github.com/Amanieu/parking_lot/blob/master/src/rwlock.rs
const W: usize = 20;
const N: usize = if cfg!(miri) { 40 } else { 100 };
// This test spawns `W` writer threads, where each will increment a counter `N` times, ensuring
// that the value they wrote has not changed after downgrading.
let rw = Arc::new(RwLock::new(0));
// Spawn the writers that will do `W * N` operations and checks.
let handles: Vec<_> = (0..W)
.map(|_| {
let rw = rw.clone();
thread::spawn(move || {
for _ in 0..N {
// Increment the counter.
let mut write_guard = rw.write().unwrap();
*write_guard += 1;
let cur_val = *write_guard;
// Downgrade the lock to read mode, where the value protected cannot be modified.
let read_guard = RwLockWriteGuard::downgrade(write_guard);
assert_eq!(cur_val, *read_guard);
}
})
})
.collect();
for handle in handles {
handle.join().unwrap();
}
assert_eq!(*rw.read().unwrap(), W * N);
}
#[test]
// FIXME: On macOS we use a provenance-incorrect implementation and Miri catches that issue.
// See <https://github.com/rust-lang/rust/issues/121950> for details.
#[cfg_attr(all(miri, target_os = "macos"), ignore)]
fn test_downgrade_atomic() {
const NEW_VALUE: i32 = -1;
// This test checks that `downgrade` is atomic, meaning as soon as a write lock has been
// downgraded, the lock must be in read mode and no other threads can take the write lock to
// modify the protected value.
// `W` is the number of evil writer threads.
const W: usize = 20;
let rwlock = Arc::new(RwLock::new(0));
// Spawns many evil writer threads that will try and write to the locked value before the
// initial writer (who has the exclusive lock) can read after it downgrades.
// If the `RwLock` behaves correctly, then the initial writer should read the value it wrote
// itself as no other thread should be able to mutate the protected value.
// Put the lock in write mode, causing all future threads trying to access this go to sleep.
let mut main_write_guard = rwlock.write().unwrap();
// Spawn all of the evil writer threads. They will each increment the protected value by 1.
let handles: Vec<_> = (0..W)
.map(|_| {
let rwlock = rwlock.clone();
thread::spawn(move || {
// Will go to sleep since the main thread initially has the write lock.
let mut evil_guard = rwlock.write().unwrap();
*evil_guard += 1;
})
})
.collect();
// Wait for a good amount of time so that evil threads go to sleep.
// Note: this is not strictly necessary...
let eternity = crate::time::Duration::from_millis(42);
thread::sleep(eternity);
// Once everyone is asleep, set the value to `NEW_VALUE`.
*main_write_guard = NEW_VALUE;
// Atomically downgrade the write guard into a read guard.
let main_read_guard = RwLockWriteGuard::downgrade(main_write_guard);
// If the above is not atomic, then it would be possible for an evil thread to get in front of
// this read and change the value to be non-negative.
assert_eq!(*main_read_guard, NEW_VALUE, "`downgrade` was not atomic");
// Drop the main read guard and allow the evil writer threads to start incrementing.
drop(main_read_guard);
for handle in handles {
handle.join().unwrap();
}
let final_check = rwlock.read().unwrap();
assert_eq!(*final_check, W as i32 + NEW_VALUE);
}

8
library/std/src/sync/reentrant_lock.rs
View file

@ -1,6 +1,3 @@
#[cfg(all(test, not(any(target_os = "emscripten", target_os = "wasi"))))]
mod tests;
use cfg_if::cfg_if;
use crate::cell::UnsafeCell;
@ -324,7 +321,10 @@ impl<T: ?Sized> ReentrantLock<T> {
/// Otherwise, an RAII guard is returned.
///
/// This function does not block.
pub(crate) fn try_lock(&self) -> Option<ReentrantLockGuard<'_, T>> {
// FIXME maybe make it a public part of the API?
#[unstable(issue = "none", feature = "std_internals")]
#[doc(hidden)]
pub fn try_lock(&self) -> Option<ReentrantLockGuard<'_, T>> {
let this_thread = current_id();
// Safety: We only touch lock_count when we own the inner mutex.
// Additionally, we only call `self.owner.set()` while holding

53
library/std/src/sync/reentrant_lock/tests.rs
View file

@ -1,53 +0,0 @@
use super::ReentrantLock;
use crate::cell::RefCell;
use crate::sync::Arc;
use crate::thread;
#[test]
fn smoke() {
let l = ReentrantLock::new(());
{
let a = l.lock();
{
let b = l.lock();
{
let c = l.lock();
assert_eq!(*c, ());
}
assert_eq!(*b, ());
}
assert_eq!(*a, ());
}
}
#[test]
fn is_mutex() {
let l = Arc::new(ReentrantLock::new(RefCell::new(0)));
let l2 = l.clone();
let lock = l.lock();
let child = thread::spawn(move || {
let lock = l2.lock();
assert_eq!(*lock.borrow(), 4950);
});
for i in 0..100 {
let lock = l.lock();
*lock.borrow_mut() += i;
}
drop(lock);
child.join().unwrap();
}
#[test]
fn trylock_works() {
let l = Arc::new(ReentrantLock::new(()));
let l2 = l.clone();
let _lock = l.try_lock();
let _lock2 = l.try_lock();
thread::spawn(move || {
let lock = l2.try_lock();
assert!(lock.is_none());
})
.join()
.unwrap();
let _lock3 = l.try_lock();
}