sync: move CancellationToken to tokio-util (#2721)

* sync: move CancellationToken to tokio-util

The `CancellationToken` utility is only available with the
`tokio_unstable` flag. This was done as the API is not final, but it
adds friction for users.

This patch moves `CancellationToken` to tokio-util where it is generally
available. The tokio-util crate does not have any constraints on
breaking change releases.

* fix clippy

* clippy again

tokio-util/Cargo.toml

@@ -46,6 +46,7 @@ tokio = { version = "0.3.0", path = "../tokio", features = ["full"] }
 tokio-test = { version = "0.3.0", path = "../tokio-test" }
 
 futures = "0.3.0"
+futures-test = "0.3.5"
 
 [package.metadata.docs.rs]
 all-features = true

tokio-util/src/lib.rs

@@ -24,6 +24,8 @@
 #[macro_use]
 mod cfg;
 
+mod loom;
+
 cfg_codec! {
     pub mod codec;
 }
@@ -35,3 +37,5 @@ cfg_udp! {
 cfg_compat! {
     pub mod compat;
 }
+
+pub mod sync;

tokio-util/src/loom.rs

@@ -0,0 +1 @@
+pub(crate) use std::sync;

tokio-util/src/sync/cancellation_token.rs

@@ -3,7 +3,7 @@
 
 use crate::loom::sync::atomic::AtomicUsize;
 use crate::loom::sync::Mutex;
-use crate::util::intrusive_double_linked_list::{LinkedList, ListNode};
+use crate::sync::intrusive_double_linked_list::{LinkedList, ListNode};
 
 use core::future::Future;
 use core::pin::Pin;
@@ -129,6 +129,12 @@ impl Drop for CancellationToken {
     }
 }
 
+impl Default for CancellationToken {
+    fn default() -> CancellationToken {
+        CancellationToken::new()
+    }
+}
+
 impl CancellationToken {
     /// Creates a new CancellationToken in the non-cancelled state.
     pub fn new() -> CancellationToken {

tokio-util/src/sync/mod.rs

@@ -0,0 +1,6 @@
+//! Synchronization primitives
+
+mod cancellation_token;
+pub use cancellation_token::{CancellationToken, WaitForCancellationFuture};
+
+mod intrusive_double_linked_list;

tokio-util/src/sync/tests/mod.rs

@@ -0,0 +1 @@
+

tokio-util/tests/sync_cancellation_token.rs

@@ -1,7 +1,5 @@
-#![cfg(tokio_unstable)]
-
 use tokio::pin;
-use tokio::sync::CancellationToken;
+use tokio_util::sync::CancellationToken;
 
 use core::future::Future;
 use core::task::{Context, Poll};
@@ -186,8 +184,8 @@ fn drop_multiple_child_tokens() {
     for drop_first_child_first in &[true, false] {
         let token = CancellationToken::new();
         let mut child_tokens = [None, None, None];
-        for i in 0..child_tokens.len() {
-            child_tokens[i] = Some(token.child_token());
+        for child in &mut child_tokens {
+            *child = Some(token.child_token());
         }
 
         assert!(!token.is_cancelled());

tokio/Cargo.toml

@@ -125,7 +125,6 @@ optional = true
 [dev-dependencies]
 tokio-test = { version = "0.3.0", path = "../tokio-test" }
 futures = { version = "0.3.0", features = ["async-await"] }
-futures-test = "0.3.0"
 proptest = "0.9.4"
 tempfile = "3.1.0"
 

tokio/src/macros/cfg.rs

@@ -354,16 +354,6 @@ macro_rules! cfg_uds {
     }
 }
 
-macro_rules! cfg_unstable {
-    ($($item:item)*) => {
-        $(
-            #[cfg(tokio_unstable)]
-            #[cfg_attr(docsrs, doc(cfg(tokio_unstable)))]
-            $item
-        )*
-    }
-}
-
 macro_rules! cfg_trace {
     ($($item:item)*) => {
         $(

tokio/src/sync/mod.rs

@@ -434,11 +434,6 @@ cfg_sync! {
 
     pub mod broadcast;
 
-    cfg_unstable! {
-        mod cancellation_token;
-        pub use cancellation_token::{CancellationToken, WaitForCancellationFuture};
-    }
-
     pub mod mpsc;
 
     mod mutex;

tokio/src/sync/tests/mod.rs

@@ -7,8 +7,6 @@ cfg_not_loom! {
 cfg_loom! {
     mod loom_atomic_waker;
     mod loom_broadcast;
-    #[cfg(tokio_unstable)]
-    mod loom_cancellation_token;
     mod loom_list;
     mod loom_mpsc;
     mod loom_notify;

tokio/src/util/mod.rs

@@ -24,5 +24,3 @@ pub(crate) mod trace;
 #[cfg(any(feature = "macros", feature = "stream"))]
 #[cfg_attr(not(feature = "macros"), allow(unreachable_pub))]
 pub use rand::thread_rng_n;
-
-pub(crate) mod intrusive_double_linked_list;

tokio/tests/async_send_sync.rs

@@ -259,6 +259,3 @@ async_assert_fn!(tokio::time::timeout_at(Instant, BoxFutureSync<()>): Send & Syn
 async_assert_fn!(tokio::time::timeout_at(Instant, BoxFutureSend<()>): Send & !Sync);
 async_assert_fn!(tokio::time::timeout_at(Instant, BoxFuture<()>): !Send & !Sync);
 async_assert_fn!(tokio::time::Interval::tick(_): Send & Sync);
-
-#[cfg(tokio_unstable)]
-assert_value!(tokio::sync::CancellationToken: Send & Sync);