Merge branch 'master' into noah/warnings

README.md

@@ -205,8 +205,8 @@ works with the MSRV of that minor release.
 
 ## 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.
+Tokio doesn't follow a fixed release schedule, but we typically make one minor
+release each month. We make patch releases for bugfixes as necessary.
 
 ## Bug patching policy
 
@@ -216,8 +216,8 @@ 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.36.x` - LTS release until March 2025. (MSRV 1.63)
  * `1.38.x` - LTS release until July 2025. (MSRV 1.63)
+ * `1.43.x` - LTS release until March 2026. (MSRV 1.70)
 
 Each LTS release will continue to receive backported fixes for at least a year.
 If you wish to use a fixed minor release in your project, we recommend that you
@@ -227,7 +227,7 @@ 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.32.x` patch release, you
 can use the following dependency specification:
 ```text
-tokio = { version = "~1.32", features = [...] }
+tokio = { version = "~1.38", features = [...] }
 ```
 
 ### Previous LTS releases
@@ -238,6 +238,7 @@ tokio = { version = "~1.32", features = [...] }
  * `1.20.x` - LTS release until September 2023.
  * `1.25.x` - LTS release until March 2024.
  * `1.32.x` - LTS release until September 2024.
+ * `1.36.x` - LTS release until March 2025.
 
 ## License
 

tokio-util/src/codec/any_delimiter_codec.rs

@@ -217,7 +217,7 @@ where
 
     fn encode(&mut self, chunk: T, buf: &mut BytesMut) -> Result<(), AnyDelimiterCodecError> {
         let chunk = chunk.as_ref();
-        buf.reserve(chunk.len() + 1);
+        buf.reserve(chunk.len() + self.sequence_writer.len());
         buf.put(chunk.as_bytes());
         buf.put(self.sequence_writer.as_ref());
 

tokio/README.md

@@ -205,8 +205,8 @@ works with the MSRV of that minor release.
 
 ## 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.
+Tokio doesn't follow a fixed release schedule, but we typically make one minor
+release each month. We make patch releases for bugfixes as necessary.
 
 ## Bug patching policy
 
@@ -216,8 +216,8 @@ 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.36.x` - LTS release until March 2025. (MSRV 1.63)
  * `1.38.x` - LTS release until July 2025. (MSRV 1.63)
+ * `1.43.x` - LTS release until March 2026. (MSRV 1.70)
 
 Each LTS release will continue to receive backported fixes for at least a year.
 If you wish to use a fixed minor release in your project, we recommend that you
@@ -227,7 +227,7 @@ 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.32.x` patch release, you
 can use the following dependency specification:
 ```text
-tokio = { version = "~1.32", features = [...] }
+tokio = { version = "~1.38", features = [...] }
 ```
 
 ### Previous LTS releases
@@ -238,6 +238,7 @@ tokio = { version = "~1.32", features = [...] }
  * `1.20.x` - LTS release until September 2023.
  * `1.25.x` - LTS release until March 2024.
  * `1.32.x` - LTS release until September 2024.
+ * `1.36.x` - LTS release until March 2025.
 
 ## License
 

tokio/src/process/mod.rs

@@ -1382,7 +1382,7 @@ impl Child {
 
 /// The standard input stream for spawned children.
 ///
-/// This type implements the `AsyncWrite` trait to pass data to the stdin handle of
+/// This type implements the `AsyncWrite` trait to pass data to the stdin
 /// handle of a child process asynchronously.
 #[derive(Debug)]
 pub struct ChildStdin {

tokio/src/task/coop/mod.rs

@@ -305,7 +305,7 @@ cfg_coop! {
 
                 Poll::Ready(restore)
             } else {
-                cx.waker().wake_by_ref();
+                defer(cx);
                 Poll::Pending
             }
         }).unwrap_or(Poll::Ready(RestoreOnPending(Cell::new(Budget::unconstrained()))))
@@ -325,11 +325,19 @@ cfg_coop! {
             #[inline(always)]
             fn inc_budget_forced_yield_count() {}
         }
+
+        fn defer(cx: &mut Context<'_>) {
+            context::defer(cx.waker());
+        }
     }
 
     cfg_not_rt! {
         #[inline(always)]
         fn inc_budget_forced_yield_count() {}
+
+        fn defer(cx: &mut Context<'_>) {
+            cx.waker().wake_by_ref();
+        }
     }
 
     impl Budget {

tokio/tests/coop_budget.rs

@@ -81,7 +81,7 @@ async fn coop_budget_udp_send_recv() {
 
 #[tokio::test]
 async fn test_has_budget_remaining() {
-    // At the begining budget should be available
+    // At the beginning, budget should be available.
     assert!(has_budget_remaining());
 
     // Deplete the budget

tokio/tests/rt_common.rs

@@ -745,7 +745,25 @@ rt_test! {
     #[cfg_attr(miri, ignore)] // No `socket` in miri.
     fn yield_defers_until_park() {
         for _ in 0..10 {
-            if yield_defers_until_park_inner() {
+            if yield_defers_until_park_inner(false) {
+                // test passed
+                return;
+            }
+
+            // Wait a bit and run the test again.
+            std::thread::sleep(std::time::Duration::from_secs(2));
+        }
+
+        panic!("yield_defers_until_park is failing consistently");
+    }
+
+    /// Same as above, but with cooperative scheduling.
+    #[test]
+    #[cfg(not(target_os="wasi"))]
+    #[cfg_attr(miri, ignore)] // No `socket` in miri.
+    fn coop_yield_defers_until_park() {
+        for _ in 0..10 {
+            if yield_defers_until_park_inner(true) {
                 // test passed
                 return;
             }
@@ -760,10 +778,12 @@ rt_test! {
     /// Implementation of `yield_defers_until_park` test. Returns `true` if the
     /// test passed.
     #[cfg(not(target_os="wasi"))]
-    fn yield_defers_until_park_inner() -> bool {
+    fn yield_defers_until_park_inner(use_coop: bool) -> bool {
         use std::sync::atomic::{AtomicBool, Ordering::SeqCst};
         use std::sync::Barrier;
 
+        const BUDGET: usize = 128;
+
         let rt = rt();
 
         let flag = Arc::new(AtomicBool::new(false));
@@ -802,7 +822,15 @@ rt_test! {
                         // Yield until connected
                         let mut cnt = 0;
                         while !flag_clone.load(SeqCst){
-                            tokio::task::yield_now().await;
+                            if use_coop {
+                                // Consume a good chunk of budget, which should
+                                // force at least one yield.
+                                for _ in 0..BUDGET {
+                                    tokio::task::consume_budget().await;
+                                }
+                            } else {
+                                tokio::task::yield_now().await;
+                            }
                             cnt += 1;
 
                             if cnt >= 10 {