sync: add track_caller to public APIs (#4808)

Functions that may panic can be annotated with `#[track_caller]` so that
in the event of a panic, the function where the user called the
panicking function is shown instead of the file and line within Tokio
source.

This change adds `#[track_caller]` to all the public APIs in the sync
module of the tokio crate where the documentation describes how the
function may panic due to incorrect context or inputs.

In cases where `#[track_caller]` does not work, it has been left out.
For example, it currently does not work on async functions, blocks, or
closures. So any call stack that passes through one of these before
reaching the actual panic is not able to show the calling site outside
of tokio as the panic location.

The following functions have call stacks that pass through closures:
* `sync::watch::Sender::send_modify`
* `sync::watch::Sender::send_if_modified`

Additionally, in the above functions it is a panic inside the supplied
closure which causes the function to panic, and so showing the location
of the panic itself is desirable.

The following functions are async:
* `sync::mpsc::bounded::Sender::send_timeout`

Tests are included to cover each potentially panicking function.

Refs: #4413

Author: hds

tokio/src/future/block_on.rs

@@ -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()

tokio/src/runtime/enter.rs

@@ -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;

tokio/src/sync/broadcast.rs

@@ -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");

tokio/src/sync/mpsc/bounded.rs

@@ -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);
@@ -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())
@@ -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))

tokio/src/sync/mpsc/unbounded.rs

@@ -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())

tokio/src/sync/mutex.rs

@@ -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())

tokio/src/sync/oneshot.rs

@@ -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)

tokio/src/sync/rwlock.rs

@@ -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())
@@ -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())

tokio/tests/sync_panic.rs

@@ -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()
+}