Merge branch 'tokio-rs:master' into master

.github/workflows/ci.yml

@@ -17,6 +17,8 @@ env:
   # Change to specific Rust release to pin
   rust_stable: stable
   rust_nightly: nightly-2024-05-05
+  # Pin a specific miri version
+  rust_miri_nightly: nightly-2024-09-19
   rust_clippy: '1.77'
   # When updating this, also update:
   # - README.md
@@ -413,17 +415,19 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
-      - name: Install Rust ${{ env.rust_nightly }}
+      - name: Install Rust ${{ env.rust_miri_nightly }}
         uses: dtolnay/rust-toolchain@stable
         with:
-          toolchain: ${{ env.rust_nightly }}
+          toolchain: ${{ env.rust_miri_nightly }}
           components: miri
+      - name: Install cargo-nextest
+        uses: taiki-e/install-action@v2
+        with:
+          tool: cargo-nextest
       - uses: Swatinem/rust-cache@v2
       - name: miri
-        # Many of tests in tokio/tests and doctests use #[tokio::test] or
-        # #[tokio::main] that calls epoll_create1 that Miri does not support.
         run: |
-          cargo miri test --features full --lib --no-fail-fast
+          cargo miri nextest run --features full --lib --tests --no-fail-fast
         working-directory: tokio
         env:
           MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-strict-provenance -Zmiri-retag-fields

tests-build/tests/macros.rs

@@ -1,4 +1,5 @@
 #[test]
+#[cfg_attr(miri, ignore)]
 fn compile_fail_full() {
     let t = trybuild::TestCases::new();
 

tests-integration/tests/process_stdio.rs

@@ -1,5 +1,5 @@
 #![warn(rust_2018_idioms)]
-#![cfg(all(feature = "full", not(target_os = "wasi")))]
+#![cfg(all(feature = "full", not(target_os = "wasi"), not(miri)))]
 
 use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
 use tokio::join;

tokio-util/tests/spawn_pinned.rs

@@ -1,5 +1,7 @@
 #![warn(rust_2018_idioms)]
 #![cfg(not(target_os = "wasi"))] // Wasi doesn't support threads
+// Blocked on https://github.com/rust-lang/miri/issues/3911
+#![cfg(not(miri))]
 
 use std::rc::Rc;
 use std::sync::Arc;

tokio-util/tests/time_delay_queue.rs

@@ -92,6 +92,7 @@ async fn single_short_delay() {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // Too slow on miri.
 async fn multi_delay_at_start() {
     time::pause();
 

tokio-util/tests/udp.rs

@@ -1,5 +1,6 @@
 #![warn(rust_2018_idioms)]
 #![cfg(not(target_os = "wasi"))] // Wasi doesn't support UDP
+#![cfg(not(miri))] // No `socket` in Miri.
 
 use tokio::net::UdpSocket;
 use tokio_stream::StreamExt;

tokio/src/sync/watch.rs

@@ -1336,6 +1336,29 @@ impl<T> Sender<T> {
         self.shared.ref_count_rx.load(Relaxed)
     }
 
+    /// Returns the number of senders that currently exist.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::sync::watch;
+    ///
+    /// #[tokio::main]
+    /// async fn main() {
+    ///     let (tx1, rx) = watch::channel("hello");
+    ///
+    ///     assert_eq!(1, tx1.sender_count());
+    ///
+    ///     let tx2 = tx1.clone();
+    ///
+    ///     assert_eq!(2, tx1.sender_count());
+    ///     assert_eq!(2, tx2.sender_count());
+    /// }
+    /// ```
+    pub fn sender_count(&self) -> usize {
+        self.shared.ref_count_tx.load(Relaxed)
+    }
+
     /// Returns `true` if senders belong to the same channel.
     ///
     /// # Examples

tokio/tests/buffered.rs

@@ -9,6 +9,7 @@ use std::net::TcpStream;
 use std::thread;
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)]
 async fn echo_server() {
     const N: usize = 1024;
 

tokio/tests/coop_budget.rs

@@ -22,6 +22,7 @@ use tokio::net::UdpSocket;
 /// Since we are both sending and receiving, that should happen once per 64 packets, because budgets are of size 128
 /// and there are two budget events per packet, a send and a recv.
 #[tokio::test]
+#[cfg_attr(miri, ignore)]
 async fn coop_budget_udp_send_recv() {
     const BUDGET: usize = 128;
     const N_ITERATIONS: usize = 1024;

tokio/tests/fs_copy.rs

@@ -5,6 +5,7 @@ use tempfile::tempdir;
 use tokio::fs;
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)]
 async fn copy() {
     let dir = tempdir().unwrap();
 
@@ -21,6 +22,7 @@ async fn copy() {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)]
 async fn copy_permissions() {
     let dir = tempdir().unwrap();
     let from_path = dir.path().join("foo.txt");

tokio/tests/fs_file.rs

@@ -203,6 +203,7 @@ async fn set_max_buf_size_write() {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)]
 #[cfg(unix)]
 async fn file_debug_fmt() {
     let tempfile = tempfile();

tokio/tests/fs_link.rs

@@ -7,6 +7,7 @@ use std::io::Write;
 use tempfile::tempdir;
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)]
 async fn test_hard_link() {
     let dir = tempdir().unwrap();
     let src = dir.path().join("src.txt");

tokio/tests/fs_try_exists.rs

@@ -5,6 +5,7 @@ use tempfile::tempdir;
 use tokio::fs;
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)]
 async fn try_exists() {
     let dir = tempdir().unwrap();
 

tokio/tests/io_async_fd.rs

@@ -1,5 +1,5 @@
 #![warn(rust_2018_idioms)]
-#![cfg(all(unix, feature = "full"))]
+#![cfg(all(unix, feature = "full", not(miri)))]
 
 use std::os::unix::io::{AsRawFd, RawFd};
 use std::sync::{
@@ -655,6 +655,7 @@ fn send_oob_data<S: AsRawFd>(stream: &S, data: &[u8]) -> io::Result<usize> {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)]
 async fn clear_ready_matching_clears_ready() {
     use tokio::io::{Interest, Ready};
 
@@ -678,6 +679,7 @@ async fn clear_ready_matching_clears_ready() {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)]
 async fn clear_ready_matching_clears_ready_mut() {
     use tokio::io::{Interest, Ready};
 
@@ -702,6 +704,7 @@ async fn clear_ready_matching_clears_ready_mut() {
 
 #[tokio::test]
 #[cfg(target_os = "linux")]
+#[cfg_attr(miri, ignore)]
 async fn await_error_readiness_timestamping() {
     use std::net::{Ipv4Addr, SocketAddr};
 
@@ -758,6 +761,7 @@ fn configure_timestamping_socket(udp_socket: &std::net::UdpSocket) -> std::io::R
 
 #[tokio::test]
 #[cfg(target_os = "linux")]
+#[cfg_attr(miri, ignore)]
 async fn await_error_readiness_invalid_address() {
     use std::net::{Ipv4Addr, SocketAddr};
     use tokio::io::{Interest, Ready};

tokio/tests/io_copy_bidirectional.rs

@@ -1,5 +1,5 @@
 #![warn(rust_2018_idioms)]
-#![cfg(all(feature = "full", not(target_os = "wasi")))] // Wasi does not support bind()
+#![cfg(all(feature = "full", not(target_os = "wasi"), not(miri)))] // Wasi does not support bind()
 
 use std::time::Duration;
 use tokio::io::{self, copy_bidirectional, AsyncReadExt, AsyncWriteExt};

tokio/tests/io_driver.rs

@@ -1,6 +1,6 @@
 #![warn(rust_2018_idioms)]
 // Wasi does not support panic recovery or threading
-#![cfg(all(feature = "full", not(target_os = "wasi")))]
+#![cfg(all(feature = "full", not(target_os = "wasi"), not(miri)))]
 
 use tokio::net::TcpListener;
 use tokio::runtime;

tokio/tests/io_driver_drop.rs

@@ -1,5 +1,5 @@
 #![warn(rust_2018_idioms)]
-#![cfg(all(feature = "full", not(target_os = "wasi")))] // Wasi does not support bind
+#![cfg(all(feature = "full", not(target_os = "wasi"), not(miri)))] // Wasi does not support bind
 
 use tokio::net::TcpListener;
 use tokio::runtime;

tokio/tests/io_read_to_end.rs

@@ -79,6 +79,7 @@ async fn read_to_end_uninit() {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)] // too slow with miri
 async fn read_to_end_doesnt_grow_with_capacity() {
     let arr: Vec<u8> = (0..100).collect();
 

tokio/tests/io_repeat.rs

@@ -1,5 +1,5 @@
 #![warn(rust_2018_idioms)]
-#![cfg(feature = "full")]
+#![cfg(all(feature = "full", not(miri)))]
 
 use tokio::io::AsyncReadExt;
 

tokio/tests/net_bind_resource.rs

@@ -1,5 +1,5 @@
 #![warn(rust_2018_idioms)]
-#![cfg(all(feature = "full", not(target_os = "wasi")))] // Wasi doesn't support panic recovery or bind
+#![cfg(all(feature = "full", not(target_os = "wasi"), not(miri)))] // Wasi doesn't support panic recovery or bind
 
 use tokio::net::TcpListener;
 

tokio/tests/net_lookup_host.rs

@@ -23,6 +23,7 @@ async fn lookup_str_socket_addr() {
 }
 
 #[tokio::test]
+#[cfg_attr(miri, ignore)]
 async fn resolve_dns() -> io::Result<()> {
     let mut hosts = net::lookup_host("localhost:3000").await?;
     let host = hosts.next().unwrap();

tokio/tests/net_panic.rs

@@ -1,5 +1,5 @@
 #![warn(rust_2018_idioms)]
-#![cfg(all(feature = "full", not(target_os = "wasi")))]
+#![cfg(all(feature = "full", not(target_os = "wasi"), not(miri)))]
 #![cfg(panic = "unwind")]
 
 use std::error::Error;

tokio/tests/net_unix_pipe.rs

@@ -1,5 +1,6 @@
 #![cfg(feature = "full")]
 #![cfg(unix)]
+#![cfg(not(miri))]
 
 use tokio::io::{AsyncReadExt, AsyncWriteExt, Interest};
 use tokio::net::unix::pipe;

tokio/tests/no_rt.rs

@@ -1,4 +1,4 @@
-#![cfg(all(feature = "full", not(target_os = "wasi")))] // Wasi does not support panic recovery
+#![cfg(all(feature = "full", not(target_os = "wasi"), not(miri)))] // Wasi does not support panic recovery
 
 use tokio::net::TcpStream;
 use tokio::sync::oneshot;

tokio/tests/process_arg0.rs

@@ -1,5 +1,5 @@
 #![warn(rust_2018_idioms)]
-#![cfg(all(feature = "full", unix))]
+#![cfg(all(feature = "full", unix, not(miri)))]
 
 use tokio::process::Command;
 

tokio/tests/process_change_of_runtime.rs

@@ -3,7 +3,7 @@
 // This tests test the behavior of `process::Command::spawn` when it is used
 // outside runtime, and when `process::Child::wait ` is used in a different
 // runtime from which `process::Command::spawn` is used.
-#![cfg(all(unix, not(target_os = "freebsd")))]
+#![cfg(all(unix, not(target_os = "freebsd"), not(miri)))]
 
 use std::process::Stdio;
 use tokio::{process::Command, runtime::Runtime};

tokio/tests/process_issue_2174.rs

@@ -7,7 +7,7 @@
 // It is expected that `EVFILT_WRITE` would be returned with either the
 // `EV_EOF` or `EV_ERROR` flag set. If either flag is set a write would be
 // attempted, but that does not seem to occur.
-#![cfg(all(unix, not(target_os = "freebsd")))]
+#![cfg(all(unix, not(target_os = "freebsd"), not(miri)))]
 
 use std::process::Stdio;
 use std::time::Duration;

tokio/tests/process_issue_42.rs

@@ -1,6 +1,7 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
 #![cfg(unix)]
+#![cfg(not(miri))]
 
 use futures::future::join_all;
 use std::process::Stdio;

tokio/tests/process_kill_on_drop.rs

@@ -1,4 +1,4 @@
-#![cfg(all(unix, feature = "process"))]
+#![cfg(all(unix, feature = "process", not(miri)))]
 #![warn(rust_2018_idioms)]
 
 use std::io::ErrorKind;

tokio/tests/process_raw_handle.rs

@@ -1,6 +1,7 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
 #![cfg(windows)]
+#![cfg(not(miri))]
 
 use tokio::process::Command;
 use windows_sys::Win32::System::Threading::GetProcessId;

tokio/tests/process_smoke.rs

@@ -1,5 +1,5 @@
 #![warn(rust_2018_idioms)]
-#![cfg(all(feature = "full", not(target_os = "wasi")))] // Wasi cannot run system commands
+#![cfg(all(feature = "full", not(target_os = "wasi"), not(miri)))] // Wasi cannot run system commands
 
 use tokio::process::Command;
 use tokio_test::assert_ok;

tokio/tests/rt_basic.rs

@@ -1,6 +1,7 @@
 #![allow(unknown_lints, unexpected_cfgs)]
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
+#![cfg(not(miri))] // Possible bug on Miri.
 
 use tokio::runtime::Runtime;
 use tokio::sync::oneshot;

tokio/tests/rt_common.rs

@@ -2,6 +2,7 @@
 #![allow(clippy::needless_range_loop)]
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
+#![cfg(not(miri))]
 
 // Tests to run on both current-thread & multi-thread runtime variants.
 

tokio/tests/rt_handle_block_on.rs

@@ -1,5 +1,5 @@
 #![warn(rust_2018_idioms)]
-#![cfg(feature = "full")]
+#![cfg(all(feature = "full", not(miri)))]
 
 // All io tests that deal with shutdown is currently ignored because there are known bugs in with
 // shutting down the io driver while concurrently registering new resources. See

tokio/tests/rt_threaded.rs

@@ -1,6 +1,7 @@
 #![allow(unknown_lints, unexpected_cfgs)]
 #![warn(rust_2018_idioms)]
-#![cfg(all(feature = "full", not(target_os = "wasi")))]
+// Too slow on miri.
+#![cfg(all(feature = "full", not(target_os = "wasi"), not(miri)))]
 
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::{TcpListener, TcpStream};
@@ -321,6 +322,8 @@ fn start_stop_callbacks_called() {
 }
 
 #[test]
+// too slow on miri
+#[cfg_attr(miri, ignore)]
 fn blocking() {
     // used for notifying the main thread
     const NUM: usize = 1_000;

tokio/tests/signal_ctrl_c.rs

@@ -1,6 +1,7 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
 #![cfg(unix)]
+#![cfg(not(miri))] // No `sigaction` on Miri
 
 mod support {
     pub mod signal;

tokio/tests/signal_drop_recv.rs

@@ -1,6 +1,7 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
 #![cfg(unix)]
+#![cfg(not(miri))]
 
 mod support {
     pub mod signal;

tokio/tests/signal_drop_rt.rs

@@ -1,6 +1,7 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
 #![cfg(unix)]
+#![cfg(not(miri))]
 
 mod support {
     pub mod signal;

tokio/tests/signal_drop_signal.rs

@@ -1,6 +1,7 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
 #![cfg(unix)]
+#![cfg(not(miri))]
 
 mod support {
     pub mod signal;