Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions tokio/src/future/block_on.rs
Original file line number Diff line number Diff line change
@@ -1,13 +1,15 @@
use std::future::Future;

cfg_rt! {
#[track_caller]
pub(crate) fn block_on<F: Future>(f: F) -> F::Output {
let mut e = crate::runtime::enter::enter(false);
e.block_on(f).unwrap()
}
}

cfg_not_rt! {
#[track_caller]
pub(crate) fn block_on<F: Future>(f: F) -> F::Output {
let mut park = crate::park::thread::CachedParkThread::new();
park.block_on(f).unwrap()
Expand Down
1 change: 1 addition & 0 deletions tokio/src/runtime/enter.rs
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ cfg_rt! {

/// Marks the current thread as being within the dynamic extent of an
/// executor.
#[track_caller]
pub(crate) fn enter(allow_blocking: bool) -> Enter {
if let Some(enter) = try_enter(allow_blocking) {
return enter;
Expand Down
1 change: 1 addition & 0 deletions tokio/src/sync/broadcast.rs
Original file line number Diff line number Diff line change
Expand Up @@ -430,6 +430,7 @@ const MAX_RECEIVERS: usize = usize::MAX >> 2;
///
/// This will panic if `capacity` is equal to `0` or larger
/// than `usize::MAX / 2`.
#[track_caller]
pub fn channel<T: Clone>(mut capacity: usize) -> (Sender<T>, Receiver<T>) {
assert!(capacity > 0, "capacity is empty");
assert!(capacity <= usize::MAX >> 1, "requested capacity too large");
Expand Down
3 changes: 3 additions & 0 deletions tokio/src/sync/mpsc/bounded.rs
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,7 @@ pub struct Receiver<T> {
/// }
/// }
/// ```
#[track_caller]
pub fn channel<T>(buffer: usize) -> (Sender<T>, Receiver<T>) {
assert!(buffer > 0, "mpsc bounded channel requires buffer > 0");
let semaphore = (semaphore::Semaphore::new(buffer), buffer);
Expand Down Expand Up @@ -281,6 +282,7 @@ impl<T> Receiver<T> {
/// sync_code.join().unwrap()
/// }
/// ```
#[track_caller]
#[cfg(feature = "sync")]
pub fn blocking_recv(&mut self) -> Option<T> {
crate::future::block_on(self.recv())
Expand Down Expand Up @@ -650,6 +652,7 @@ impl<T> Sender<T> {
/// sync_code.join().unwrap()
/// }
/// ```
#[track_caller]
#[cfg(feature = "sync")]
pub fn blocking_send(&self, value: T) -> Result<(), SendError<T>> {
crate::future::block_on(self.send(value))
Expand Down
1 change: 1 addition & 0 deletions tokio/src/sync/mpsc/unbounded.rs
Original file line number Diff line number Diff line change
Expand Up @@ -206,6 +206,7 @@ impl<T> UnboundedReceiver<T> {
/// sync_code.join().unwrap();
/// }
/// ```
#[track_caller]
#[cfg(feature = "sync")]
pub fn blocking_recv(&mut self) -> Option<T> {
crate::future::block_on(self.recv())
Expand Down
1 change: 1 addition & 0 deletions tokio/src/sync/mutex.rs
Original file line number Diff line number Diff line change
Expand Up @@ -411,6 +411,7 @@ impl<T: ?Sized> Mutex<T> {
/// }
///
/// ```
#[track_caller]
#[cfg(feature = "sync")]
pub fn blocking_lock(&self) -> MutexGuard<'_, T> {
crate::future::block_on(self.lock())
Expand Down
1 change: 1 addition & 0 deletions tokio/src/sync/oneshot.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1052,6 +1052,7 @@ impl<T> Receiver<T> {
/// sync_code.join().unwrap();
/// }
/// ```
#[track_caller]
#[cfg(feature = "sync")]
pub fn blocking_recv(self) -> Result<T, RecvError> {
crate::future::block_on(self)
Expand Down
2 changes: 2 additions & 0 deletions tokio/src/sync/rwlock.rs
Original file line number Diff line number Diff line change
Expand Up @@ -506,6 +506,7 @@ impl<T: ?Sized> RwLock<T> {
/// assert!(rwlock.try_write().is_ok());
/// }
/// ```
#[track_caller]
#[cfg(feature = "sync")]
pub fn blocking_read(&self) -> RwLockReadGuard<'_, T> {
crate::future::block_on(self.read())
Expand Down Expand Up @@ -840,6 +841,7 @@ impl<T: ?Sized> RwLock<T> {
/// assert_eq!(*read_lock, 2);
/// }
/// ```
#[track_caller]
#[cfg(feature = "sync")]
pub fn blocking_write(&self) -> RwLockWriteGuard<'_, T> {
crate::future::block_on(self.write())
Expand Down
165 changes: 165 additions & 0 deletions tokio/tests/sync_panic.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,165 @@
#![warn(rust_2018_idioms)]
#![cfg(feature = "full")]

use std::error::Error;
use tokio::{
runtime::{Builder, Runtime},
sync::{broadcast, mpsc, oneshot, Mutex, RwLock},
};

mod support {
pub mod panic;
}
use support::panic::test_panic;

#[test]
fn broadcast_channel_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let (_, _) = broadcast::channel::<u32>(0);
});

// The panic location should be in this file
assert_eq!(&panic_location_file.unwrap(), file!());

Ok(())
}

#[test]
fn mutex_blocking_lock_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let rt = basic();
rt.block_on(async {
let mutex = Mutex::new(5_u32);
mutex.blocking_lock();
});
});

// The panic location should be in this file
assert_eq!(&panic_location_file.unwrap(), file!());

Ok(())
}

#[test]
fn oneshot_blocking_recv_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let rt = basic();
rt.block_on(async {
let (_tx, rx) = oneshot::channel::<u8>();
let _ = rx.blocking_recv();
});
});

// The panic location should be in this file
assert_eq!(&panic_location_file.unwrap(), file!());

Ok(())
}

#[test]
fn rwlock_with_max_readers_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let _ = RwLock::<u8>::with_max_readers(0, (u32::MAX >> 3) + 1);
});

// The panic location should be in this file
assert_eq!(&panic_location_file.unwrap(), file!());

Ok(())
}

#[test]
fn rwlock_blocking_read_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let rt = basic();
rt.block_on(async {
let lock = RwLock::<u8>::new(0);
let _ = lock.blocking_read();
});
});

// The panic location should be in this file
assert_eq!(&panic_location_file.unwrap(), file!());

Ok(())
}

#[test]
fn rwlock_blocking_write_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let rt = basic();
rt.block_on(async {
let lock = RwLock::<u8>::new(0);
let _ = lock.blocking_write();
});
});

// The panic location should be in this file
assert_eq!(&panic_location_file.unwrap(), file!());

Ok(())
}

#[test]
fn mpsc_bounded_channel_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let (_, _) = mpsc::channel::<u8>(0);
});

// The panic location should be in this file
assert_eq!(&panic_location_file.unwrap(), file!());

Ok(())
}

#[test]
fn mpsc_bounded_receiver_blocking_recv_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let rt = basic();
let (_tx, mut rx) = mpsc::channel::<u8>(1);
rt.block_on(async {
let _ = rx.blocking_recv();
});
});

// The panic location should be in this file
assert_eq!(&panic_location_file.unwrap(), file!());

Ok(())
}

#[test]
fn mpsc_bounded_sender_blocking_send_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let rt = basic();
let (tx, _rx) = mpsc::channel::<u8>(1);
rt.block_on(async {
let _ = tx.blocking_send(3);
});
});

// The panic location should be in this file
assert_eq!(&panic_location_file.unwrap(), file!());

Ok(())
}

#[test]
fn mpsc_unbounded_receiver_blocking_recv_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let rt = basic();
let (_tx, mut rx) = mpsc::unbounded_channel::<u8>();
rt.block_on(async {
let _ = rx.blocking_recv();
});
});

// The panic location should be in this file
assert_eq!(&panic_location_file.unwrap(), file!());

Ok(())
}

fn basic() -> Runtime {
Builder::new_current_thread().enable_all().build().unwrap()
}