Merge branch 'master' and 'merge-1.8.3'

.github/workflows/ci.yml

@@ -9,7 +9,7 @@ name: CI
 env:
   RUSTFLAGS: -Dwarnings
   RUST_BACKTRACE: 1
-  nightly: nightly-2021-04-25
+  nightly: nightly-2021-07-09
   minrust: 1.45.2
 
 jobs:

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.8.0", features = ["full"] }
+tokio = { version = "1.9.0", features = ["full"] }
 ```
 Then, on your main.rs:
 
@@ -140,8 +140,7 @@ several other libraries, including:
 
 * [`tower`]: A library of modular and reusable components for building robust networking clients and servers.
 
-* [`tracing`] (formerly `tokio-trace`): A framework for application-level
-  tracing and async-aware diagnostics.
+* [`tracing`] (formerly `tokio-trace`): A framework for application-level tracing and async-aware diagnostics.
 
 * [`rdbc`]: A Rust database connectivity library for MySQL, Postgres and SQLite.
 
@@ -164,9 +163,35 @@ several other libraries, including:
 
 ## Supported Rust Versions
 
-Tokio is built against the latest stable release. The minimum supported version is 1.45.
-The current Tokio version is not guaranteed to build on Rust versions earlier than the
-minimum supported version.
+Tokio is built against the latest stable release. The minimum supported version
+is 1.45.  The current Tokio version is not guaranteed to build on Rust versions
+earlier than the minimum supported version.
+
+## Release schedule
+
+Tokio doesn't follow a fixed release schedule, but we typically make one to two
+new minor releases each month. We make patch releases for bugfixes as necessary.
+
+## Bug patching policy
+
+For the purposes of making patch releases with bugfixes, we have designated
+certain minor releases as LTS (long term support) releases. Whenever a bug
+warrants a patch release with a fix for the bug, it will be backported and
+released as a new patch release for each LTS minor version. Our current LTS
+releases are:
+
+ * `1.8.x` - LTS release until February 2022.
+
+Each LTS release will continue to receive backported fixes for at least half a
+year. If you wish to use a fixed minor release in your project, we recommend
+that you use an LTS release.
+
+To use a fixed minor version, you can specify the version with a tilde. For
+example, to specify that you wish to use the newest `1.8.x` patch release, you
+can use the following dependency specification:
+```text
+tokio = { version = "~1.8", features = [...] }
+```
 
 ## License
 

benches/spawn.rs

@@ -2,63 +2,83 @@
 //! This essentially measure the time to enqueue a task in the local and remote
 //! case.
 
+#[macro_use]
+extern crate bencher;
+
 use bencher::{black_box, Bencher};
 
 async fn work() -> usize {
     let val = 1 + 1;
+    tokio::task::yield_now().await;
     black_box(val)
 }
 
-fn basic_scheduler_local_spawn(bench: &mut Bencher) {
+fn basic_scheduler_spawn(bench: &mut Bencher) {
     let runtime = tokio::runtime::Builder::new_current_thread()
         .build()
         .unwrap();
-    runtime.block_on(async {
-        bench.iter(|| {
+    bench.iter(|| {
+        runtime.block_on(async {
             let h = tokio::spawn(work());
-            black_box(h);
-        })
+            assert_eq!(h.await.unwrap(), 2);
+        });
     });
 }
 
-fn threaded_scheduler_local_spawn(bench: &mut Bencher) {
+fn basic_scheduler_spawn10(bench: &mut Bencher) {
     let runtime = tokio::runtime::Builder::new_current_thread()
         .build()
         .unwrap();
-    runtime.block_on(async {
-        bench.iter(|| {
-            let h = tokio::spawn(work());
-            black_box(h);
-        })
+    bench.iter(|| {
+        runtime.block_on(async {
+            let mut handles = Vec::with_capacity(10);
+            for _ in 0..10 {
+                handles.push(tokio::spawn(work()));
+            }
+            for handle in handles {
+                assert_eq!(handle.await.unwrap(), 2);
+            }
+        });
     });
 }
 
-fn basic_scheduler_remote_spawn(bench: &mut Bencher) {
-    let runtime = tokio::runtime::Builder::new_current_thread()
+fn threaded_scheduler_spawn(bench: &mut Bencher) {
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(1)
         .build()
         .unwrap();
-
     bench.iter(|| {
-        let h = runtime.spawn(work());
-        black_box(h);
+        runtime.block_on(async {
+            let h = tokio::spawn(work());
+            assert_eq!(h.await.unwrap(), 2);
+        });
     });
 }
 
-fn threaded_scheduler_remote_spawn(bench: &mut Bencher) {
-    let runtime = tokio::runtime::Builder::new_multi_thread().build().unwrap();
-
+fn threaded_scheduler_spawn10(bench: &mut Bencher) {
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(1)
+        .build()
+        .unwrap();
     bench.iter(|| {
-        let h = runtime.spawn(work());
-        black_box(h);
+        runtime.block_on(async {
+            let mut handles = Vec::with_capacity(10);
+            for _ in 0..10 {
+                handles.push(tokio::spawn(work()));
+            }
+            for handle in handles {
+                assert_eq!(handle.await.unwrap(), 2);
+            }
+        });
     });
 }
 
 bencher::benchmark_group!(
     spawn,
-    basic_scheduler_local_spawn,
-    threaded_scheduler_local_spawn,
-    basic_scheduler_remote_spawn,
-    threaded_scheduler_remote_spawn
+    basic_scheduler_spawn,
+    basic_scheduler_spawn10,
+    threaded_scheduler_spawn,
+    threaded_scheduler_spawn10,
 );
 
 bencher::benchmark_main!(spawn);

tokio-macros/CHANGELOG.md

@@ -1,3 +1,9 @@
+# 1.3.0 (July 7, 2021)
+
+- macros: don't trigger `clippy::unwrap_used` ([#3926])
+
+[#3926]: https://github.com/tokio-rs/tokio/pull/3926
+
 # 1.2.0 (May 14, 2021)
 
 - macros: forward input arguments in `#[tokio::test]` ([#3691])

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.2.0"
+version = "1.3.0"
 edition = "2018"
 authors = ["Tokio Contributors <team@tokio.rs>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
-documentation = "https://docs.rs/tokio-macros/1.2.0/tokio_macros"
+documentation = "https://docs.rs/tokio-macros/1.3.0/tokio_macros"
 description = """
 Tokio's proc macros.
 """

tokio-macros/src/entry.rs

@@ -201,12 +201,15 @@ fn parse_knobs(
     for arg in args {
         match arg {
             syn::NestedMeta::Meta(syn::Meta::NameValue(namevalue)) => {
-                let ident = namevalue.path.get_ident();
-                if ident.is_none() {
-                    let msg = "Must have specified ident";
-                    return Err(syn::Error::new_spanned(namevalue, msg));
-                }
-                match ident.unwrap().to_string().to_lowercase().as_str() {
+                let ident = namevalue
+                    .path
+                    .get_ident()
+                    .ok_or_else(|| {
+                        syn::Error::new_spanned(&namevalue, "Must have specified ident")
+                    })?
+                    .to_string()
+                    .to_lowercase();
+                match ident.as_str() {
                     "worker_threads" => {
                         config.set_worker_threads(
                             namevalue.lit.clone(),
@@ -239,12 +242,11 @@ fn parse_knobs(
                 }
             }
             syn::NestedMeta::Meta(syn::Meta::Path(path)) => {
-                let ident = path.get_ident();
-                if ident.is_none() {
-                    let msg = "Must have specified ident";
-                    return Err(syn::Error::new_spanned(path, msg));
-                }
-                let name = ident.unwrap().to_string().to_lowercase();
+                let name = path
+                    .get_ident()
+                    .ok_or_else(|| syn::Error::new_spanned(&path, "Must have specified ident"))?
+                    .to_string()
+                    .to_lowercase();
                 let msg = match name.as_str() {
                     "threaded_scheduler" | "multi_thread" => {
                         format!(
@@ -326,11 +328,11 @@ fn parse_knobs(
             #rt
                 .enable_all()
                 .build()
-                .unwrap()
+                .expect("Failed building the Runtime")
                 .block_on(async #body)
         }
     })
-    .unwrap();
+    .expect("Parsing failure");
     input.block.brace_token = brace_token;
 
     let result = quote! {

tokio-macros/src/lib.rs

@@ -5,7 +5,7 @@
     rust_2018_idioms,
     unreachable_pub
 )]
-#![cfg_attr(docsrs, deny(broken_intra_doc_links))]
+#![cfg_attr(docsrs, deny(rustdoc::broken_intra_doc_links))]
 #![doc(test(
     no_crate_inject,
     attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables))

tokio-stream/CHANGELOG.md

@@ -1,3 +1,13 @@
+# 0.1.7 (July 7, 2021)
+
+### Fixed
+
+- sync: fix watch wrapper ([#3914])
+- time: fix `Timeout::size_hint` ([#3902])
+
+[#3902]: https://github.com/tokio-rs/tokio/pull/3902
+[#3914]: https://github.com/tokio-rs/tokio/pull/3914
+
 # 0.1.6 (May 14, 2021)
 
 ### Added

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.6"
+version = "0.1.7"
 edition = "2018"
 authors = ["Tokio Contributors <team@tokio.rs>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
-documentation = "https://docs.rs/tokio-stream/0.1.6/tokio_stream"
+documentation = "https://docs.rs/tokio-stream/0.1.7/tokio_stream"
 description = """
 Utilities to work with `Stream` and `tokio`.
 """
@@ -30,7 +30,7 @@ signal = ["tokio/signal"]
 [dependencies]
 futures-core = { version = "0.3.0" }
 pin-project-lite = "0.2.0"
-tokio = { version = "1.2.0", path = "../tokio", features = ["sync"] }
+tokio = { version = "1.8.0", path = "../tokio", features = ["sync"] }
 tokio-util = { version = "0.6.3", path = "../tokio-util", optional = true }
 
 [dev-dependencies]

tokio-stream/src/lib.rs

@@ -10,7 +10,7 @@
     unreachable_pub
 )]
 #![cfg_attr(docsrs, feature(doc_cfg))]
-#![cfg_attr(docsrs, deny(broken_intra_doc_links))]
+#![cfg_attr(docsrs, deny(rustdoc::broken_intra_doc_links))]
 #![doc(test(
     no_crate_inject,
     attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables))

tokio-stream/src/stream_ext/collect.rs

@@ -113,7 +113,7 @@ impl<T: AsRef<str>> sealed::FromStreamPriv<T> for String {
     }
 
     fn finalize(_: sealed::Internal, collection: &mut String) -> String {
-        mem::replace(collection, String::new())
+        mem::take(collection)
     }
 }
 
@@ -132,7 +132,7 @@ impl<T> sealed::FromStreamPriv<T> for Vec<T> {
     }
 
     fn finalize(_: sealed::Internal, collection: &mut Vec<T>) -> Vec<T> {
-        mem::replace(collection, vec![])
+        mem::take(collection)
     }
 }
 

tokio-stream/src/stream_map.rs

@@ -364,11 +364,11 @@ impl<K, V> StreamMap<K, V> {
     /// # Examples
     ///
     /// ```
-    /// use std::collections::HashMap;
+    /// use tokio_stream::{StreamMap, pending};
     ///
-    /// let mut a = HashMap::new();
+    /// let mut a = StreamMap::new();
     /// assert!(a.is_empty());
-    /// a.insert(1, "a");
+    /// a.insert(1, pending::<i32>());
     /// assert!(!a.is_empty());
     /// ```
     pub fn is_empty(&self) -> bool {

tokio-stream/src/wrappers/watch.rs

@@ -72,10 +72,10 @@ impl<T: Clone + 'static + Send + Sync> Stream for WatchStream<T> {
     type Item = T;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let (result, rx) = ready!(self.inner.poll(cx));
+        let (result, mut rx) = ready!(self.inner.poll(cx));
         match result {
             Ok(_) => {
-                let received = (*rx.borrow()).clone();
+                let received = (*rx.borrow_and_update()).clone();
                 self.inner.set(make_future(rx));
                 Poll::Ready(Some(received))
             }

tokio-stream/tests/watch.rs

@@ -0,0 +1,27 @@
+use tokio::sync::watch;
+use tokio_stream::wrappers::WatchStream;
+use tokio_stream::StreamExt;
+
+#[tokio::test]
+async fn message_not_twice() {
+    let (tx, rx) = watch::channel("hello");
+
+    let mut counter = 0;
+    let mut stream = WatchStream::new(rx).map(move |payload| {
+        println!("{}", payload);
+        if payload == "goodbye" {
+            counter += 1;
+        }
+        if counter >= 2 {
+            panic!("too many goodbyes");
+        }
+    });
+
+    let task = tokio::spawn(async move { while stream.next().await.is_some() {} });
+
+    // Send goodbye just once
+    tx.send("goodbye").unwrap();
+
+    drop(tx);
+    task.await.unwrap();
+}

tokio-test/src/lib.rs

@@ -4,7 +4,7 @@
     rust_2018_idioms,
     unreachable_pub
 )]
-#![cfg_attr(docsrs, deny(broken_intra_doc_links))]
+#![cfg_attr(docsrs, deny(rustdoc::broken_intra_doc_links))]
 #![doc(test(
     no_crate_inject,
     attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables))