Merge tag 'tokio-1.26.0' into solid-rs/1.x

.cirrus.yml

@@ -1,5 +1,7 @@
+only_if: $CIRRUS_TAG == '' && ($CIRRUS_PR != '' || $CIRRUS_BRANCH == 'master' || $CIRRUS_BRANCH =~ 'tokio-.*')
+auto_cancellation: $CIRRUS_BRANCH != 'master' && $CIRRUS_BRANCH !=~ 'tokio-.*'
 freebsd_instance:
-  image: freebsd-12-4-release-amd64
+  image_family: freebsd-12-4
 env:
   RUST_STABLE: stable
   RUST_NIGHTLY: nightly-2022-10-25
@@ -11,7 +13,6 @@ env:
 # the system's binaries, so the environment shouldn't matter.
 task:
   name: FreeBSD 64-bit
-  auto_cancellation: $CIRRUS_BRANCH != 'master' && $CIRRUS_BRANCH !=~ 'tokio-.*'
   setup_script:
     - pkg install -y bash curl
     - curl https://sh.rustup.rs -sSf --output rustup.sh
@@ -26,7 +27,6 @@ task:
 
 task:
   name: FreeBSD docs
-  auto_cancellation: $CIRRUS_BRANCH != 'master' && $CIRRUS_BRANCH !=~ 'tokio-.*'
   env:
     RUSTFLAGS: --cfg docsrs --cfg tokio_unstable
     RUSTDOCFLAGS: --cfg docsrs --cfg tokio_unstable -Dwarnings
@@ -44,7 +44,6 @@ task:
 
 task:
   name: FreeBSD 32-bit
-  auto_cancellation: $CIRRUS_BRANCH != 'master' && $CIRRUS_BRANCH !=~ 'tokio-.*'
   setup_script:
     - pkg install -y bash curl
     - curl https://sh.rustup.rs -sSf --output rustup.sh

.github/workflows/ci.yml

@@ -42,6 +42,7 @@ jobs:
       - test-unstable
       - miri
       - asan
+      - semver
       - cross-check
       - cross-test
       - no-atomic-u64
@@ -208,7 +209,6 @@ jobs:
         working-directory: tokio
         env:
           MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-strict-provenance -Zmiri-retag-fields
-          PROPTEST_CASES: 10
 
   asan:
     name: asan
@@ -230,6 +230,23 @@ jobs:
           # Ignore `trybuild` errors as they are irrelevant and flaky on nightly
           TRYBUILD: overwrite
 
+  semver:
+    name: semver
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Install Rust ${{ env.rust_stable }}
+        uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: ${{ env.rust_stable }}
+      - name: Install cargo-semver-checks
+        uses: taiki-e/install-action@v2
+        with:
+          tool: cargo-semver-checks
+      - name: Check semver compatibility
+        run: |
+          cargo semver-checks check-release --release-type minor
+
   cross-check:
     name: cross-check
     runs-on: ubuntu-latest

CONTRIBUTING.md

@@ -187,7 +187,7 @@ LOOM_MAX_PREEMPTIONS=1 RUSTFLAGS="--cfg loom" \
 
 You can run miri tests with
 ```
-MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-tag-raw-pointers" PROPTEST_CASES=10 \
+MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-tag-raw-pointers" \
     cargo +nightly miri test --features full --lib
 ```
 
@@ -209,6 +209,31 @@ utilities available to use in tests, no matter the crate being tested.
 The best strategy for writing a new integration test is to look at existing
 integration tests in the crate and follow the style.
 
+#### Fuzz tests
+
+Some of our crates include a set of fuzz tests, this will be marked by a
+directory `fuzz`. It is a good idea to run fuzz tests after each change.
+To get started with fuzz testing you'll need to install 
+[cargo-fuzz](https://github.com/rust-fuzz/cargo-fuzz).
+
+`cargo install cargo-fuzz`
+
+To list the available fuzzing harnesses you can run;
+
+```bash
+$ cd tokio
+$ cargo fuzz list
+fuzz_linked_list
+````
+
+Running a fuzz test is as simple as;
+
+`cargo fuzz run fuzz_linked_list`
+
+**NOTE**: Keep in mind that by default when running a fuzz test the fuzz
+harness will run forever and will only exit if you `ctrl-c` or it finds
+a bug.
+
 #### Documentation tests
 
 Ideally, every API has at least one [documentation test] that demonstrates how to

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.25.0", features = ["full"] }
+tokio = { version = "1.26.0", features = ["full"] }
 ```
 Then, on your main.rs:
 

benches/Cargo.toml

@@ -4,6 +4,9 @@ version = "0.0.0"
 publish = false
 edition = "2018"
 
+[features]
+test-util = ["tokio/test-util"]
+
 [dependencies]
 tokio = { version = "1.5.0", path = "../tokio", features = ["full"] }
 bencher = "0.1.5"
@@ -27,6 +30,16 @@ name = "sync_mpsc"
 path = "sync_mpsc.rs"
 harness = false
 
+[[bench]]
+name = "sync_mpsc_oneshot"
+path = "sync_mpsc_oneshot.rs"
+harness = false
+
+[[bench]]
+name = "sync_watch"
+path = "sync_watch.rs"
+harness = false
+
 [[bench]]
 name = "rt_multi_threaded"
 path = "rt_multi_threaded.rs"
@@ -57,3 +70,8 @@ harness = false
 name = "copy"
 path = "copy.rs"
 harness = false
+
+[[bench]]
+name = "time_now"
+path = "time_now.rs"
+harness = false

benches/sync_mpsc_oneshot.rs

@@ -0,0 +1,53 @@
+use bencher::{benchmark_group, benchmark_main, Bencher};
+use tokio::{
+    runtime::Runtime,
+    sync::{mpsc, oneshot},
+};
+
+fn request_reply_current_thread(b: &mut Bencher) {
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .build()
+        .unwrap();
+
+    request_reply(b, rt);
+}
+
+fn request_reply_multi_threaded(b: &mut Bencher) {
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(1)
+        .build()
+        .unwrap();
+
+    request_reply(b, rt);
+}
+
+fn request_reply(b: &mut Bencher, rt: Runtime) {
+    let tx = rt.block_on(async move {
+        let (tx, mut rx) = mpsc::channel::<oneshot::Sender<()>>(10);
+        tokio::spawn(async move {
+            while let Some(reply) = rx.recv().await {
+                reply.send(()).unwrap();
+            }
+        });
+        tx
+    });
+
+    b.iter(|| {
+        let task_tx = tx.clone();
+        rt.block_on(async move {
+            for _ in 0..1_000 {
+                let (o_tx, o_rx) = oneshot::channel();
+                task_tx.send(o_tx).await.unwrap();
+                let _ = o_rx.await;
+            }
+        })
+    });
+}
+
+benchmark_group!(
+    sync_mpsc_oneshot_group,
+    request_reply_current_thread,
+    request_reply_multi_threaded,
+);
+
+benchmark_main!(sync_mpsc_oneshot_group);

benches/sync_watch.rs

@@ -0,0 +1,64 @@
+use bencher::{black_box, Bencher};
+use rand::prelude::*;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
+use tokio::sync::{watch, Notify};
+
+fn rt() -> tokio::runtime::Runtime {
+    tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(6)
+        .build()
+        .unwrap()
+}
+
+fn do_work(rng: &mut impl RngCore) -> u32 {
+    use std::fmt::Write;
+    let mut message = String::new();
+    for i in 1..=10 {
+        let _ = write!(&mut message, " {i}={}", rng.gen::<f64>());
+    }
+    message
+        .as_bytes()
+        .iter()
+        .map(|&c| c as u32)
+        .fold(0, u32::wrapping_add)
+}
+
+fn contention_resubscribe(b: &mut Bencher) {
+    const NTASK: u64 = 1000;
+
+    let rt = rt();
+    let (snd, rcv) = watch::channel(0i32);
+    let wg = Arc::new((AtomicU64::new(0), Notify::new()));
+    for n in 0..NTASK {
+        let mut rcv = rcv.clone();
+        let wg = wg.clone();
+        let mut rng = rand::rngs::StdRng::seed_from_u64(n);
+        rt.spawn(async move {
+            while rcv.changed().await.is_ok() {
+                let _ = *rcv.borrow(); // contend on rwlock
+                let r = do_work(&mut rng);
+                let _ = black_box(r);
+                if wg.0.fetch_sub(1, Ordering::Release) == 1 {
+                    wg.1.notify_one();
+                }
+            }
+        });
+    }
+
+    b.iter(|| {
+        rt.block_on(async {
+            for _ in 0..100 {
+                assert_eq!(wg.0.fetch_add(NTASK, Ordering::Relaxed), 0);
+                let _ = snd.send(black_box(42));
+                while wg.0.load(Ordering::Acquire) > 0 {
+                    wg.1.notified().await;
+                }
+            }
+        });
+    });
+}
+
+bencher::benchmark_group!(contention, contention_resubscribe);
+
+bencher::benchmark_main!(contention);

benches/time_now.rs

@@ -0,0 +1,25 @@
+//! Benchmark spawning a task onto the basic and threaded Tokio executors.
+//! 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};
+
+fn time_now_current_thread(bench: &mut Bencher) {
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_time()
+        .build()
+        .unwrap();
+
+    bench.iter(|| {
+        rt.block_on(async {
+            black_box(tokio::time::Instant::now());
+        })
+    })
+}
+
+bencher::benchmark_group!(time_now, time_now_current_thread,);
+
+bencher::benchmark_main!(time_now);

examples/Cargo.toml

@@ -25,7 +25,7 @@ once_cell = "1.5.2"
 rand = "0.8.3"
 
 [target.'cfg(windows)'.dev-dependencies.windows-sys]
-version = "0.42.0"
+version = "0.45"
 
 [[example]]
 name = "chat"

tokio-macros/src/lib.rs

@@ -39,6 +39,13 @@ use proc_macro::TokenStream;
 /// function is called often, it is preferable to create the runtime using the
 /// runtime builder so the runtime can be reused across calls.
 ///
+/// # Non-worker async function
+///
+/// Note that the async function marked with this macro does not run as a
+/// worker. The expectation is that other tasks are spawned by the function here.
+/// Awaiting on other futures from the function provided here will not
+/// perform as fast as those spawned as workers.
+///
 /// # Multi-threaded runtime
 ///
 /// To use the multi-threaded runtime, the macro can be configured using

tokio-stream/CHANGELOG.md

@@ -1,3 +1,13 @@
+# 0.1.12 (January 20, 2023)
+
+- time: remove `Unpin` bound on `Throttle` methods ([#5105])
+- time: document that `throttle` operates on ms granularity ([#5101])
+- sync: add `WatchStream::from_changes` ([#5432])
+
+[#5105]: https://github.com/tokio-rs/tokio/pull/5105
+[#5101]: https://github.com/tokio-rs/tokio/pull/5101
+[#5432]: https://github.com/tokio-rs/tokio/pull/5432
+
 # 0.1.11 (October 11, 2022)
 
 - time: allow `StreamExt::chunks_timeout` outside of a runtime ([#5036])

tokio-stream/Cargo.toml

@@ -4,7 +4,7 @@ name = "tokio-stream"
 # - Remove path dependencies
 # - Update CHANGELOG.md.
 # - Create "tokio-stream-0.1.x" git tag.
-version = "0.1.11"
+version = "0.1.12"
 edition = "2018"
 rust-version = "1.49"
 authors = ["Tokio Contributors <team@tokio.rs>"]
@@ -38,9 +38,6 @@ parking_lot = "0.12.0"
 tokio-test = { path = "../tokio-test" }
 futures = { version = "0.3", default-features = false }
 
-[target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies]
-proptest = "1"
-
 [package.metadata.docs.rs]
 all-features = true
 rustdoc-args = ["--cfg", "docsrs"]

tokio-stream/fuzz/.gitignore

@@ -0,0 +1,4 @@
+target
+corpus
+artifacts
+coverage

tokio-stream/fuzz/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "tokio-stream-fuzz"
+version = "0.0.0"
+publish = false
+edition = "2018"
+
+[package.metadata]
+cargo-fuzz = true
+
+[dependencies]
+libfuzzer-sys = "0.4"
+tokio-test = { path = "../../tokio-test" }
+
+[dependencies.tokio-stream]
+path = ".."
+
+
+# Prevent this from interfering with workspaces
+[workspace]
+members = ["."]
+
+[profile.release]
+debug = 1
+
+[[bin]]
+name = "fuzz_stream_map"
+path = "fuzz_targets/fuzz_stream_map.rs"
+test = false
+doc = false