chore: prepare Tokio 1.7.3

Tokio 1.7.0 introduced a change intended to eagerly shutdown newly
spawned tasks if the runtime is in the process of shutting down.
However, it introduced a bug where already spawned tasks could be
shutdown too early, resulting in the potential introduction of deadlocks
if tasks acquired mutexes in drop handlers.

Fixes #3869

.clippy.toml

@@ -0,0 +1 @@
+msrv = "1.45"

.github/FUNDING.yml

@@ -0,0 +1,3 @@
+# These are supported funding model platforms
+
+github: [tokio-rs]

.github/workflows/ci.yml

@@ -238,7 +238,7 @@ jobs:
           cargo hack --remove-dev-deps --workspace
           # Update Cargo.lock to minimal version dependencies.
           cargo update -Z minimal-versions
-          cargo check --all-features
+          cargo hack check --all-features --ignore-private
 
   fmt:
     name: fmt
@@ -265,7 +265,7 @@ jobs:
     steps:
       - uses: actions/checkout@v2
       - name: Install Rust
-        run: rustup update ${{ env.minrust }} && rustup default ${{ env.minrust }}
+        run: rustup update 1.52.1 && rustup default 1.52.1
       - name: Install clippy
         run: rustup component add clippy
 

README.md

@@ -56,7 +56,7 @@ Make sure you activated the full features of the tokio crate on Cargo.toml:
 
 ```toml
 [dependencies]
-tokio = { version = "1.5.0", features = ["full"] }
+tokio = { version = "1.7.0", features = ["full"] }
 ```
 Then, on your main.rs:
 
@@ -66,7 +66,7 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let mut listener = TcpListener::bind("127.0.0.1:8080").await?;
+    let listener = TcpListener::bind("127.0.0.1:8080").await?;
 
     loop {
         let (mut socket, _) = listener.accept().await?;

examples/Cargo.toml

@@ -22,7 +22,10 @@ serde_json = "1.0"
 httparse = "1.0"
 time = "0.1"
 once_cell = "1.5.2"
+rand = "0.8.3"
 
+[target.'cfg(windows)'.dev-dependencies.winapi]
+version = "0.3.8"
 
 [[example]]
 name = "chat"
@@ -76,3 +79,11 @@ path = "custom-executor.rs"
 [[example]]
 name = "custom-executor-tokio-context"
 path = "custom-executor-tokio-context.rs"
+
+[[example]]
+name = "named-pipe"
+path = "named-pipe.rs"
+
+[[example]]
+name = "named-pipe-multi-client"
+path = "named-pipe-multi-client.rs"

examples/named-pipe-multi-client.rs

@@ -0,0 +1,98 @@
+use std::io;
+
+#[cfg(windows)]
+async fn windows_main() -> io::Result<()> {
+    use std::time::Duration;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    use tokio::net::windows::named_pipe::{ClientOptions, ServerOptions};
+    use tokio::time;
+    use winapi::shared::winerror;
+
+    const PIPE_NAME: &str = r"\\.\pipe\named-pipe-multi-client";
+    const N: usize = 10;
+
+    // The first server needs to be constructed early so that clients can
+    // be correctly connected. Otherwise a waiting client will error.
+    //
+    // Here we also make use of `first_pipe_instance`, which will ensure
+    // that there are no other servers up and running already.
+    let mut server = ServerOptions::new()
+        .first_pipe_instance(true)
+        .create(PIPE_NAME)?;
+
+    let server = tokio::spawn(async move {
+        // Artificial workload.
+        time::sleep(Duration::from_secs(1)).await;
+
+        for _ in 0..N {
+            // Wait for client to connect.
+            server.connect().await?;
+            let mut inner = server;
+
+            // Construct the next server to be connected before sending the one
+            // we already have of onto a task. This ensures that the server
+            // isn't closed (after it's done in the task) before a new one is
+            // available. Otherwise the client might error with
+            // `io::ErrorKind::NotFound`.
+            server = ServerOptions::new().create(PIPE_NAME)?;
+
+            let _ = tokio::spawn(async move {
+                let mut buf = vec![0u8; 4];
+                inner.read_exact(&mut buf).await?;
+                inner.write_all(b"pong").await?;
+                Ok::<_, io::Error>(())
+            });
+        }
+
+        Ok::<_, io::Error>(())
+    });
+
+    let mut clients = Vec::new();
+
+    for _ in 0..N {
+        clients.push(tokio::spawn(async move {
+            // This showcases a generic connect loop.
+            //
+            // We immediately try to create a client, if it's not found or
+            // the pipe is busy we use the specialized wait function on the
+            // client builder.
+            let mut client = loop {
+                match ClientOptions::new().open(PIPE_NAME) {
+                    Ok(client) => break client,
+                    Err(e) if e.raw_os_error() == Some(winerror::ERROR_PIPE_BUSY as i32) => (),
+                    Err(e) => return Err(e),
+                }
+
+                time::sleep(Duration::from_millis(5)).await;
+            };
+
+            let mut buf = [0u8; 4];
+            client.write_all(b"ping").await?;
+            client.read_exact(&mut buf).await?;
+            Ok::<_, io::Error>(buf)
+        }));
+    }
+
+    for client in clients {
+        let result = client.await?;
+        assert_eq!(&result?[..], b"pong");
+    }
+
+    server.await??;
+    Ok(())
+}
+
+#[tokio::main]
+async fn main() -> io::Result<()> {
+    #[cfg(windows)]
+    {
+        windows_main().await?;
+    }
+
+    #[cfg(not(windows))]
+    {
+        println!("Named pipes are only supported on Windows!");
+    }
+
+    Ok(())
+}

examples/named-pipe.rs

@@ -0,0 +1,60 @@
+use std::io;
+
+#[cfg(windows)]
+async fn windows_main() -> io::Result<()> {
+    use tokio::io::AsyncWriteExt;
+    use tokio::io::{AsyncBufReadExt, BufReader};
+    use tokio::net::windows::named_pipe::{ClientOptions, ServerOptions};
+
+    const PIPE_NAME: &str = r"\\.\pipe\named-pipe-single-client";
+
+    let server = ServerOptions::new().create(PIPE_NAME)?;
+
+    let server = tokio::spawn(async move {
+        // Note: we wait for a client to connect.
+        server.connect().await?;
+
+        let mut server = BufReader::new(server);
+
+        let mut buf = String::new();
+        server.read_line(&mut buf).await?;
+        server.write_all(b"pong\n").await?;
+        Ok::<_, io::Error>(buf)
+    });
+
+    let client = tokio::spawn(async move {
+        // There's no need to use a connect loop here, since we know that the
+        // server is already up - `open` was called before spawning any of the
+        // tasks.
+        let client = ClientOptions::new().open(PIPE_NAME)?;
+
+        let mut client = BufReader::new(client);
+
+        let mut buf = String::new();
+        client.write_all(b"ping\n").await?;
+        client.read_line(&mut buf).await?;
+        Ok::<_, io::Error>(buf)
+    });
+
+    let (server, client) = tokio::try_join!(server, client)?;
+
+    assert_eq!(server?, "ping\n");
+    assert_eq!(client?, "pong\n");
+
+    Ok(())
+}
+
+#[tokio::main]
+async fn main() -> io::Result<()> {
+    #[cfg(windows)]
+    {
+        windows_main().await?;
+    }
+
+    #[cfg(not(windows))]
+    {
+        println!("Named pipes are only supported on Windows!");
+    }
+
+    Ok(())
+}

tokio-macros/CHANGELOG.md

@@ -1,3 +1,13 @@
+# 1.2.0 (May 14, 2021)
+
+- macros: forward input arguments in `#[tokio::test]` ([#3691])
+- macros: improve diagnostics on type mismatch ([#3766])
+- macros: various error message improvements ([#3677])
+
+[#3677]: https://github.com/tokio-rs/tokio/pull/3677
+[#3691]: https://github.com/tokio-rs/tokio/pull/3691
+[#3766]: https://github.com/tokio-rs/tokio/pull/3766
+
 # 1.1.0 (February 5, 2021)
 
 - add `start_paused` option to macros ([#3492])

tokio-macros/Cargo.toml

@@ -6,13 +6,13 @@ name = "tokio-macros"
 #   - Cargo.toml
 # - Update CHANGELOG.md.
 # - Create "tokio-macros-1.0.x" git tag.
-version = "1.1.0"
+version = "1.2.0"
 edition = "2018"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
-documentation = "https://docs.rs/tokio-macros/1.1.0/tokio_macros"
+documentation = "https://docs.rs/tokio-macros/1.2.0/tokio_macros"
 description = """
 Tokio's proc macros.
 """

tokio-stream/CHANGELOG.md

@@ -1,3 +1,16 @@
+# 0.1.6 (May 14, 2021)
+
+### Added
+
+- stream: implement `Error` and `Display` for `BroadcastStreamRecvError` ([#3745])
+
+### Fixed
+
+- stream: avoid yielding in `AllFuture` and `AnyFuture` ([#3625])
+
+[#3745]: https://github.com/tokio-rs/tokio/pull/3745
+[#3625]: https://github.com/tokio-rs/tokio/pull/3625
+
 # 0.1.5 (March 20, 2021)
 
 ### Fixed

tokio-stream/Cargo.toml

@@ -6,13 +6,13 @@ name = "tokio-stream"
 #   - Cargo.toml
 # - Update CHANGELOG.md.
 # - Create "tokio-stream-0.1.x" git tag.
-version = "0.1.5"
+version = "0.1.6"
 edition = "2018"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
-documentation = "https://docs.rs/tokio-stream/0.1.5/tokio_stream"
+documentation = "https://docs.rs/tokio-stream/0.1.6/tokio_stream"
 description = """
 Utilities to work with `Stream` and `tokio`.
 """

tokio-stream/tests/async_send_sync.rs

@@ -1,3 +1,5 @@
+#![allow(clippy::diverging_sub_expression)]
+
 use std::rc::Rc;
 
 #[allow(dead_code)]

tokio-stream/tests/stream_chain.rs

@@ -89,12 +89,12 @@ fn size_overflow() {
         }
 
         fn size_hint(&self) -> (usize, Option<usize>) {
-            (usize::max_value(), Some(usize::max_value()))
+            (usize::MAX, Some(usize::MAX))
         }
     }
 
     let m1 = Monster;
     let m2 = Monster;
     let m = m1.chain(m2);
-    assert_eq!(m.size_hint(), (usize::max_value(), None));
+    assert_eq!(m.size_hint(), (usize::MAX, None));
 }

tokio-stream/tests/stream_merge.rs

@@ -72,12 +72,12 @@ fn size_overflow() {
         }
 
         fn size_hint(&self) -> (usize, Option<usize>) {
-            (usize::max_value(), Some(usize::max_value()))
+            (usize::MAX, Some(usize::MAX))
         }
     }
 
     let m1 = Monster;
     let m2 = Monster;
     let m = m1.merge(m2);
-    assert_eq!(m.size_hint(), (usize::max_value(), None));
+    assert_eq!(m.size_hint(), (usize::MAX, None));
 }

tokio-test/CHANGELOG.md

@@ -1,3 +1,9 @@
+# 0.4.2 (May 14, 2021)
+
+- test: add `assert_elapsed!` macro ([#3728])
+
+[#3728]: https://github.com/tokio-rs/tokio/pull/3728
+
 # 0.4.1 (March 10, 2021)
 
 - Fix `io::Mock` to be `Send` and `Sync` ([#3594])

tokio-test/Cargo.toml

@@ -6,13 +6,13 @@ name = "tokio-test"
 #   - Cargo.toml
 # - Update CHANGELOG.md.
 # - Create "tokio-test-0.4.x" git tag.
-version = "0.4.1"
+version = "0.4.2"
 edition = "2018"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
-documentation = "https://docs.rs/tokio-test/0.4.1/tokio_test"
+documentation = "https://docs.rs/tokio-test/0.4.2/tokio_test"
 description = """
 Testing utilities for Tokio- and futures-based code
 """

tokio-util/CHANGELOG.md

@@ -1,3 +1,13 @@
+# 0.6.7 (May 14, 2021)
+
+### Added
+
+- udp: make `UdpFramed` take `Borrow<UdpSocket>` ([#3451])
+- compat: implement `AsRawFd`/`AsRawHandle` for `Compat<T>` ([#3765])
+
+[#3451]: https://github.com/tokio-rs/tokio/pull/3451
+[#3765]: https://github.com/tokio-rs/tokio/pull/3765
+
 # 0.6.6 (April 12, 2021)
 
 ### Added

tokio-util/Cargo.toml

@@ -6,13 +6,13 @@ name = "tokio-util"
 #   - Cargo.toml
 # - Update CHANGELOG.md.
 # - Create "tokio-util-0.6.x" git tag.
-version = "0.6.6"
+version = "0.6.7"
 edition = "2018"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
-documentation = "https://docs.rs/tokio-util/0.6.6/tokio_util"
+documentation = "https://docs.rs/tokio-util/0.6.7/tokio_util"
 description = """
 Additional utilities for working with Tokio.
 """

tokio-util/src/time/mod.rs

@@ -12,9 +12,9 @@ use std::time::Duration;
 
 mod wheel;
 
-#[doc(inline)]
 pub mod delay_queue;
 
+#[doc(inline)]
 pub use delay_queue::DelayQueue;
 
 // ===== Internal utils =====

tokio-util/tests/poll_semaphore.rs

@@ -6,9 +6,9 @@ use tokio_util::sync::PollSemaphore;
 
 type SemRet = Option<OwnedSemaphorePermit>;
 
-fn semaphore_poll<'a>(
-    sem: &'a mut PollSemaphore,
-) -> tokio_test::task::Spawn<impl Future<Output = SemRet> + 'a> {
+fn semaphore_poll(
+    sem: &mut PollSemaphore,
+) -> tokio_test::task::Spawn<impl Future<Output = SemRet> + '_> {
     let fut = futures::future::poll_fn(move |cx| sem.poll_acquire(cx));
     tokio_test::task::spawn(fut)
 }

tokio/CHANGELOG.md

@@ -1,3 +1,63 @@
+# 1.7.3 (July 19, 2021)
+
+Fixes a missed edge case from 1.7.2.
+
+### Fixed
+
+- runtime: drop canceled future on next poll (#3965)
+
+# 1.7.2 (July 6, 2021)
+
+Forward ports 1.5.1 fixes.
+
+### Fixed
+
+- runtime: remotely abort tasks on `JoinHandle::abort` ([#3934])
+
+[#3934]: https://github.com/tokio-rs/tokio/pull/3934
+
+# 1.7.1 (June 18, 2021)
+
+### Fixed
+
+- runtime: fix early task shutdown during runtime shutdown ([#3870])
+
+[#3870]: https://github.com/tokio-rs/tokio/pull/3870
+
+# 1.7.0 (June 15, 2021)
+
+### Added
+
+- net: add named pipes on windows ([#3760])
+- net: add `TcpSocket` from `std::net::TcpStream` conversion ([#3838])
+- sync: add `receiver_count` to `watch::Sender` ([#3729])
+- sync: export `sync::notify::Notified` future publicly ([#3840])
+- tracing: instrument task wakers ([#3836])
+
+### Fixed
+
+- macros: suppress `clippy::default_numeric_fallback` lint in generated code ([#3831])
+- runtime: immediately drop new tasks when runtime is shut down ([#3752])
+- sync: deprecate unused `mpsc::RecvError` type ([#3833])
+
+### Documented
+
+- io: clarify EOF condition for `AsyncReadExt::read_buf` ([#3850])
+- io: clarify limits on return values of `AsyncWrite::poll_write` ([#3820])
+- sync: add examples to Semaphore ([#3808])
+
+[#3729]: https://github.com/tokio-rs/tokio/pull/3729
+[#3752]: https://github.com/tokio-rs/tokio/pull/3752
+[#3760]: https://github.com/tokio-rs/tokio/pull/3760
+[#3808]: https://github.com/tokio-rs/tokio/pull/3808
+[#3820]: https://github.com/tokio-rs/tokio/pull/3820
+[#3831]: https://github.com/tokio-rs/tokio/pull/3831
+[#3833]: https://github.com/tokio-rs/tokio/pull/3833
+[#3836]: https://github.com/tokio-rs/tokio/pull/3836
+[#3838]: https://github.com/tokio-rs/tokio/pull/3838
+[#3840]: https://github.com/tokio-rs/tokio/pull/3840
+[#3850]: https://github.com/tokio-rs/tokio/pull/3850
+
 # 1.6.3 (July 6, 2021)
 
 Forward ports 1.5.1 fixes.

tokio/Cargo.toml

@@ -7,12 +7,12 @@ name = "tokio"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v1.0.x" git tag.
-version = "1.6.3"
+version = "1.7.3"
 edition = "2018"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 readme = "README.md"
-documentation = "https://docs.rs/tokio/1.6.3/tokio/"
+documentation = "https://docs.rs/tokio/1.7.3/tokio/"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
 description = """
@@ -54,6 +54,7 @@ net = [
   "mio/tcp",
   "mio/udp",
   "mio/uds",
+  "winapi/namedpipeapi",
 ]
 process = [
   "bytes",
@@ -115,6 +116,9 @@ version = "0.3.8"
 default-features = false
 optional = true
 
+[target.'cfg(windows)'.dev-dependencies.ntapi]
+version = "0.3.6"
+
 [dev-dependencies]
 tokio-test = { version = "0.4.0", path = "../tokio-test" }
 tokio-stream = { version = "0.1", path = "../tokio-stream" }
@@ -123,6 +127,7 @@ proptest = "1"
 rand = "0.8.0"
 tempfile = "3.1.0"
 async-stream = "0.3"
+socket2 = "0.4"
 
 [target.'cfg(loom)'.dev-dependencies]
 loom = { version = "0.5", features = ["futures", "checkpoint"] }

tokio/src/doc/mod.rs

@@ -0,0 +1,23 @@
+//! Types which are documented locally in the Tokio crate, but does not actually
+//! live here.
+//!
+//! **Note** this module is only visible on docs.rs, you cannot use it directly
+//! in your own code.
+
+/// The name of a type which is not defined here.
+///
+/// This is typically used as an alias for another type, like so:
+///
+/// ```rust,ignore
+/// /// See [some::other::location](https://example.com).
+/// type DEFINED_ELSEWHERE = crate::doc::NotDefinedHere;
+/// ```
+///
+/// This type is uninhabitable like the [`never` type] to ensure that no one
+/// will ever accidentally use it.
+///
+/// [`never` type]: https://doc.rust-lang.org/std/primitive.never.html
+pub enum NotDefinedHere {}
+
+pub mod os;
+pub mod winapi;

tokio/src/doc/os.rs

@@ -0,0 +1,26 @@
+//! See [std::os](https://doc.rust-lang.org/std/os/index.html).
+
+/// Platform-specific extensions to `std` for Windows.
+///
+/// See [std::os::windows](https://doc.rust-lang.org/std/os/windows/index.html).
+pub mod windows {
+    /// Windows-specific extensions to general I/O primitives.
+    ///
+    /// See [std::os::windows::io](https://doc.rust-lang.org/std/os/windows/io/index.html).
+    pub mod io {
+        /// See [std::os::windows::io::RawHandle](https://doc.rust-lang.org/std/os/windows/io/type.RawHandle.html)
+        pub type RawHandle = crate::doc::NotDefinedHere;
+
+        /// See [std::os::windows::io::AsRawHandle](https://doc.rust-lang.org/std/os/windows/io/trait.AsRawHandle.html)
+        pub trait AsRawHandle {
+            /// See [std::os::windows::io::FromRawHandle::from_raw_handle](https://doc.rust-lang.org/std/os/windows/io/trait.AsRawHandle.html#tymethod.as_raw_handle)
+            fn as_raw_handle(&self) -> RawHandle;
+        }
+
+        /// See [std::os::windows::io::FromRawHandle](https://doc.rust-lang.org/std/os/windows/io/trait.FromRawHandle.html)
+        pub trait FromRawHandle {
+            /// See [std::os::windows::io::FromRawHandle::from_raw_handle](https://doc.rust-lang.org/std/os/windows/io/trait.FromRawHandle.html#tymethod.from_raw_handle)
+            unsafe fn from_raw_handle(handle: RawHandle) -> Self;
+        }
+    }
+}

tokio/src/doc/winapi.rs

@@ -0,0 +1,66 @@
+//! See [winapi].
+//!
+//! [winapi]: https://docs.rs/winapi
+
+/// See [winapi::shared](https://docs.rs/winapi/*/winapi/shared/index.html).
+pub mod shared {
+    /// See [winapi::shared::winerror](https://docs.rs/winapi/*/winapi/shared/winerror/index.html).
+    #[allow(non_camel_case_types)]
+    pub mod winerror {
+        /// See [winapi::shared::winerror::ERROR_ACCESS_DENIED][winapi]
+        ///
+        /// [winapi]: https://docs.rs/winapi/*/winapi/shared/winerror/constant.ERROR_ACCESS_DENIED.html
+        pub type ERROR_ACCESS_DENIED = crate::doc::NotDefinedHere;
+
+        /// See [winapi::shared::winerror::ERROR_PIPE_BUSY][winapi]
+        ///
+        /// [winapi]: https://docs.rs/winapi/*/winapi/shared/winerror/constant.ERROR_PIPE_BUSY.html
+        pub type ERROR_PIPE_BUSY = crate::doc::NotDefinedHere;
+
+        /// See [winapi::shared::winerror::ERROR_MORE_DATA][winapi]
+        ///
+        /// [winapi]: https://docs.rs/winapi/*/winapi/shared/winerror/constant.ERROR_MORE_DATA.html
+        pub type ERROR_MORE_DATA = crate::doc::NotDefinedHere;
+    }
+}
+
+/// See [winapi::um](https://docs.rs/winapi/*/winapi/um/index.html).
+pub mod um {
+    /// See [winapi::um::winbase](https://docs.rs/winapi/*/winapi/um/winbase/index.html).
+    #[allow(non_camel_case_types)]
+    pub mod winbase {
+        /// See [winapi::um::winbase::PIPE_TYPE_MESSAGE][winapi]
+        ///
+        /// [winapi]: https://docs.rs/winapi/*/winapi/um/winbase/constant.PIPE_TYPE_MESSAGE.html
+        pub type PIPE_TYPE_MESSAGE = crate::doc::NotDefinedHere;
+
+        /// See [winapi::um::winbase::PIPE_TYPE_BYTE][winapi]
+        ///
+        /// [winapi]: https://docs.rs/winapi/*/winapi/um/winbase/constant.PIPE_TYPE_BYTE.html
+        pub type PIPE_TYPE_BYTE = crate::doc::NotDefinedHere;
+
+        /// See [winapi::um::winbase::PIPE_CLIENT_END][winapi]
+        ///
+        /// [winapi]: https://docs.rs/winapi/*/winapi/um/winbase/constant.PIPE_CLIENT_END.html
+        pub type PIPE_CLIENT_END = crate::doc::NotDefinedHere;
+
+        /// See [winapi::um::winbase::PIPE_SERVER_END][winapi]
+        ///
+        /// [winapi]: https://docs.rs/winapi/*/winapi/um/winbase/constant.PIPE_SERVER_END.html
+        pub type PIPE_SERVER_END = crate::doc::NotDefinedHere;
+
+        /// See [winapi::um::winbase::SECURITY_IDENTIFICATION][winapi]
+        ///
+        /// [winapi]: https://docs.rs/winapi/*/winapi/um/winbase/constant.SECURITY_IDENTIFICATION.html
+        pub type SECURITY_IDENTIFICATION = crate::doc::NotDefinedHere;
+    }
+
+    /// See [winapi::um::minwinbase](https://docs.rs/winapi/*/winapi/um/minwinbase/index.html).
+    #[allow(non_camel_case_types)]
+    pub mod minwinbase {
+        /// See [winapi::um::minwinbase::SECURITY_ATTRIBUTES][winapi]
+        ///
+        /// [winapi]: https://docs.rs/winapi/*/winapi/um/minwinbase/constant.SECURITY_ATTRIBUTES.html
+        pub type SECURITY_ATTRIBUTES = crate::doc::NotDefinedHere;
+    }
+}

tokio/src/future/mod.rs

@@ -22,3 +22,14 @@ cfg_sync! {
     mod block_on;
     pub(crate) use block_on::block_on;
 }
+
+cfg_trace! {
+    mod trace;
+    pub(crate) use trace::InstrumentedFuture as Future;
+}
+
+cfg_not_trace! {
+    cfg_rt! {
+        pub(crate) use std::future::Future;
+    }
+}

tokio/src/future/trace.rs

@@ -0,0 +1,11 @@
+use std::future::Future;
+
+pub(crate) trait InstrumentedFuture: Future {
+    fn id(&self) -> Option<tracing::Id>;
+}
+
+impl<F: Future> InstrumentedFuture for tracing::instrument::Instrumented<F> {
+    fn id(&self) -> Option<tracing::Id> {
+        self.span().id()
+    }
+}

tokio/src/io/async_write.rs

@@ -45,7 +45,11 @@ use std::task::{Context, Poll};
 pub trait AsyncWrite {
     /// Attempt to write bytes from `buf` into the object.
     ///
-    /// On success, returns `Poll::Ready(Ok(num_bytes_written))`.
+    /// On success, returns `Poll::Ready(Ok(num_bytes_written))`. If successful,
+    /// then it must be guaranteed that `n <= buf.len()`. A return value of `0`
+    /// typically means that the underlying object is no longer able to accept
+    /// bytes and will likely not be able to in the future as well, or that the
+    /// buffer provided is empty.
     ///
     /// If the object is not ready for writing, the method returns
     /// `Poll::Pending` and arranges for the current task (via

tokio/src/io/util/async_read_ext.rs

@@ -108,6 +108,8 @@ cfg_io_util! {
         /// This function does not provide any guarantees about whether it
         /// completes immediately or asynchronously
         ///
+        /// # Return
+        ///
         /// If the return value of this method is `Ok(n)`, then it must be
         /// guaranteed that `0 <= n <= buf.len()`. A nonzero `n` value indicates
         /// that the buffer `buf` has been filled in with `n` bytes of data from
@@ -180,9 +182,14 @@ cfg_io_util! {
         ///
         /// # Return
         ///
-        /// On a successful read, the number of read bytes is returned. If the
-        /// supplied buffer is not empty and the function returns `Ok(0)` then
-        /// the source has reached an "end-of-file" event.
+        /// A nonzero `n` value indicates that the buffer `buf` has been filled
+        /// in with `n` bytes of data from this source. If `n` is `0`, then it
+        /// can indicate one of two scenarios:
+        ///
+        /// 1. This reader has reached its "end of file" and will likely no longer
+        ///    be able to produce bytes. Note that this does not mean that the
+        ///    reader will *always* no longer be able to produce bytes.
+        /// 2. The buffer specified had a remaining capacity of zero.
         ///
         /// # Errors
         ///
@@ -579,7 +586,7 @@ cfg_io_util! {
             /// async fn main() -> io::Result<()> {
             ///     let mut reader = Cursor::new(vec![0x80, 0, 0, 0, 0, 0, 0, 0]);
             ///
-            ///     assert_eq!(i64::min_value(), reader.read_i64().await?);
+            ///     assert_eq!(i64::MIN, reader.read_i64().await?);
             ///     Ok(())
             /// }
             /// ```
@@ -659,7 +666,7 @@ cfg_io_util! {
             ///         0, 0, 0, 0, 0, 0, 0, 0
             ///     ]);
             ///
-            ///     assert_eq!(i128::min_value(), reader.read_i128().await?);
+            ///     assert_eq!(i128::MIN, reader.read_i128().await?);
             ///     Ok(())
             /// }
             /// ```

tokio/src/io/util/async_write_ext.rs

@@ -621,8 +621,8 @@ cfg_io_util! {
             /// async fn main() -> io::Result<()> {
             ///     let mut writer = Vec::new();
             ///
-            ///     writer.write_i64(i64::min_value()).await?;
-            ///     writer.write_i64(i64::max_value()).await?;
+            ///     writer.write_i64(i64::MIN).await?;
+            ///     writer.write_i64(i64::MAX).await?;
             ///
             ///     assert_eq!(writer, b"\x80\x00\x00\x00\x00\x00\x00\x00\x7f\xff\xff\xff\xff\xff\xff\xff");
             ///     Ok(())
@@ -699,7 +699,7 @@ cfg_io_util! {
             /// async fn main() -> io::Result<()> {
             ///     let mut writer = Vec::new();
             ///
-            ///     writer.write_i128(i128::min_value()).await?;
+            ///     writer.write_i128(i128::MIN).await?;
             ///
             ///     assert_eq!(writer, vec![
             ///         0x80, 0, 0, 0, 0, 0, 0, 0,
@@ -930,8 +930,8 @@ cfg_io_util! {
             /// async fn main() -> io::Result<()> {
             ///     let mut writer = Vec::new();
             ///
-            ///     writer.write_i64_le(i64::min_value()).await?;
-            ///     writer.write_i64_le(i64::max_value()).await?;
+            ///     writer.write_i64_le(i64::MIN).await?;
+            ///     writer.write_i64_le(i64::MAX).await?;
             ///
             ///     assert_eq!(writer, b"\x00\x00\x00\x00\x00\x00\x00\x80\xff\xff\xff\xff\xff\xff\xff\x7f");
             ///     Ok(())
@@ -1008,7 +1008,7 @@ cfg_io_util! {
             /// async fn main() -> io::Result<()> {
             ///     let mut writer = Vec::new();
             ///
-            ///     writer.write_i128_le(i128::min_value()).await?;
+            ///     writer.write_i128_le(i128::MIN).await?;
             ///
             ///     assert_eq!(writer, vec![
             ///          0, 0, 0, 0, 0, 0, 0,

tokio/src/io/util/buf_reader.rs

@@ -198,7 +198,7 @@ impl<R: AsyncRead + AsyncSeek> AsyncSeek for BufReader<R> {
                 // it should be safe to assume that remainder fits within an i64 as the alternative
                 // means we managed to allocate 8 exbibytes and that's absurd.
                 // But it's not out of the realm of possibility for some weird underlying reader to
-                // support seeking by i64::min_value() so we need to handle underflow when subtracting
+                // support seeking by i64::MIN so we need to handle underflow when subtracting
                 // remainder.
                 if let Some(offset) = n.checked_sub(remainder) {
                     self.as_mut()

tokio/src/io/util/buf_stream.rs

@@ -1,8 +1,8 @@
 use crate::io::util::{BufReader, BufWriter};
-use crate::io::{AsyncBufRead, AsyncRead, AsyncWrite, ReadBuf};
+use crate::io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite, ReadBuf};
 
 use pin_project_lite::pin_project;
-use std::io;
+use std::io::{self, SeekFrom};
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
@@ -146,6 +146,34 @@ impl<RW: AsyncRead + AsyncWrite> AsyncRead for BufStream<RW> {
     }
 }
 
+/// Seek to an offset, in bytes, in the underlying stream.
+///
+/// The position used for seeking with `SeekFrom::Current(_)` is the
+/// position the underlying stream would be at if the `BufStream` had no
+/// internal buffer.
+///
+/// Seeking always discards the internal buffer, even if the seek position
+/// would otherwise fall within it. This guarantees that calling
+/// `.into_inner()` immediately after a seek yields the underlying reader
+/// at the same position.
+///
+/// See [`AsyncSeek`] for more details.
+///
+/// Note: In the edge case where you're seeking with `SeekFrom::Current(n)`
+/// where `n` minus the internal buffer length overflows an `i64`, two
+/// seeks will be performed instead of one. If the second seek returns
+/// `Err`, the underlying reader will be left at the same position it would
+/// have if you called `seek` with `SeekFrom::Current(0)`.
+impl<RW: AsyncRead + AsyncWrite + AsyncSeek> AsyncSeek for BufStream<RW> {
+    fn start_seek(self: Pin<&mut Self>, position: SeekFrom) -> io::Result<()> {
+        self.project().inner.start_seek(position)
+    }
+
+    fn poll_complete(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>> {
+        self.project().inner.poll_complete(cx)
+    }
+}
+
 impl<RW: AsyncRead + AsyncWrite> AsyncBufRead for BufStream<RW> {
     fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
         self.project().inner.poll_fill_buf(cx)

tokio/src/lib.rs

@@ -9,6 +9,7 @@
     rust_2018_idioms,
     unreachable_pub
 )]
+#![deny(unused_must_use)]
 #![cfg_attr(docsrs, deny(broken_intra_doc_links))]
 #![doc(test(
     no_crate_inject,
@@ -442,6 +443,28 @@ mod util;
 /// ```
 pub mod stream {}
 
+// local re-exports of platform specific things, allowing for decent
+// documentation to be shimmed in on docs.rs
+
+#[cfg(docsrs)]
+pub mod doc;
+
+#[cfg(docsrs)]
+#[allow(unused)]
+pub(crate) use self::doc::os;
+
+#[cfg(not(docsrs))]
+#[allow(unused)]
+pub(crate) use std::os;
+
+#[cfg(docsrs)]
+#[allow(unused)]
+pub(crate) use self::doc::winapi;
+
+#[cfg(all(not(docsrs), windows, feature = "net"))]
+#[allow(unused)]
+pub(crate) use ::winapi;
+
 cfg_macros! {
     /// Implementation detail of the `select!` macro. This macro is **not**
     /// intended to be used as part of the public API and is permitted to
@@ -453,15 +476,20 @@ cfg_macros! {
         #[cfg(feature = "rt-multi-thread")]
         #[cfg(not(test))] // Work around for rust-lang/rust#62127
         #[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
+        #[doc(inline)]
         pub use tokio_macros::main;
 
         #[cfg(feature = "rt-multi-thread")]
         #[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
+        #[doc(inline)]
         pub use tokio_macros::test;
 
         cfg_not_rt_multi_thread! {
             #[cfg(not(test))] // Work around for rust-lang/rust#62127
+            #[doc(inline)]
             pub use tokio_macros::main_rt as main;
+
+            #[doc(inline)]
             pub use tokio_macros::test_rt as test;
         }
     }
@@ -469,7 +497,10 @@ cfg_macros! {
     // Always fail if rt is not enabled.
     cfg_not_rt! {
         #[cfg(not(test))]
+        #[doc(inline)]
         pub use tokio_macros::main_fail as main;
+
+        #[doc(inline)]
         pub use tokio_macros::test_fail as test;
     }
 }

tokio/src/macros/cfg.rs

@@ -157,7 +157,6 @@ macro_rules! cfg_macros {
         $(
             #[cfg(feature = "macros")]
             #[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
-            #[doc(inline)]
             $item
         )*
     }
@@ -183,6 +182,16 @@ macro_rules! cfg_net_unix {
     }
 }
 
+macro_rules! cfg_net_windows {
+    ($($item:item)*) => {
+        $(
+            #[cfg(all(any(docsrs, windows), feature = "net"))]
+            #[cfg_attr(docsrs, doc(cfg(all(windows, feature = "net"))))]
+            $item
+        )*
+    }
+}
+
 macro_rules! cfg_process {
     ($($item:item)*) => {
         $(

tokio/src/macros/select.rs

@@ -398,7 +398,7 @@ macro_rules! select {
         // set the appropriate bit in `disabled`.
         $(
             if !$c {
-                let mask = 1 << $crate::count!( $($skip)* );
+                let mask: util::Mask = 1 << $crate::count!( $($skip)* );
                 disabled |= mask;
             }
         )*

tokio/src/net/mod.rs

@@ -46,3 +46,7 @@ cfg_net_unix! {
     pub use unix::listener::UnixListener;
     pub use unix::stream::UnixStream;
 }
+
+cfg_net_windows! {
+    pub mod windows;
+}

tokio/src/net/tcp/socket.rs

@@ -482,6 +482,48 @@ impl TcpSocket {
         let mio = self.inner.listen(backlog)?;
         TcpListener::new(mio)
     }
+
+    /// Converts a [`std::net::TcpStream`] into a `TcpSocket`. The provided
+    /// socket must not have been connected prior to calling this function. This
+    /// function is typically used together with crates such as [`socket2`] to
+    /// configure socket options that are not available on `TcpSocket`.
+    ///
+    /// [`std::net::TcpStream`]: struct@std::net::TcpStream
+    /// [`socket2`]: https://docs.rs/socket2/
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::net::TcpSocket;
+    /// use socket2::{Domain, Socket, Type};
+    ///
+    /// #[tokio::main]
+    /// async fn main() -> std::io::Result<()> {
+    ///     
+    ///     let socket2_socket = Socket::new(Domain::IPV4, Type::STREAM, None)?;
+    ///
+    ///     let socket = TcpSocket::from_std_stream(socket2_socket.into());
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub fn from_std_stream(std_stream: std::net::TcpStream) -> TcpSocket {
+        #[cfg(unix)]
+        {
+            use std::os::unix::io::{FromRawFd, IntoRawFd};
+
+            let raw_fd = std_stream.into_raw_fd();
+            unsafe { TcpSocket::from_raw_fd(raw_fd) }
+        }
+
+        #[cfg(windows)]
+        {
+            use std::os::windows::io::{FromRawSocket, IntoRawSocket};
+
+            let raw_socket = std_stream.into_raw_socket();
+            unsafe { TcpSocket::from_raw_socket(raw_socket) }
+        }
+    }
 }
 
 impl fmt::Debug for TcpSocket {

tokio/src/net/unix/ucred.rs

@@ -73,7 +73,7 @@ pub(crate) mod impl_linux {
 
             // These paranoid checks should be optimized-out
             assert!(mem::size_of::<u32>() <= mem::size_of::<usize>());
-            assert!(ucred_size <= u32::max_value() as usize);
+            assert!(ucred_size <= u32::MAX as usize);
 
             let mut ucred_size = ucred_size as socklen_t;
 

tokio/src/net/windows/mod.rs

@@ -0,0 +1,3 @@
+//! Windows specific network types.
+
+pub mod named_pipe;

tokio/src/runtime/basic_scheduler.rs

@@ -84,13 +84,13 @@ unsafe impl Send for Entry {}
 
 /// Scheduler state shared between threads.
 struct Shared {
-    /// Remote run queue
-    queue: Mutex<VecDeque<Entry>>,
+    /// Remote run queue. None if the `Runtime` has been dropped.
+    queue: Mutex<Option<VecDeque<Entry>>>,
 
-    /// Unpark the blocked thread
+    /// Unpark the blocked thread.
     unpark: Box<dyn Unpark>,
 
-    // indicates whether the blocked on thread was woken
+    /// Indicates whether the blocked on thread was woken.
     woken: AtomicBool,
 }
 
@@ -124,7 +124,7 @@ impl<P: Park> BasicScheduler<P> {
 
         let spawner = Spawner {
             shared: Arc::new(Shared {
-                queue: Mutex::new(VecDeque::with_capacity(INITIAL_CAPACITY)),
+                queue: Mutex::new(Some(VecDeque::with_capacity(INITIAL_CAPACITY))),
                 unpark: unpark as Box<dyn Unpark>,
                 woken: AtomicBool::new(false),
             }),
@@ -351,18 +351,29 @@ impl<P: Park> Drop for BasicScheduler<P> {
                 task.shutdown();
             }
 
-            // Drain remote queue
-            for entry in scheduler.spawner.shared.queue.lock().drain(..) {
-                match entry {
-                    Entry::Schedule(task) => {
-                        task.shutdown();
-                    }
-                    Entry::Release(..) => {
-                        // Do nothing, each entry in the linked list was *just*
-                        // dropped by the scheduler above.
+            // Drain remote queue and set it to None
+            let mut remote_queue = scheduler.spawner.shared.queue.lock();
+
+            // Using `Option::take` to replace the shared queue with `None`.
+            if let Some(remote_queue) = remote_queue.take() {
+                for entry in remote_queue {
+                    match entry {
+                        Entry::Schedule(task) => {
+                            task.shutdown();
+                        }
+                        Entry::Release(..) => {
+                            // Do nothing, each entry in the linked list was *just*
+                            // dropped by the scheduler above.
+                        }
                     }
                 }
             }
+            // By dropping the mutex lock after the full duration of the above loop,
+            // any thread that sees the queue in the `None` state is guaranteed that
+            // the runtime has fully shut down.
+            //
+            // The assert below is unrelated to this mutex.
+            drop(remote_queue);
 
             assert!(context.tasks.borrow().owned.is_empty());
         });
@@ -381,7 +392,7 @@ impl Spawner {
     /// Spawns a future onto the thread pool
     pub(crate) fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
     where
-        F: Future + Send + 'static,
+        F: crate::future::Future + Send + 'static,
         F::Output: Send + 'static,
     {
         let (task, handle) = task::joinable(future);
@@ -390,7 +401,10 @@ impl Spawner {
     }
 
     fn pop(&self) -> Option<Entry> {
-        self.shared.queue.lock().pop_front()
+        match self.shared.queue.lock().as_mut() {
+            Some(queue) => queue.pop_front(),
+            None => None,
+        }
     }
 
     fn waker_ref(&self) -> WakerRef<'_> {
@@ -429,7 +443,19 @@ impl Schedule for Arc<Shared> {
                 // safety: the task is inserted in the list in `bind`.
                 unsafe { cx.tasks.borrow_mut().owned.remove(ptr) }
             } else {
-                self.queue.lock().push_back(Entry::Release(ptr));
+                // By sending an `Entry::Release` to the runtime, we ask the
+                // runtime to remove this task from the linked list in
+                // `Tasks::owned`.
+                //
+                // If the queue is `None`, then the task was already removed
+                // from that list in the destructor of `BasicScheduler`. We do
+                // not do anything in this case for the same reason that
+                // `Entry::Release` messages are ignored in the remote queue
+                // drain loop of `BasicScheduler`'s destructor.
+                if let Some(queue) = self.queue.lock().as_mut() {
+                    queue.push_back(Entry::Release(ptr));
+                }
+
                 self.unpark.unpark();
                 // Returning `None` here prevents the task plumbing from being
                 // freed. It is then up to the scheduler through the queue we
@@ -445,8 +471,17 @@ impl Schedule for Arc<Shared> {
                 cx.tasks.borrow_mut().queue.push_back(task);
             }
             _ => {
-                self.queue.lock().push_back(Entry::Schedule(task));
-                self.unpark.unpark();
+                let mut guard = self.queue.lock();
+                if let Some(queue) = guard.as_mut() {
+                    queue.push_back(Entry::Schedule(task));
+                    drop(guard);
+                    self.unpark.unpark();
+                } else {
+                    // The runtime has shut down. We drop the new task
+                    // immediately.
+                    drop(guard);
+                    task.shutdown();
+                }
             }
         });
     }

tokio/src/runtime/blocking/pool.rs

@@ -4,7 +4,6 @@ use crate::loom::sync::{Arc, Condvar, Mutex};
 use crate::loom::thread;
 use crate::runtime::blocking::schedule::NoopSchedule;
 use crate::runtime::blocking::shutdown;
-use crate::runtime::blocking::task::BlockingTask;
 use crate::runtime::builder::ThreadNameFn;
 use crate::runtime::context;
 use crate::runtime::task::{self, JoinHandle};
@@ -86,18 +85,6 @@ where
     rt.spawn_blocking(func)
 }
 
-#[allow(dead_code)]
-pub(crate) fn try_spawn_blocking<F, R>(func: F) -> Result<(), ()>
-where
-    F: FnOnce() -> R + Send + 'static,
-    R: Send + 'static,
-{
-    let rt = context::current().expect(CONTEXT_MISSING_ERROR);
-
-    let (task, _handle) = task::joinable(BlockingTask::new(func));
-    rt.blocking_spawner.spawn(task, &rt)
-}
-
 // ===== impl BlockingPool =====
 
 impl BlockingPool {

tokio/src/runtime/handle.rs

@@ -174,8 +174,11 @@ impl Handle {
         F: FnOnce() -> R + Send + 'static,
         R: Send + 'static,
     {
+        let fut = BlockingTask::new(func);
+
         #[cfg(all(tokio_unstable, feature = "tracing"))]
-        let func = {
+        let fut = {
+            use tracing::Instrument;
             #[cfg(tokio_track_caller)]
             let location = std::panic::Location::caller();
             #[cfg(tokio_track_caller)]
@@ -193,12 +196,9 @@ impl Handle {
                 kind = %"blocking",
                 function = %std::any::type_name::<F>(),
             );
-            move || {
-                let _g = span.enter();
-                func()
-            }
+            fut.instrument(span)
         };
-        let (task, handle) = task::joinable(BlockingTask::new(func));
+        let (task, handle) = task::joinable(fut);
         let _ = self.blocking_spawner.spawn(task, &self);
         handle
     }

tokio/src/runtime/queue.rs

@@ -109,7 +109,10 @@ impl<T> Local<T> {
     }
 
     /// Pushes a task to the back of the local queue, skipping the LIFO slot.
-    pub(super) fn push_back(&mut self, mut task: task::Notified<T>, inject: &Inject<T>) {
+    pub(super) fn push_back(&mut self, mut task: task::Notified<T>, inject: &Inject<T>)
+    where
+        T: crate::runtime::task::Schedule,
+    {
         let tail = loop {
             let head = self.inner.head.load(Acquire);
             let (steal, real) = unpack(head);
@@ -121,9 +124,14 @@ impl<T> Local<T> {
                 // There is capacity for the task
                 break tail;
             } else if steal != real {
-                // Concurrently stealing, this will free up capacity, so
-                // only push the new task onto the inject queue
-                inject.push(task);
+                // Concurrently stealing, this will free up capacity, so only
+                // push the new task onto the inject queue
+                //
+                // If the task failes to be pushed on the injection queue, there
+                // is nothing to be done at this point as the task cannot be a
+                // newly spawned task. Shutting down this task is handled by the
+                // worker shutdown process.
+                let _ = inject.push(task);
                 return;
             } else {
                 // Push the current task and half of the queue into the
@@ -504,16 +512,19 @@ impl<T: 'static> Inject<T> {
     }
 
     /// Pushes a value into the queue.
-    pub(super) fn push(&self, task: task::Notified<T>) {
+    ///
+    /// Returns `Err(task)` if pushing fails due to the queue being shutdown.
+    /// The caller is expected to call `shutdown()` on the task **if and only
+    /// if** it is a newly spawned task.
+    pub(super) fn push(&self, task: task::Notified<T>) -> Result<(), task::Notified<T>>
+    where
+        T: crate::runtime::task::Schedule,
+    {
         // Acquire queue lock
         let mut p = self.pointers.lock();
 
         if p.is_closed {
-            // Drop the mutex to avoid a potential deadlock when
-            // re-entering.
-            drop(p);
-            drop(task);
-            return;
+            return Err(task);
         }
 
         // safety: only mutated with the lock held
@@ -532,6 +543,7 @@ impl<T: 'static> Inject<T> {
         p.tail = Some(task);
 
         self.len.store(len + 1, Release);
+        Ok(())
     }
 
     pub(super) fn push_batch(
@@ -617,7 +629,7 @@ fn set_next(header: NonNull<task::Header>, val: Option<NonNull<task::Header>>) {
 /// Split the head value into the real head and the index a stealer is working
 /// on.
 fn unpack(n: u32) -> (u16, u16) {
-    let real = n & u16::max_value() as u32;
+    let real = n & u16::MAX as u32;
     let steal = n >> 16;
 
     (steal as u16, real as u16)
@@ -630,5 +642,5 @@ fn pack(steal: u16, real: u16) -> u32 {
 
 #[test]
 fn test_local_queue_capacity() {
-    assert!(LOCAL_QUEUE_CAPACITY - 1 <= u8::max_value() as usize);
+    assert!(LOCAL_QUEUE_CAPACITY - 1 <= u8::MAX as usize);
 }

tokio/src/runtime/spawner.rs

@@ -1,8 +1,7 @@
 cfg_rt! {
+    use crate::future::Future;
     use crate::runtime::basic_scheduler;
     use crate::task::JoinHandle;
-
-    use std::future::Future;
 }
 
 cfg_rt_multi_thread! {

tokio/src/runtime/task/core.rs

@@ -9,13 +9,13 @@
 //! Make sure to consult the relevant safety section of each function before
 //! use.
 
+use crate::future::Future;
 use crate::loom::cell::UnsafeCell;
 use crate::runtime::task::raw::{self, Vtable};
 use crate::runtime::task::state::State;
 use crate::runtime::task::{Notified, Schedule, Task};
 use crate::util::linked_list;
 
-use std::future::Future;
 use std::pin::Pin;
 use std::ptr::NonNull;
 use std::task::{Context, Poll, Waker};
@@ -71,6 +71,10 @@ pub(crate) struct Header {
 
     /// Table of function pointers for executing actions on the task.
     pub(super) vtable: &'static Vtable,
+
+    /// The tracing ID for this instrumented task.
+    #[cfg(all(tokio_unstable, feature = "tracing"))]
+    pub(super) id: Option<tracing::Id>,
 }
 
 unsafe impl Send for Header {}
@@ -93,6 +97,8 @@ impl<T: Future, S: Schedule> Cell<T, S> {
     /// Allocates a new task cell, containing the header, trailer, and core
     /// structures.
     pub(super) fn new(future: T, state: State) -> Box<Cell<T, S>> {
+        #[cfg(all(tokio_unstable, feature = "tracing"))]
+        let id = future.id();
         Box::new(Cell {
             header: Header {
                 state,
@@ -100,6 +106,8 @@ impl<T: Future, S: Schedule> Cell<T, S> {
                 queue_next: UnsafeCell::new(None),
                 stack_next: UnsafeCell::new(None),
                 vtable: raw::vtable::<T, S>(),
+                #[cfg(all(tokio_unstable, feature = "tracing"))]
+                id,
             },
             core: Core {
                 scheduler: Scheduler {

tokio/src/runtime/task/harness.rs

@@ -1,9 +1,9 @@
+use crate::future::Future;
 use crate::runtime::task::core::{Cell, Core, CoreStage, Header, Scheduler, Trailer};
 use crate::runtime::task::state::Snapshot;
 use crate::runtime::task::waker::waker_ref;
 use crate::runtime::task::{JoinError, Notified, Schedule, Task};
 
-use std::future::Future;
 use std::mem;
 use std::panic;
 use std::ptr::NonNull;
@@ -146,6 +146,11 @@ where
         }
     }
 
+    #[cfg(all(tokio_unstable, feature = "tracing"))]
+    pub(super) fn id(&self) -> Option<&tracing::Id> {
+        self.header().id.as_ref()
+    }
+
     /// Forcibly shutdown the task
     ///
     /// Attempt to transition to `Running` in order to forcibly shutdown the
@@ -415,7 +420,7 @@ fn poll_future<T: Future>(
     cx: Context<'_>,
 ) -> PollFuture<T::Output> {
     if snapshot.is_cancelled() {
-        PollFuture::Complete(Err(JoinError::cancelled()), snapshot.is_join_interested())
+        PollFuture::Complete(Err(cancel_task(core)), snapshot.is_join_interested())
     } else {
         let res = panic::catch_unwind(panic::AssertUnwindSafe(|| {
             struct Guard<'a, T: Future> {

tokio/src/runtime/task/mod.rs

@@ -26,9 +26,9 @@ cfg_rt_multi_thread! {
     pub(crate) use self::stack::TransferStack;
 }
 
+use crate::future::Future;
 use crate::util::linked_list;
 
-use std::future::Future;
 use std::marker::PhantomData;
 use std::ptr::NonNull;
 use std::{fmt, mem};

tokio/src/runtime/task/raw.rs

@@ -1,6 +1,6 @@
+use crate::future::Future;
 use crate::runtime::task::{Cell, Harness, Header, Schedule, State};
 
-use std::future::Future;
 use std::ptr::NonNull;
 use std::task::{Poll, Waker};
 

tokio/src/runtime/task/state.rs

@@ -29,12 +29,15 @@ const LIFECYCLE_MASK: usize = 0b11;
 const NOTIFIED: usize = 0b100;
 
 /// The join handle is still around
+#[allow(clippy::unusual_byte_groupings)] // https://github.com/rust-lang/rust-clippy/issues/6556
 const JOIN_INTEREST: usize = 0b1_000;
 
 /// A join handle waker has been set
+#[allow(clippy::unusual_byte_groupings)] // https://github.com/rust-lang/rust-clippy/issues/6556
 const JOIN_WAKER: usize = 0b10_000;
 
 /// The task has been forcibly cancelled.
+#[allow(clippy::unusual_byte_groupings)] // https://github.com/rust-lang/rust-clippy/issues/6556
 const CANCELLED: usize = 0b100_000;
 
 /// All bits
@@ -52,7 +55,7 @@ const REF_ONE: usize = 1 << REF_COUNT_SHIFT;
 /// State a task is initialized with
 ///
 /// A task is initialized with two references: one for the scheduler and one for
-/// the `JoinHandle`. As the task starts with a `JoinHandle`, `JOIN_INTERST` is
+/// the `JoinHandle`. As the task starts with a `JoinHandle`, `JOIN_INTEREST` is
 /// set. A new task is immediately pushed into the run queue for execution and
 /// starts with the `NOTIFIED` flag set.
 const INITIAL_STATE: usize = (REF_ONE * 2) | JOIN_INTEREST | NOTIFIED;
@@ -64,7 +67,7 @@ impl State {
     pub(super) fn new() -> State {
         // A task is initialized with three references: one for the scheduler,
         // one for the `JoinHandle`, one for the task handle made available in
-        // release. As the task starts with a `JoinHandle`, `JOIN_INTERST` is
+        // release. As the task starts with a `JoinHandle`, `JOIN_INTEREST` is
         // set. A new task is immediately pushed into the run queue for
         // execution and starts with the `NOTIFIED` flag set.
         State {
@@ -315,7 +318,7 @@ impl State {
         let prev = self.val.fetch_add(REF_ONE, Relaxed);
 
         // If the reference count overflowed, abort.
-        if prev > isize::max_value() as usize {
+        if prev > isize::MAX as usize {
             process::abort();
         }
     }
@@ -419,7 +422,7 @@ impl Snapshot {
     }
 
     fn ref_inc(&mut self) {
-        assert!(self.0 <= isize::max_value() as usize);
+        assert!(self.0 <= isize::MAX as usize);
         self.0 += REF_ONE;
     }
 

tokio/src/runtime/task/waker.rs

@@ -1,7 +1,7 @@
+use crate::future::Future;
 use crate::runtime::task::harness::Harness;
 use crate::runtime::task::{Header, Schedule};
 
-use std::future::Future;
 use std::marker::PhantomData;
 use std::mem::ManuallyDrop;
 use std::ops;
@@ -44,12 +44,38 @@ impl<S> ops::Deref for WakerRef<'_, S> {
     }
 }
 
+cfg_trace! {
+    macro_rules! trace {
+        ($harness:expr, $op:expr) => {
+            if let Some(id) = $harness.id() {
+                tracing::trace!(
+                    target: "tokio::task::waker",
+                    op = $op,
+                    task.id = id.into_u64(),
+                );
+            }
+        }
+    }
+}
+
+cfg_not_trace! {
+    macro_rules! trace {
+        ($harness:expr, $op:expr) => {
+            // noop
+            let _ = &$harness;
+        }
+    }
+}
+
 unsafe fn clone_waker<T, S>(ptr: *const ()) -> RawWaker
 where
     T: Future,
     S: Schedule,
 {
     let header = ptr as *const Header;
+    let ptr = NonNull::new_unchecked(ptr as *mut Header);
+    let harness = Harness::<T, S>::from_raw(ptr);
+    trace!(harness, "waker.clone");
     (*header).state.ref_inc();
     raw_waker::<T, S>(header)
 }
@@ -61,6 +87,7 @@ where
 {
     let ptr = NonNull::new_unchecked(ptr as *mut Header);
     let harness = Harness::<T, S>::from_raw(ptr);
+    trace!(harness, "waker.drop");
     harness.drop_reference();
 }
 
@@ -71,6 +98,7 @@ where
 {
     let ptr = NonNull::new_unchecked(ptr as *mut Header);
     let harness = Harness::<T, S>::from_raw(ptr);
+    trace!(harness, "waker.wake");
     harness.wake_by_val();
 }
 
@@ -82,6 +110,7 @@ where
 {
     let ptr = NonNull::new_unchecked(ptr as *mut Header);
     let harness = Harness::<T, S>::from_raw(ptr);
+    trace!(harness, "waker.wake_by_ref");
     harness.wake_by_ref();
 }
 

tokio/src/runtime/tests/loom_shutdown_join.rs

@@ -0,0 +1,28 @@
+use crate::runtime::{Builder, Handle};
+
+#[test]
+fn join_handle_cancel_on_shutdown() {
+    let mut builder = loom::model::Builder::new();
+    builder.preemption_bound = Some(2);
+    builder.check(|| {
+        use futures::future::FutureExt;
+
+        let rt = Builder::new_multi_thread()
+            .worker_threads(2)
+            .build()
+            .unwrap();
+
+        let handle = rt.block_on(async move { Handle::current() });
+
+        let jh1 = handle.spawn(futures::future::pending::<()>());
+
+        drop(rt);
+
+        let jh2 = handle.spawn(futures::future::pending::<()>());
+
+        let err1 = jh1.now_or_never().unwrap().unwrap_err();
+        let err2 = jh2.now_or_never().unwrap().unwrap_err();
+        assert!(err1.is_cancelled());
+        assert!(err2.is_cancelled());
+    });
+}

tokio/src/runtime/tests/mod.rs

@@ -4,6 +4,7 @@ cfg_loom! {
     mod loom_oneshot;
     mod loom_pool;
     mod loom_queue;
+    mod loom_shutdown_join;
 }
 
 cfg_not_loom! {

tokio/src/runtime/tests/task.rs

@@ -79,7 +79,7 @@ static CURRENT: TryLock<Option<Runtime>> = TryLock::new(None);
 
 impl Runtime {
     fn tick(&self) -> usize {
-        self.tick_max(usize::max_value())
+        self.tick_max(usize::MAX)
     }
 
     fn tick_max(&self, max: usize) -> usize {

tokio/src/runtime/thread_pool/mod.rs

@@ -90,11 +90,17 @@ impl Spawner {
     /// Spawns a future onto the thread pool
     pub(crate) fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
     where
-        F: Future + Send + 'static,
+        F: crate::future::Future + Send + 'static,
         F::Output: Send + 'static,
     {
         let (task, handle) = task::joinable(future);
-        self.shared.schedule(task, false);
+
+        if let Err(task) = self.shared.schedule(task, false) {
+            // The newly spawned task could not be scheduled because the runtime
+            // is shutting down. The task must be explicitly shutdown at this point.
+            task.shutdown();
+        }
+
         handle
     }
 

tokio/src/runtime/thread_pool/worker.rs

@@ -709,16 +709,22 @@ impl task::Schedule for Arc<Worker> {
     }
 
     fn schedule(&self, task: Notified) {
-        self.shared.schedule(task, false);
+        // Because this is not a newly spawned task, if scheduling fails due to
+        // the runtime shutting down, there is no special work that must happen
+        // here.
+        let _ = self.shared.schedule(task, false);
     }
 
     fn yield_now(&self, task: Notified) {
-        self.shared.schedule(task, true);
+        // Because this is not a newly spawned task, if scheduling fails due to
+        // the runtime shutting down, there is no special work that must happen
+        // here.
+        let _ = self.shared.schedule(task, true);
     }
 }
 
 impl Shared {
-    pub(super) fn schedule(&self, task: Notified, is_yield: bool) {
+    pub(super) fn schedule(&self, task: Notified, is_yield: bool) -> Result<(), Notified> {
         CURRENT.with(|maybe_cx| {
             if let Some(cx) = maybe_cx {
                 // Make sure the task is part of the **current** scheduler.
@@ -726,15 +732,16 @@ impl Shared {
                     // And the current thread still holds a core
                     if let Some(core) = cx.core.borrow_mut().as_mut() {
                         self.schedule_local(core, task, is_yield);
-                        return;
+                        return Ok(());
                     }
                 }
             }
 
             // Otherwise, use the inject queue
-            self.inject.push(task);
+            self.inject.push(task)?;
             self.notify_parked();
-        });
+            Ok(())
+        })
     }
 
     fn schedule_local(&self, core: &mut Core, task: Notified, is_yield: bool) {
@@ -823,7 +830,9 @@ impl Shared {
         }
 
         // Drain the injection queue
-        while self.inject.pop().is_some() {}
+        while let Some(task) = self.inject.pop() {
+            task.shutdown();
+        }
     }
 
     fn ptr_eq(&self, other: &Shared) -> bool {

tokio/src/sync/mod.rs

@@ -428,6 +428,11 @@
 //!   bounding of any kind.
 
 cfg_sync! {
+    /// Named future types.
+    pub mod futures {
+        pub use super::notify::Notified;
+    }
+
     mod barrier;
     pub use barrier::{Barrier, BarrierWaitResult};
 

tokio/src/sync/mpsc/bounded.rs

@@ -65,7 +65,7 @@ pub struct Receiver<T> {
 /// with backpressure.
 ///
 /// The channel will buffer up to the provided number of messages.  Once the
-/// buffer is full, attempts to `send` new messages will wait until a message is
+/// buffer is full, attempts to send new messages will wait until a message is
 /// received from the channel. The provided buffer capacity must be at least 1.
 ///
 /// All data sent on `Sender` will become available on `Receiver` in the same
@@ -76,7 +76,7 @@ pub struct Receiver<T> {
 ///
 /// If the `Receiver` is disconnected while trying to `send`, the `send` method
 /// will return a `SendError`. Similarly, if `Sender` is disconnected while
-/// trying to `recv`, the `recv` method will return a `RecvError`.
+/// trying to `recv`, the `recv` method will return `None`.
 ///
 /// # Panics
 ///
@@ -887,7 +887,7 @@ impl<T> Sender<T> {
     ///     let permit = tx.reserve().await.unwrap();
     ///     assert_eq!(tx.capacity(), 4);
     ///
-    ///     // Sending and receiving a value increases the caapcity by one.
+    ///     // Sending and receiving a value increases the capacity by one.
     ///     permit.send(());
     ///     rx.recv().await.unwrap();
     ///     assert_eq!(tx.capacity(), 5);

tokio/src/sync/mpsc/error.rs

@@ -55,14 +55,18 @@ impl<T> From<SendError<T>> for TrySendError<T> {
 
 /// Error returned by `Receiver`.
 #[derive(Debug)]
+#[doc(hidden)]
+#[deprecated(note = "This type is unused because recv returns an Option.")]
 pub struct RecvError(());
 
+#[allow(deprecated)]
 impl fmt::Display for RecvError {
     fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(fmt, "channel closed")
     }
 }
 
+#[allow(deprecated)]
 impl Error for RecvError {}
 
 cfg_time! {

tokio/src/sync/notify.rs

@@ -140,7 +140,7 @@ struct Waiter {
     _p: PhantomPinned,
 }
 
-/// Future returned from `notified()`
+/// Future returned from [`Notify::notified()`]
 #[derive(Debug)]
 pub struct Notified<'a> {
     /// The `Notify` being received on.

tokio/src/sync/semaphore.rs

@@ -24,7 +24,55 @@ use std::sync::Arc;
 /// To use the `Semaphore` in a poll function, you can use the [`PollSemaphore`]
 /// utility.
 ///
+/// # Examples
+///
+/// Basic usage:
+///
+/// ```
+/// use tokio::sync::{Semaphore, TryAcquireError};
+///
+/// #[tokio::main]
+/// async fn main() {
+///     let semaphore = Semaphore::new(3);
+///
+///     let a_permit = semaphore.acquire().await.unwrap();
+///     let two_permits = semaphore.acquire_many(2).await.unwrap();
+///
+///     assert_eq!(semaphore.available_permits(), 0);
+///
+///     let permit_attempt = semaphore.try_acquire();
+///     assert_eq!(permit_attempt.err(), Some(TryAcquireError::NoPermits));
+/// }
+/// ```
+///
+/// Use [`Semaphore::acquire_owned`] to move permits across tasks:
+///
+/// ```
+/// use std::sync::Arc;
+/// use tokio::sync::Semaphore;
+///
+/// #[tokio::main]
+/// async fn main() {
+///     let semaphore = Arc::new(Semaphore::new(3));
+///     let mut join_handles = Vec::new();
+///
+///     for _ in 0..5 {
+///         let permit = semaphore.clone().acquire_owned().await.unwrap();
+///         join_handles.push(tokio::spawn(async move {
+///             // perform task...
+///             // explicitly own `permit` in the task
+///             drop(permit);
+///         }));
+///     }
+///
+///     for handle in join_handles {
+///         handle.await.unwrap();
+///     }
+/// }
+/// ```
+///
 /// [`PollSemaphore`]: https://docs.rs/tokio-util/0.6/tokio_util/sync/struct.PollSemaphore.html
+/// [`Semaphore::acquire_owned`]: crate::sync::Semaphore::acquire_owned
 #[derive(Debug)]
 pub struct Semaphore {
     /// The low level semaphore
@@ -79,6 +127,15 @@ impl Semaphore {
     }
 
     /// Creates a new semaphore with the initial number of permits.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::sync::Semaphore;
+    ///
+    /// static SEM: Semaphore = Semaphore::const_new(10);
+    /// ```
+    ///
     #[cfg(all(feature = "parking_lot", not(all(loom, test))))]
     #[cfg_attr(docsrs, doc(cfg(feature = "parking_lot")))]
     pub const fn const_new(permits: usize) -> Self {
@@ -105,6 +162,26 @@ impl Semaphore {
     /// Otherwise, this returns a [`SemaphorePermit`] representing the
     /// acquired permit.
     ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::sync::Semaphore;
+    ///
+    /// #[tokio::main]
+    /// async fn main() {
+    ///     let semaphore = Semaphore::new(2);
+    ///
+    ///     let permit_1 = semaphore.acquire().await.unwrap();
+    ///     assert_eq!(semaphore.available_permits(), 1);
+    ///
+    ///     let permit_2 = semaphore.acquire().await.unwrap();
+    ///     assert_eq!(semaphore.available_permits(), 0);
+    ///
+    ///     drop(permit_1);
+    ///     assert_eq!(semaphore.available_permits(), 1);
+    /// }
+    /// ```
+    ///
     /// [`AcquireError`]: crate::sync::AcquireError
     /// [`SemaphorePermit`]: crate::sync::SemaphorePermit
     pub async fn acquire(&self) -> Result<SemaphorePermit<'_>, AcquireError> {
@@ -121,6 +198,20 @@ impl Semaphore {
     /// Otherwise, this returns a [`SemaphorePermit`] representing the
     /// acquired permits.
     ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::sync::Semaphore;
+    ///
+    /// #[tokio::main]
+    /// async fn main() {
+    ///     let semaphore = Semaphore::new(5);
+    ///
+    ///     let permit = semaphore.acquire_many(3).await.unwrap();
+    ///     assert_eq!(semaphore.available_permits(), 2);
+    /// }
+    /// ```
+    ///
     /// [`AcquireError`]: crate::sync::AcquireError
     /// [`SemaphorePermit`]: crate::sync::SemaphorePermit
     pub async fn acquire_many(&self, n: u32) -> Result<SemaphorePermit<'_>, AcquireError> {
@@ -137,6 +228,25 @@ impl Semaphore {
     /// and a [`TryAcquireError::NoPermits`] if there are no permits left. Otherwise,
     /// this returns a [`SemaphorePermit`] representing the acquired permits.
     ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::sync::{Semaphore, TryAcquireError};
+    ///
+    /// # fn main() {
+    /// let semaphore = Semaphore::new(2);
+    ///
+    /// let permit_1 = semaphore.try_acquire().unwrap();
+    /// assert_eq!(semaphore.available_permits(), 1);
+    ///
+    /// let permit_2 = semaphore.try_acquire().unwrap();
+    /// assert_eq!(semaphore.available_permits(), 0);
+    ///
+    /// let permit_3 = semaphore.try_acquire();
+    /// assert_eq!(permit_3.err(), Some(TryAcquireError::NoPermits));
+    /// # }
+    /// ```
+    ///
     /// [`TryAcquireError::Closed`]: crate::sync::TryAcquireError::Closed
     /// [`TryAcquireError::NoPermits`]: crate::sync::TryAcquireError::NoPermits
     /// [`SemaphorePermit`]: crate::sync::SemaphorePermit
@@ -153,8 +263,24 @@ impl Semaphore {
     /// Tries to acquire `n` permits from the semaphore.
     ///
     /// If the semaphore has been closed, this returns a [`TryAcquireError::Closed`]
-    /// and a [`TryAcquireError::NoPermits`] if there are no permits left. Otherwise,
-    /// this returns a [`SemaphorePermit`] representing the acquired permits.
+    /// and a [`TryAcquireError::NoPermits`] if there are not enough permits left.
+    /// Otherwise, this returns a [`SemaphorePermit`] representing the acquired permits.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::sync::{Semaphore, TryAcquireError};
+    ///
+    /// # fn main() {
+    /// let semaphore = Semaphore::new(4);
+    ///
+    /// let permit_1 = semaphore.try_acquire_many(3).unwrap();
+    /// assert_eq!(semaphore.available_permits(), 1);
+    ///
+    /// let permit_2 = semaphore.try_acquire_many(2);
+    /// assert_eq!(permit_2.err(), Some(TryAcquireError::NoPermits));
+    /// # }
+    /// ```
     ///
     /// [`TryAcquireError::Closed`]: crate::sync::TryAcquireError::Closed
     /// [`TryAcquireError::NoPermits`]: crate::sync::TryAcquireError::NoPermits
@@ -176,6 +302,32 @@ impl Semaphore {
     /// Otherwise, this returns a [`OwnedSemaphorePermit`] representing the
     /// acquired permit.
     ///
+    /// # Examples
+    ///
+    /// ```
+    /// use std::sync::Arc;
+    /// use tokio::sync::Semaphore;
+    ///
+    /// #[tokio::main]
+    /// async fn main() {
+    ///     let semaphore = Arc::new(Semaphore::new(3));
+    ///     let mut join_handles = Vec::new();
+    ///
+    ///     for _ in 0..5 {
+    ///         let permit = semaphore.clone().acquire_owned().await.unwrap();
+    ///         join_handles.push(tokio::spawn(async move {
+    ///             // perform task...
+    ///             // explicitly own `permit` in the task
+    ///             drop(permit);
+    ///         }));
+    ///     }
+    ///
+    ///     for handle in join_handles {
+    ///         handle.await.unwrap();
+    ///     }
+    /// }
+    /// ```
+    ///
     /// [`Arc`]: std::sync::Arc
     /// [`AcquireError`]: crate::sync::AcquireError
     /// [`OwnedSemaphorePermit`]: crate::sync::OwnedSemaphorePermit
@@ -194,6 +346,32 @@ impl Semaphore {
     /// Otherwise, this returns a [`OwnedSemaphorePermit`] representing the
     /// acquired permit.
     ///
+    /// # Examples
+    ///
+    /// ```
+    /// use std::sync::Arc;
+    /// use tokio::sync::Semaphore;
+    ///
+    /// #[tokio::main]
+    /// async fn main() {
+    ///     let semaphore = Arc::new(Semaphore::new(10));
+    ///     let mut join_handles = Vec::new();
+    ///
+    ///     for _ in 0..5 {
+    ///         let permit = semaphore.clone().acquire_many_owned(2).await.unwrap();
+    ///         join_handles.push(tokio::spawn(async move {
+    ///             // perform task...
+    ///             // explicitly own `permit` in the task
+    ///             drop(permit);
+    ///         }));
+    ///     }
+    ///
+    ///     for handle in join_handles {
+    ///         handle.await.unwrap();
+    ///     }
+    /// }
+    /// ```
+    ///
     /// [`Arc`]: std::sync::Arc
     /// [`AcquireError`]: crate::sync::AcquireError
     /// [`OwnedSemaphorePermit`]: crate::sync::OwnedSemaphorePermit
@@ -216,6 +394,26 @@ impl Semaphore {
     /// Otherwise, this returns a [`OwnedSemaphorePermit`] representing the
     /// acquired permit.
     ///
+    /// # Examples
+    ///
+    /// ```
+    /// use std::sync::Arc;
+    /// use tokio::sync::{Semaphore, TryAcquireError};
+    ///
+    /// # fn main() {
+    /// let semaphore = Arc::new(Semaphore::new(2));
+    ///
+    /// let permit_1 = Arc::clone(&semaphore).try_acquire_owned().unwrap();
+    /// assert_eq!(semaphore.available_permits(), 1);
+    ///
+    /// let permit_2 = Arc::clone(&semaphore).try_acquire_owned().unwrap();
+    /// assert_eq!(semaphore.available_permits(), 0);
+    ///
+    /// let permit_3 = semaphore.try_acquire_owned();
+    /// assert_eq!(permit_3.err(), Some(TryAcquireError::NoPermits));
+    /// # }
+    /// ```
+    ///
     /// [`Arc`]: std::sync::Arc
     /// [`TryAcquireError::Closed`]: crate::sync::TryAcquireError::Closed
     /// [`TryAcquireError::NoPermits`]: crate::sync::TryAcquireError::NoPermits
@@ -238,6 +436,23 @@ impl Semaphore {
     /// Otherwise, this returns a [`OwnedSemaphorePermit`] representing the
     /// acquired permit.
     ///
+    /// # Examples
+    ///
+    /// ```
+    /// use std::sync::Arc;
+    /// use tokio::sync::{Semaphore, TryAcquireError};
+    ///
+    /// # fn main() {
+    /// let semaphore = Arc::new(Semaphore::new(4));
+    ///
+    /// let permit_1 = Arc::clone(&semaphore).try_acquire_many_owned(3).unwrap();
+    /// assert_eq!(semaphore.available_permits(), 1);
+    ///
+    /// let permit_2 = semaphore.try_acquire_many_owned(2);
+    /// assert_eq!(permit_2.err(), Some(TryAcquireError::NoPermits));
+    /// # }
+    /// ```
+    ///
     /// [`Arc`]: std::sync::Arc
     /// [`TryAcquireError::Closed`]: crate::sync::TryAcquireError::Closed
     /// [`TryAcquireError::NoPermits`]: crate::sync::TryAcquireError::NoPermits

tokio/src/sync/task/atomic_waker.rs

@@ -29,7 +29,7 @@ pub(crate) struct AtomicWaker {
 
 // `AtomicWaker` is a multi-consumer, single-producer transfer cell. The cell
 // stores a `Waker` value produced by calls to `register` and many threads can
-// race to take the waker by calling `wake.
+// race to take the waker by calling `wake`.
 //
 // If a new `Waker` instance is produced by calling `register` before an existing
 // one is consumed, then the existing one is overwritten.

tokio/src/sync/watch.rs

@@ -417,6 +417,28 @@ impl<T> Sender<T> {
             Receiver::from_shared(version, shared)
         }
     }
+
+    /// Returns the number of receivers that currently exist
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::sync::watch;
+    ///
+    /// #[tokio::main]
+    /// async fn main() {
+    ///     let (tx, rx1) = watch::channel("hello");
+    ///
+    ///     assert_eq!(1, tx.receiver_count());
+    ///
+    ///     let mut _rx2 = rx1.clone();
+    ///
+    ///     assert_eq!(2, tx.receiver_count());
+    /// }
+    /// ```
+    pub fn receiver_count(&self) -> usize {
+        self.shared.ref_count_rx.load(Relaxed)
+    }
 }
 
 impl<T> Drop for Sender<T> {

tokio/src/task/mod.rs

@@ -86,7 +86,7 @@
 //! ```
 //!
 //! Again, like `std::thread`'s [`JoinHandle` type][thread_join], if the spawned
-//! task panics, awaiting its `JoinHandle` will return a [`JoinError`]`. For
+//! task panics, awaiting its `JoinHandle` will return a [`JoinError`]. For
 //! example:
 //!
 //! ```

tokio/src/time/driver/entry.rs

@@ -68,7 +68,7 @@ use std::{marker::PhantomPinned, pin::Pin, ptr::NonNull};
 
 type TimerResult = Result<(), crate::time::error::Error>;
 
-const STATE_DEREGISTERED: u64 = u64::max_value();
+const STATE_DEREGISTERED: u64 = u64::MAX;
 const STATE_PENDING_FIRE: u64 = STATE_DEREGISTERED - 1;
 const STATE_MIN_VALUE: u64 = STATE_PENDING_FIRE;
 
@@ -85,10 +85,10 @@ const STATE_MIN_VALUE: u64 = STATE_PENDING_FIRE;
 /// requires only the driver lock.
 pub(super) struct StateCell {
     /// Holds either the scheduled expiration time for this timer, or (if the
-    /// timer has been fired and is unregistered), `u64::max_value()`.
+    /// timer has been fired and is unregistered), `u64::MAX`.
     state: AtomicU64,
     /// If the timer is fired (an Acquire order read on state shows
-    /// `u64::max_value()`), holds the result that should be returned from
+    /// `u64::MAX`), holds the result that should be returned from
     /// polling the timer. Otherwise, the contents are unspecified and reading
     /// without holding the driver lock is undefined behavior.
     result: UnsafeCell<TimerResult>,
@@ -125,7 +125,7 @@ impl StateCell {
     fn when(&self) -> Option<u64> {
         let cur_state = self.state.load(Ordering::Relaxed);
 
-        if cur_state == u64::max_value() {
+        if cur_state == u64::MAX {
             None
         } else {
             Some(cur_state)
@@ -271,7 +271,7 @@ impl StateCell {
     /// ordering, but is conservative - if it returns false, the timer is
     /// definitely _not_ registered.
     pub(super) fn might_be_registered(&self) -> bool {
-        self.state.load(Ordering::Relaxed) != u64::max_value()
+        self.state.load(Ordering::Relaxed) != u64::MAX
     }
 }
 
@@ -591,7 +591,7 @@ impl TimerHandle {
         match self.inner.as_ref().state.mark_pending(not_after) {
             Ok(()) => {
                 // mark this as being on the pending queue in cached_when
-                self.inner.as_ref().set_cached_when(u64::max_value());
+                self.inner.as_ref().set_cached_when(u64::MAX);
                 Ok(())
             }
             Err(tick) => {

tokio/src/time/driver/wheel/mod.rs

@@ -119,7 +119,7 @@ impl Wheel {
     pub(crate) unsafe fn remove(&mut self, item: NonNull<TimerShared>) {
         unsafe {
             let when = item.as_ref().cached_when();
-            if when == u64::max_value() {
+            if when == u64::MAX {
                 self.pending.remove(item);
             } else {
                 debug_assert!(

tokio/src/util/linked_list.rs

@@ -50,6 +50,7 @@ pub(crate) unsafe trait Link {
     type Target;
 
     /// Convert the handle to a raw pointer without consuming the handle
+    #[allow(clippy::wrong_self_convention)]
     fn as_raw(handle: &Self::Handle) -> NonNull<Self::Target>;
 
     /// Convert the raw pointer to a handle

tokio/src/util/wake.rs

@@ -54,11 +54,7 @@ unsafe fn inc_ref_count<T: Wake>(data: *const ()) {
     let arc = ManuallyDrop::new(Arc::<T>::from_raw(data as *const T));
 
     // Now increase refcount, but don't drop new refcount either
-    let arc_clone: ManuallyDrop<_> = arc.clone();
-
-    // Drop explicitly to avoid clippy warnings
-    drop(arc);
-    drop(arc_clone);
+    let _arc_clone: ManuallyDrop<_> = arc.clone();
 }
 
 unsafe fn clone_arc_raw<T: Wake>(data: *const ()) -> RawWaker {

tokio/tests/async_send_sync.rs

@@ -1,6 +1,6 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
-#![allow(clippy::type_complexity)]
+#![allow(clippy::type_complexity, clippy::diverging_sub_expression)]
 
 use std::cell::Cell;
 use std::future::Future;

tokio/tests/io_buf_reader.rs

@@ -120,10 +120,7 @@ async fn test_buffered_reader_seek() {
 
     assert_eq!(reader.seek(SeekFrom::Start(3)).await.unwrap(), 3);
     assert_eq!(run_fill_buf!(reader).unwrap(), &[0, 1][..]);
-    assert!(reader
-        .seek(SeekFrom::Current(i64::min_value()))
-        .await
-        .is_err());
+    assert!(reader.seek(SeekFrom::Current(i64::MIN)).await.is_err());
     assert_eq!(run_fill_buf!(reader).unwrap(), &[0, 1][..]);
     assert_eq!(reader.seek(SeekFrom::Current(1)).await.unwrap(), 4);
     assert_eq!(run_fill_buf!(reader).unwrap(), &[1, 2][..]);
@@ -163,7 +160,7 @@ async fn test_buffered_reader_seek_underflow() {
                     self.pos = self.pos.wrapping_add(n as u64);
                 }
                 SeekFrom::End(n) => {
-                    self.pos = u64::max_value().wrapping_add(n as u64);
+                    self.pos = u64::MAX.wrapping_add(n as u64);
                 }
             }
             Ok(())
@@ -175,18 +172,12 @@ async fn test_buffered_reader_seek_underflow() {
 
     let mut reader = BufReader::with_capacity(5, PositionReader { pos: 0 });
     assert_eq!(run_fill_buf!(reader).unwrap(), &[0, 1, 2, 3, 4][..]);
-    assert_eq!(
-        reader.seek(SeekFrom::End(-5)).await.unwrap(),
-        u64::max_value() - 5
-    );
+    assert_eq!(reader.seek(SeekFrom::End(-5)).await.unwrap(), u64::MAX - 5);
     assert_eq!(run_fill_buf!(reader).unwrap().len(), 5);
     // the following seek will require two underlying seeks
     let expected = 9_223_372_036_854_775_802;
     assert_eq!(
-        reader
-            .seek(SeekFrom::Current(i64::min_value()))
-            .await
-            .unwrap(),
+        reader.seek(SeekFrom::Current(i64::MIN)).await.unwrap(),
         expected
     );
     assert_eq!(run_fill_buf!(reader).unwrap().len(), 5);
@@ -350,10 +341,7 @@ async fn maybe_pending_seek() {
 
     assert_eq!(reader.seek(SeekFrom::Current(3)).await.unwrap(), 3);
     assert_eq!(run_fill_buf!(reader).unwrap(), &[0, 1][..]);
-    assert!(reader
-        .seek(SeekFrom::Current(i64::min_value()))
-        .await
-        .is_err());
+    assert!(reader.seek(SeekFrom::Current(i64::MIN)).await.is_err());
     assert_eq!(run_fill_buf!(reader).unwrap(), &[0, 1][..]);
     assert_eq!(reader.seek(SeekFrom::Current(1)).await.unwrap(), 4);
     assert_eq!(run_fill_buf!(reader).unwrap(), &[1, 2][..]);

tokio/tests/macros_select.rs

@@ -537,3 +537,13 @@ async fn biased_eventually_ready() {
 
     assert_eq!(count, 3);
 }
+
+// https://github.com/tokio-rs/tokio/issues/3830
+// https://github.com/rust-lang/rust-clippy/issues/7304
+#[warn(clippy::default_numeric_fallback)]
+pub async fn default_numeric_fallback() {
+    tokio::select! {
+        _ = async {} => (),
+        else => (),
+    }
+}

tokio/tests/macros_test.rs

@@ -2,20 +2,12 @@ use tokio::test;
 
 #[test]
 async fn test_macro_can_be_used_via_use() {
-    tokio::spawn(async {
-        assert_eq!(1 + 1, 2);
-    })
-    .await
-    .unwrap();
+    tokio::spawn(async {}).await.unwrap();
 }
 
 #[tokio::test]
 async fn test_macro_is_resilient_to_shadowing() {
-    tokio::spawn(async {
-        assert_eq!(1 + 1, 2);
-    })
-    .await
-    .unwrap();
+    tokio::spawn(async {}).await.unwrap();
 }
 
 // https://github.com/tokio-rs/tokio/issues/3403

tokio/tests/named_pipe.rs

@@ -0,0 +1,214 @@
+#![cfg(feature = "full")]
+#![cfg(all(windows))]
+
+use std::io;
+use std::mem;
+use std::os::windows::io::AsRawHandle;
+use std::time::Duration;
+use tokio::io::AsyncWriteExt;
+use tokio::net::windows::named_pipe::{ClientOptions, PipeMode, ServerOptions};
+use tokio::time;
+use winapi::shared::winerror;
+
+#[tokio::test]
+async fn test_named_pipe_client_drop() -> io::Result<()> {
+    const PIPE_NAME: &str = r"\\.\pipe\test-named-pipe-client-drop";
+
+    let mut server = ServerOptions::new().create(PIPE_NAME)?;
+
+    assert_eq!(num_instances("test-named-pipe-client-drop")?, 1);
+
+    let client = ClientOptions::new().open(PIPE_NAME)?;
+
+    server.connect().await?;
+    drop(client);
+
+    // instance will be broken because client is gone
+    match server.write_all(b"ping").await {
+        Err(e) if e.raw_os_error() == Some(winerror::ERROR_NO_DATA as i32) => (),
+        x => panic!("{:?}", x),
+    }
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_named_pipe_single_client() -> io::Result<()> {
+    use tokio::io::{AsyncBufReadExt as _, BufReader};
+
+    const PIPE_NAME: &str = r"\\.\pipe\test-named-pipe-single-client";
+
+    let server = ServerOptions::new().create(PIPE_NAME)?;
+
+    let server = tokio::spawn(async move {
+        // Note: we wait for a client to connect.
+        server.connect().await?;
+
+        let mut server = BufReader::new(server);
+
+        let mut buf = String::new();
+        server.read_line(&mut buf).await?;
+        server.write_all(b"pong\n").await?;
+        Ok::<_, io::Error>(buf)
+    });
+
+    let client = tokio::spawn(async move {
+        let client = ClientOptions::new().open(PIPE_NAME)?;
+
+        let mut client = BufReader::new(client);
+
+        let mut buf = String::new();
+        client.write_all(b"ping\n").await?;
+        client.read_line(&mut buf).await?;
+        Ok::<_, io::Error>(buf)
+    });
+
+    let (server, client) = tokio::try_join!(server, client)?;
+
+    assert_eq!(server?, "ping\n");
+    assert_eq!(client?, "pong\n");
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_named_pipe_multi_client() -> io::Result<()> {
+    use tokio::io::{AsyncBufReadExt as _, BufReader};
+
+    const PIPE_NAME: &str = r"\\.\pipe\test-named-pipe-multi-client";
+    const N: usize = 10;
+
+    // The first server needs to be constructed early so that clients can
+    // be correctly connected. Otherwise calling .wait will cause the client to
+    // error.
+    let mut server = ServerOptions::new().create(PIPE_NAME)?;
+
+    let server = tokio::spawn(async move {
+        for _ in 0..N {
+            // Wait for client to connect.
+            server.connect().await?;
+            let mut inner = BufReader::new(server);
+
+            // Construct the next server to be connected before sending the one
+            // we already have of onto a task. This ensures that the server
+            // isn't closed (after it's done in the task) before a new one is
+            // available. Otherwise the client might error with
+            // `io::ErrorKind::NotFound`.
+            server = ServerOptions::new().create(PIPE_NAME)?;
+
+            let _ = tokio::spawn(async move {
+                let mut buf = String::new();
+                inner.read_line(&mut buf).await?;
+                inner.write_all(b"pong\n").await?;
+                inner.flush().await?;
+                Ok::<_, io::Error>(())
+            });
+        }
+
+        Ok::<_, io::Error>(())
+    });
+
+    let mut clients = Vec::new();
+
+    for _ in 0..N {
+        clients.push(tokio::spawn(async move {
+            // This showcases a generic connect loop.
+            //
+            // We immediately try to create a client, if it's not found or the
+            // pipe is busy we use the specialized wait function on the client
+            // builder.
+            let client = loop {
+                match ClientOptions::new().open(PIPE_NAME) {
+                    Ok(client) => break client,
+                    Err(e) if e.raw_os_error() == Some(winerror::ERROR_PIPE_BUSY as i32) => (),
+                    Err(e) if e.kind() == io::ErrorKind::NotFound => (),
+                    Err(e) => return Err(e),
+                }
+
+                // Wait for a named pipe to become available.
+                time::sleep(Duration::from_millis(50)).await;
+            };
+
+            let mut client = BufReader::new(client);
+
+            let mut buf = String::new();
+            client.write_all(b"ping\n").await?;
+            client.flush().await?;
+            client.read_line(&mut buf).await?;
+            Ok::<_, io::Error>(buf)
+        }));
+    }
+
+    for client in clients {
+        let result = client.await?;
+        assert_eq!(result?, "pong\n");
+    }
+
+    server.await??;
+    Ok(())
+}
+
+// This tests what happens when a client tries to disconnect.
+#[tokio::test]
+async fn test_named_pipe_mode_message() -> io::Result<()> {
+    const PIPE_NAME: &str = r"\\.\pipe\test-named-pipe-mode-message";
+
+    let server = ServerOptions::new()
+        .pipe_mode(PipeMode::Message)
+        .create(PIPE_NAME)?;
+
+    let _ = ClientOptions::new().open(PIPE_NAME)?;
+    server.connect().await?;
+    Ok(())
+}
+
+fn num_instances(pipe_name: impl AsRef<str>) -> io::Result<u32> {
+    use ntapi::ntioapi;
+    use winapi::shared::ntdef;
+
+    let mut name = pipe_name.as_ref().encode_utf16().collect::<Vec<_>>();
+    let mut name = ntdef::UNICODE_STRING {
+        Length: (name.len() * mem::size_of::<u16>()) as u16,
+        MaximumLength: (name.len() * mem::size_of::<u16>()) as u16,
+        Buffer: name.as_mut_ptr(),
+    };
+    let root = std::fs::File::open(r"\\.\Pipe\")?;
+    let mut io_status_block = unsafe { mem::zeroed() };
+    let mut file_directory_information = [0_u8; 1024];
+
+    let status = unsafe {
+        ntioapi::NtQueryDirectoryFile(
+            root.as_raw_handle(),
+            std::ptr::null_mut(),
+            None,
+            std::ptr::null_mut(),
+            &mut io_status_block,
+            &mut file_directory_information as *mut _ as *mut _,
+            1024,
+            ntioapi::FileDirectoryInformation,
+            0,
+            &mut name,
+            0,
+        )
+    };
+
+    if status as u32 != winerror::NO_ERROR {
+        return Err(io::Error::last_os_error());
+    }
+
+    let info = unsafe {
+        mem::transmute::<_, &ntioapi::FILE_DIRECTORY_INFORMATION>(&file_directory_information)
+    };
+    let raw_name = unsafe {
+        std::slice::from_raw_parts(
+            info.FileName.as_ptr(),
+            info.FileNameLength as usize / mem::size_of::<u16>(),
+        )
+    };
+    let name = String::from_utf16(raw_name).unwrap();
+    let num_instances = unsafe { *info.EndOfFile.QuadPart() };
+
+    assert_eq!(name, pipe_name.as_ref());
+
+    Ok(num_instances as u32)
+}

tokio/tests/rt_handle_block_on.rs

@@ -388,6 +388,28 @@ rt_test! {
 
         rt.block_on(async { some_non_async_function() });
     }
+
+    #[test]
+    fn spawn_after_runtime_dropped() {
+        use futures::future::FutureExt;
+
+        let rt = rt();
+
+        let handle = rt.block_on(async move {
+            Handle::current()
+        });
+
+        let jh1 = handle.spawn(futures::future::pending::<()>());
+
+        drop(rt);
+
+        let jh2 = handle.spawn(futures::future::pending::<()>());
+
+        let err1 = jh1.now_or_never().unwrap().unwrap_err();
+        let err2 = jh2.now_or_never().unwrap().unwrap_err();
+        assert!(err1.is_cancelled());
+        assert!(err2.is_cancelled());
+    }
 }
 
 multi_threaded_rt_test! {

tokio/tests/rt_threaded.rs

@@ -12,8 +12,8 @@ use std::future::Future;
 use std::pin::Pin;
 use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering::Relaxed;
-use std::sync::{mpsc, Arc};
-use std::task::{Context, Poll};
+use std::sync::{mpsc, Arc, Mutex};
+use std::task::{Context, Poll, Waker};
 
 #[test]
 fn single_thread() {
@@ -405,6 +405,74 @@ async fn hang_on_shutdown() {
     tokio::time::sleep(std::time::Duration::from_secs(1)).await;
 }
 
+/// Demonstrates tokio-rs/tokio#3869
+#[test]
+fn wake_during_shutdown() {
+    struct Shared {
+        waker: Option<Waker>,
+    }
+
+    struct MyFuture {
+        shared: Arc<Mutex<Shared>>,
+        put_waker: bool,
+    }
+
+    impl MyFuture {
+        fn new() -> (Self, Self) {
+            let shared = Arc::new(Mutex::new(Shared { waker: None }));
+            let f1 = MyFuture {
+                shared: shared.clone(),
+                put_waker: true,
+            };
+            let f2 = MyFuture {
+                shared,
+                put_waker: false,
+            };
+            (f1, f2)
+        }
+    }
+
+    impl Future for MyFuture {
+        type Output = ();
+
+        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
+            let me = Pin::into_inner(self);
+            let mut lock = me.shared.lock().unwrap();
+            println!("poll {}", me.put_waker);
+            if me.put_waker {
+                println!("putting");
+                lock.waker = Some(cx.waker().clone());
+            }
+            Poll::Pending
+        }
+    }
+
+    impl Drop for MyFuture {
+        fn drop(&mut self) {
+            println!("drop {} start", self.put_waker);
+            let mut lock = self.shared.lock().unwrap();
+            if !self.put_waker {
+                lock.waker.take().unwrap().wake();
+            }
+            drop(lock);
+            println!("drop {} stop", self.put_waker);
+        }
+    }
+
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(1)
+        .enable_all()
+        .build()
+        .unwrap();
+
+    let (f1, f2) = MyFuture::new();
+
+    rt.spawn(f1);
+    rt.spawn(f2);
+
+    rt.block_on(async { tokio::time::sleep(tokio::time::Duration::from_millis(20)).await });
+}
+
 fn rt() -> Runtime {
     Runtime::new().unwrap()
 }

tokio/tests/task_abort.rs

@@ -1,6 +1,7 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
 
+use std::sync::Arc;
 use std::thread::sleep;
 use std::time::Duration;
 
@@ -138,3 +139,97 @@ fn remote_abort_local_set_3929() {
     rt.block_on(local);
     jh2.join().unwrap();
 }
+
+/// Checks that a suspended task can be aborted even if the `JoinHandle` is immediately dropped.
+/// issue #3964: <https://github.com/tokio-rs/tokio/issues/3964>.
+#[test]
+fn test_abort_wakes_task_3964() {
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_time()
+        .build()
+        .unwrap();
+
+    rt.block_on(async move {
+        let notify_dropped = Arc::new(());
+        let weak_notify_dropped = Arc::downgrade(&notify_dropped);
+
+        let handle = tokio::spawn(async move {
+            // Make sure the Arc is moved into the task
+            let _notify_dropped = notify_dropped;
+            println!("task started");
+            tokio::time::sleep(std::time::Duration::new(100, 0)).await
+        });
+
+        // wait for task to sleep.
+        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+
+        handle.abort();
+        drop(handle);
+
+        // wait for task to abort.
+        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+
+        // Check that the Arc has been dropped.
+        assert!(weak_notify_dropped.upgrade().is_none());
+    });
+}
+
+struct PanicOnDrop;
+
+impl Drop for PanicOnDrop {
+    fn drop(&mut self) {
+        panic!("Well what did you expect would happen...");
+    }
+}
+
+/// Checks that aborting a task whose destructor panics does not allow the
+/// panic to escape the task.
+#[test]
+fn test_abort_task_that_panics_on_drop_contained() {
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_time()
+        .build()
+        .unwrap();
+
+    rt.block_on(async move {
+        let handle = tokio::spawn(async move {
+            // Make sure the Arc is moved into the task
+            let _panic_dropped = PanicOnDrop;
+            println!("task started");
+            tokio::time::sleep(std::time::Duration::new(100, 0)).await
+        });
+
+        // wait for task to sleep.
+        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+
+        handle.abort();
+        drop(handle);
+
+        // wait for task to abort.
+        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+    });
+}
+
+/// Checks that aborting a task whose destructor panics has the expected result.
+#[test]
+fn test_abort_task_that_panics_on_drop_returned() {
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_time()
+        .build()
+        .unwrap();
+
+    rt.block_on(async move {
+        let handle = tokio::spawn(async move {
+            // Make sure the Arc is moved into the task
+            let _panic_dropped = PanicOnDrop;
+            println!("task started");
+            tokio::time::sleep(std::time::Duration::new(100, 0)).await
+        });
+
+        // wait for task to sleep.
+        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+
+        handle.abort();
+        assert!(handle.await.unwrap_err().is_panic());
+    });
+}

tokio/tests/task_blocking.rs

@@ -132,7 +132,7 @@ fn useful_panic_message_when_dropping_rt_in_rt() {
     let err: &'static str = err.downcast_ref::<&'static str>().unwrap();
 
     assert!(
-        err.find("Cannot drop a runtime").is_some(),
+        err.contains("Cannot drop a runtime"),
         "Wrong panic message: {:?}",
         err
     );

tokio/tests/tcp_into_std.rs

@@ -10,10 +10,11 @@ use tokio::net::TcpStream;
 #[tokio::test]
 async fn tcp_into_std() -> Result<()> {
     let mut data = [0u8; 12];
-    let listener = TcpListener::bind("127.0.0.1:34254").await?;
+    let listener = TcpListener::bind("127.0.0.1:0").await?;
+    let addr = listener.local_addr().unwrap().to_string();
 
     let handle = tokio::spawn(async {
-        let stream: TcpStream = TcpStream::connect("127.0.0.1:34254").await.unwrap();
+        let stream: TcpStream = TcpStream::connect(addr).await.unwrap();
         stream
     });
 

tokio/tests/uds_cred.rs

@@ -8,10 +8,6 @@ use libc::getegid;
 use libc::geteuid;
 
 #[tokio::test]
-#[cfg_attr(
-    target_os = "freebsd",
-    ignore = "Requires FreeBSD 12.0 or later. https://bugs.freebsd.org/bugzilla/show_bug.cgi?id=176419"
-)]
 #[cfg_attr(
     target_os = "netbsd",
     ignore = "NetBSD does not support getpeereid() for sockets created by socketpair()"