chore: release v1.25.3 (#6223)

# 1.25.3 (December 17thm 2023)

### Fixed
- io: add budgeting to `tokio::runtime::io::registration::async_io` ([#6221])

[#6221]: https://github.com/tokio-rs/tokio/pull/6221

.circleci/config.yml

@@ -6,7 +6,7 @@ jobs:
     resource_class: arm.medium
     environment:
       # Change to pin rust version
-      RUST_STABLE: stable
+      RUST_STABLE: 1.67.1
     steps:
       - checkout
       - run:

.cirrus.yml

@@ -1,7 +1,9 @@
+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-13-1
 env:
-  RUST_STABLE: stable
+  RUST_STABLE: 1.67.1
   RUST_NIGHTLY: nightly-2022-10-25
   RUSTFLAGS: -D warnings
 
@@ -11,9 +13,8 @@ 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
+    - pkg install -y bash
     - curl https://sh.rustup.rs -sSf --output rustup.sh
     - sh rustup.sh -y --profile minimal --default-toolchain $RUST_STABLE
     - . $HOME/.cargo/env
@@ -26,12 +27,11 @@ 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
   setup_script:
-    - pkg install -y bash curl
+    - pkg install -y bash
     - curl https://sh.rustup.rs -sSf --output rustup.sh
     - sh rustup.sh -y --profile minimal --default-toolchain $RUST_NIGHTLY
     - . $HOME/.cargo/env
@@ -44,9 +44,8 @@ task:
 
 task:
   name: FreeBSD 32-bit
-  auto_cancellation: $CIRRUS_BRANCH != 'master' && $CIRRUS_BRANCH !=~ 'tokio-.*'
   setup_script:
-    - pkg install -y bash curl
+    - pkg install -y bash
     - curl https://sh.rustup.rs -sSf --output rustup.sh
     - sh rustup.sh -y --profile minimal --default-toolchain $RUST_STABLE
     - . $HOME/.cargo/env

.github/workflows/ci.yml

@@ -10,7 +10,7 @@ env:
   RUSTFLAGS: -Dwarnings
   RUST_BACKTRACE: 1
   # Change to specific Rust release to pin
-  rust_stable: stable
+  rust_stable: 1.67.1
   rust_nightly: nightly-2022-11-03
   rust_clippy: 1.65.0
   # When updating this, also update:
@@ -46,7 +46,7 @@ jobs:
       - cross-test
       - no-atomic-u64
       - features
-      - minrust
+#      - minrust
       - minimal-versions
       - fmt
       - clippy
@@ -222,7 +222,8 @@ jobs:
         uses: dtolnay/rust-toolchain@master
         with:
           toolchain: ${{ env.rust_nightly }}
-      - uses: Swatinem/rust-cache@v2
+          override: true
+      # - uses: Swatinem/rust-cache@v1 -> CI failure observed due to insufficient storage space
       - name: asan
         run: cargo test --workspace --all-features --target x86_64-unknown-linux-gnu --tests -- --test-threads 1
         env:
@@ -330,42 +331,31 @@ jobs:
           target: ${{ matrix.target }}
       - uses: Swatinem/rust-cache@v2
       - name: Install cargo-hack
-        uses: taiki-e/install-action@cargo-hack
-      - name: check --feature-powerset
-        run: cargo hack check --all --feature-powerset --depth 2 -Z avoid-dev-deps --keep-going
+        uses: taiki-e/install-action@v2 
+        with: 
+          tool: cargo-hack
+      - name: check --each-feature
+        run: cargo hack check --all --each-feature -Z avoid-dev-deps
       # Try with unstable feature flags
       - name: check --feature-powerset --unstable
         run: cargo hack check --all --feature-powerset --depth 2 -Z avoid-dev-deps --keep-going
         env:
           RUSTFLAGS: --cfg tokio_unstable -Dwarnings
 
-  minrust:
-    name: minrust
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v3
-      - name: Install Rust ${{ env.rust_min }}
-        uses: dtolnay/rust-toolchain@master
-        with:
-          toolchain: ${{ env.rust_min }}
-      - uses: Swatinem/rust-cache@v2
-      # First compile just the main tokio crate with minrust and newest version
-      # of all dependencies, then pin once_cell and compile the rest of the
-      # crates with the pinned once_cell version.
-      #
-      # This is necessary because tokio-util transitively depends on once_cell,
-      # which is not compatible with the current minrust after the 1.15.0
-      # release.
-      - name: "check -p tokio --all-features"
-        run: cargo check -p tokio --all-features
-        env:
-          RUSTFLAGS: "" # remove -Dwarnings
-      - name: "pin once_cell version"
-        run: cargo update -p once_cell --precise 1.14.0
-      - name: "check --workspace --all-features"
-        run: cargo check --workspace --all-features
-        env:
-          RUSTFLAGS: "" # remove -Dwarnings
+#  minrust:
+#    name: minrust
+#    runs-on: ubuntu-latest
+#    steps:
+#      - uses: actions/checkout@v3
+#      - name: Install Rust ${{ env.rust_min }}
+#        uses: dtolnay/rust-toolchain@master
+#        with:
+#          toolchain: ${{ env.rust_min }}
+#      - uses: Swatinem/rust-cache@v2
+#      - name: "check --workspace --all-features"
+#        run: cargo check --workspace --all-features
+#        env:
+#          RUSTFLAGS: "" # remove -Dwarnings
 
   minimal-versions:
     name: minimal-versions
@@ -378,7 +368,9 @@ jobs:
           toolchain: ${{ env.rust_nightly }}
       - uses: Swatinem/rust-cache@v2
       - name: Install cargo-hack
-        uses: taiki-e/install-action@cargo-hack
+        uses: taiki-e/install-action@v2 
+        with: 
+          tool: cargo-hack
       - name: "check --all-features -Z minimal-versions"
         run: |
           # Remove dev-dependencies from Cargo.toml to prevent the next `cargo update`
@@ -488,10 +480,10 @@ jobs:
           - macos-latest
     steps:
       - uses: actions/checkout@v3
-      - name: Install Rust ${{ env.rust_stable }}
+      - name: Install Rust 1.65.0
         uses: dtolnay/rust-toolchain@master
         with:
-          toolchain: ${{ env.rust_stable }}
+          toolchain: 1.65.0
       - uses: Swatinem/rust-cache@v2
       - name: Test hyper
         run: |
@@ -555,8 +547,9 @@ jobs:
 
       # Install dependencies
       - name: Install cargo-hack
-        uses: taiki-e/install-action@cargo-hack
-
+        uses: taiki-e/install-action@v2 
+        with: 
+          tool: cargo-hack
       - name: Install wasm32-wasi target
         run: rustup target add wasm32-wasi
 

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.25.3", features = ["full"] }
 ```
 Then, on your main.rs:
 

tokio-util/tests/time_delay_queue.rs

@@ -257,6 +257,10 @@ async fn reset_twice() {
 #[tokio::test]
 async fn repeatedly_reset_entry_inserted_as_expired() {
     time::pause();
+
+    // Instants before the start of the test seem to break in wasm.
+    time::sleep(ms(1000)).await;
+
     let mut queue = task::spawn(DelayQueue::new());
     let now = Instant::now();
 
@@ -556,6 +560,10 @@ async fn reset_later_after_slot_starts() {
 #[tokio::test]
 async fn reset_inserted_expired() {
     time::pause();
+
+    // Instants before the start of the test seem to break in wasm.
+    time::sleep(ms(1000)).await;
+
     let mut queue = task::spawn(DelayQueue::new());
     let now = Instant::now();
 

tokio/CHANGELOG.md

@@ -1,3 +1,30 @@
+# 1.25.3 (December 17th, 2023)
+
+### Fixed
+- io: add budgeting to `tokio::runtime::io::registration::async_io` ([#6221])
+
+[#6221]: https://github.com/tokio-rs/tokio/pull/6221
+
+# 1.25.2 (September 22, 2023)
+
+Forward ports 1.20.6 changes.
+
+### Changed
+
+- io: use `memchr` from `libc` ([#5960])
+
+[#5960]: https://github.com/tokio-rs/tokio/pull/5960
+
+# 1.25.1 (May 28, 2023)
+
+Forward ports 1.18.6 changes.
+
+### Fixed
+
+- deps: disable default features for mio ([#5728])
+
+[#5728]: https://github.com/tokio-rs/tokio/pull/5728
+
 # 1.25.0 (January 28, 2023)
 
 ### Fixed
@@ -334,6 +361,26 @@ wasm32-wasi target is given unstable support for the `net` feature.
 [#4956]: https://github.com/tokio-rs/tokio/pull/4956
 [#4959]: https://github.com/tokio-rs/tokio/pull/4959
 
+# 1.20.6 (September 22, 2023)
+
+This is a backport of a change from 1.27.0.
+
+### Changed
+
+- io: use `memchr` from `libc` ([#5960])
+
+[#5960]: https://github.com/tokio-rs/tokio/pull/5960
+
+# 1.20.5 (May 28, 2023)
+
+Forward ports 1.18.6 changes.
+
+### Fixed
+
+- deps: disable default features for mio ([#5728])
+
+[#5728]: https://github.com/tokio-rs/tokio/pull/5728
+
 # 1.20.4 (January 17, 2023)
 
 Forward ports 1.18.5 changes.
@@ -480,6 +527,14 @@ This release fixes a bug in `Notified::enable`. ([#4747])
 [#4729]: https://github.com/tokio-rs/tokio/pull/4729
 [#4739]: https://github.com/tokio-rs/tokio/pull/4739
 
+# 1.18.6 (May 28, 2023)
+
+### Fixed
+
+- deps: disable default features for mio ([#5728])
+
+[#5728]: https://github.com/tokio-rs/tokio/pull/5728
+
 # 1.18.5 (January 17, 2023)
 
 ### Fixed

tokio/Cargo.toml

@@ -6,7 +6,7 @@ name = "tokio"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v1.x.y" git tag.
-version = "1.25.0"
+version = "1.25.3"
 edition = "2018"
 rust-version = "1.49"
 authors = ["Tokio Contributors <[email protected]>"]
@@ -42,7 +42,7 @@ full = [
 ]
 
 fs = []
-io-util = ["memchr", "bytes"]
+io-util = ["bytes"]
 # stdin, stdout, stderr
 io-std = []
 macros = ["tokio-macros"]
@@ -103,8 +103,7 @@ pin-project-lite = "0.2.0"
 
 # Everything else is optional...
 bytes = { version = "1.0.0", optional = true }
-memchr = { version = "2.2", optional = true }
-mio = { version = "0.8.4", optional = true }
+mio = { version = "0.8.4", optional = true, default-features = false }
 num_cpus = { version = "1.8.0", optional = true }
 parking_lot = { version = "0.12.0", optional = true }
 
@@ -135,9 +134,6 @@ features = [
     "Win32_Security_Authorization",
 ]
 
-[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" }

tokio/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.25.3", features = ["full"] }
 ```
 Then, on your main.rs:
 

tokio/src/future/mod.rs

@@ -20,6 +20,7 @@ cfg_sync! {
 
 cfg_trace! {
     mod trace;
+    #[allow(unused_imports)]
     pub(crate) use trace::InstrumentedFuture as Future;
 }
 

tokio/src/io/util/read_until.rs

@@ -1,4 +1,5 @@
 use crate::io::AsyncBufRead;
+use crate::util::memchr;
 
 use pin_project_lite::pin_project;
 use std::future::Future;

tokio/src/runtime/context.rs

@@ -53,18 +53,18 @@ tokio_thread_local! {
             #[cfg(feature = "rt")]
             thread_id: Cell::new(None),
 
-            /// Tracks the current runtime handle to use when spawning,
-            /// accessing drivers, etc...
+            // Tracks the current runtime handle to use when spawning,
+            // accessing drivers, etc...
             #[cfg(feature = "rt")]
             handle: RefCell::new(None),
             #[cfg(feature = "rt")]
             current_task_id: Cell::new(None),
 
-            /// Tracks if the current thread is currently driving a runtime.
-            /// Note, that if this is set to "entered", the current scheduler
-            /// handle may not reference the runtime currently executing. This
-            /// is because other runtime handles may be set to current from
-            /// within a runtime.
+            // Tracks if the current thread is currently driving a runtime.
+            // Note, that if this is set to "entered", the current scheduler
+            // handle may not reference the runtime currently executing. This
+            // is because other runtime handles may be set to current from
+            // within a runtime.
             #[cfg(feature = "rt")]
             runtime: Cell::new(EnterRuntime::NotEntered),
 

tokio/src/runtime/io/registration.rs

@@ -239,11 +239,16 @@ cfg_io_readiness! {
             loop {
                 let event = self.readiness(interest).await?;
 
+                let coop = crate::future::poll_fn(crate::runtime::coop::poll_proceed).await;
+
                 match f() {
                     Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                         self.clear_readiness(event);
                     }
-                    x => return x,
+                    x => {
+                        coop.made_progress();
+                        return x
+                    },
                 }
             }
         }

tokio/src/util/memchr.rs

@@ -0,0 +1,74 @@
+//! Search for a byte in a byte array using libc.
+//!
+//! When nothing pulls in libc, then just use a trivial implementation. Note
+//! that we only depend on libc on unix.
+
+#[cfg(not(all(unix, feature = "libc")))]
+pub(crate) fn memchr(needle: u8, haystack: &[u8]) -> Option<usize> {
+    haystack.iter().position(|val| needle == *val)
+}
+
+#[cfg(all(unix, feature = "libc"))]
+pub(crate) fn memchr(needle: u8, haystack: &[u8]) -> Option<usize> {
+    let start = haystack.as_ptr();
+
+    // SAFETY: `start` is valid for `haystack.len()` bytes.
+    let ptr = unsafe { libc::memchr(start.cast(), needle as _, haystack.len()) };
+
+    if ptr.is_null() {
+        None
+    } else {
+        Some(ptr as usize - start as usize)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::memchr;
+
+    #[test]
+    fn memchr_test() {
+        let haystack = b"123abc456\0\xffabc\n";
+
+        assert_eq!(memchr(b'1', haystack), Some(0));
+        assert_eq!(memchr(b'2', haystack), Some(1));
+        assert_eq!(memchr(b'3', haystack), Some(2));
+        assert_eq!(memchr(b'4', haystack), Some(6));
+        assert_eq!(memchr(b'5', haystack), Some(7));
+        assert_eq!(memchr(b'6', haystack), Some(8));
+        assert_eq!(memchr(b'7', haystack), None);
+        assert_eq!(memchr(b'a', haystack), Some(3));
+        assert_eq!(memchr(b'b', haystack), Some(4));
+        assert_eq!(memchr(b'c', haystack), Some(5));
+        assert_eq!(memchr(b'd', haystack), None);
+        assert_eq!(memchr(b'A', haystack), None);
+        assert_eq!(memchr(0, haystack), Some(9));
+        assert_eq!(memchr(0xff, haystack), Some(10));
+        assert_eq!(memchr(0xfe, haystack), None);
+        assert_eq!(memchr(1, haystack), None);
+        assert_eq!(memchr(b'\n', haystack), Some(14));
+        assert_eq!(memchr(b'\r', haystack), None);
+    }
+
+    #[test]
+    fn memchr_all() {
+        let mut arr = Vec::new();
+        for b in 0..=255 {
+            arr.push(b);
+        }
+        for b in 0..=255 {
+            assert_eq!(memchr(b, &arr), Some(b as usize));
+        }
+        arr.reverse();
+        for b in 0..=255 {
+            assert_eq!(memchr(b, &arr), Some(255 - b as usize));
+        }
+    }
+
+    #[test]
+    fn memchr_empty() {
+        for b in 0..=255 {
+            assert_eq!(memchr(b, b""), None);
+        }
+    }
+}

tokio/src/util/mod.rs

@@ -76,3 +76,6 @@ cfg_rt_multi_thread! {
 pub(crate) mod trace;
 
 pub(crate) mod error;
+
+#[cfg(feature = "io-util")]
+pub(crate) mod memchr;

tokio/tests/coop_budger.rs

@@ -0,0 +1,77 @@
+#![warn(rust_2018_idioms)]
+#![cfg(all(feature = "full", target_os = "linux"))]
+
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+use tokio::net::UdpSocket;
+
+/// Ensure that UDP sockets have functional budgeting
+///
+/// # Design
+/// Two sockets communicate by spamming packets from one to the other.
+///
+/// In Linux, this packet will be slammed through the entire network stack and into the receiver's buffer during the
+/// send system call because we are using the loopback interface.
+/// This happens because the softirq chain invoked on send when using the loopback interface covers virtually the
+/// entirety of the lifecycle of a packet within the kernel network stack.
+///
+/// As a result, neither socket will ever encounter an EWOULDBLOCK, and the only way for these to yield during the loop
+/// is through budgeting.
+///
+/// A second task runs in the background and increments a counter before yielding, allowing us to know how many times sockets yielded.
+/// Since we are both sending and receiving, that should happen once per 64 packets, because budgets are of size 128
+/// and there are two budget events per packet, a send and a recv.
+#[tokio::test]
+async fn coop_budget_udp_send_recv() {
+    const BUDGET: usize = 128;
+    const N_ITERATIONS: usize = 1024;
+
+    const PACKET: &[u8] = b"Hello, world";
+    const PACKET_LEN: usize = 12;
+
+    assert_eq!(
+        PACKET_LEN,
+        PACKET.len(),
+        "Defect in test, programmer can't do math"
+    );
+
+    // bind each socket to a dynamic port, forcing IPv4 addressing on the localhost interface
+    let tx = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+    let rx = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+
+    tx.connect(rx.local_addr().unwrap()).await.unwrap();
+    rx.connect(tx.local_addr().unwrap()).await.unwrap();
+
+    let tracker = Arc::new(AtomicUsize::default());
+
+    let tracker_clone = Arc::clone(&tracker);
+
+    tokio::task::yield_now().await;
+
+    tokio::spawn(async move {
+        loop {
+            tracker_clone.fetch_add(1, Ordering::SeqCst);
+
+            tokio::task::yield_now().await;
+        }
+    });
+
+    for _ in 0..N_ITERATIONS {
+        tx.send(PACKET).await.unwrap();
+
+        let mut tmp = [0; PACKET_LEN];
+
+        // ensure that we aren't somehow accumulating other
+        assert_eq!(
+            PACKET_LEN,
+            rx.recv(&mut tmp).await.unwrap(),
+            "Defect in test case, received unexpected result from socket"
+        );
+        assert_eq!(
+            PACKET, &tmp,
+            "Defect in test case, received unexpected result from socket"
+        );
+    }
+
+    assert_eq!(N_ITERATIONS / (BUDGET / 2), tracker.load(Ordering::SeqCst));
+}

tokio/tests/net_named_pipe.rs

@@ -2,13 +2,11 @@
 #![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 windows_sys::Win32::Foundation::{ERROR_NO_DATA, ERROR_PIPE_BUSY, NO_ERROR, UNICODE_STRING};
+use windows_sys::Win32::Foundation::{ERROR_NO_DATA, ERROR_PIPE_BUSY};
 
 #[tokio::test]
 async fn test_named_pipe_client_drop() -> io::Result<()> {
@@ -16,8 +14,6 @@ async fn test_named_pipe_client_drop() -> io::Result<()> {
 
     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?;
@@ -367,53 +363,3 @@ async fn test_named_pipe_access() -> io::Result<()> {
     }
     Ok(())
 }
-
-fn num_instances(pipe_name: impl AsRef<str>) -> io::Result<u32> {
-    use ntapi::ntioapi;
-
-    let mut name = pipe_name.as_ref().encode_utf16().collect::<Vec<_>>();
-    let mut name = 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 as *mut _ as _,
-            0,
-        )
-    };
-
-    if status as u32 != 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)
-}