From b8af5aad16baec0d2713453aa03db5ff29639010 Mon Sep 17 00:00:00 2001 From: tim gretler Date: Sat, 24 Jun 2023 12:13:56 +0200 Subject: [PATCH 1/4] task: add spawn_blocking methods to JoinMap (#5797) --- tokio-util/Cargo.toml | 2 +- tokio-util/src/task/join_map.rs | 54 +++++++++++++++++++++++++++++++++ 2 files changed, 55 insertions(+), 1 deletion(-) diff --git a/tokio-util/Cargo.toml b/tokio-util/Cargo.toml index 4b406d818..f16099d78 100644 --- a/tokio-util/Cargo.toml +++ b/tokio-util/Cargo.toml @@ -34,7 +34,7 @@ rt = ["tokio/rt", "tokio/sync", "futures-util", "hashbrown"] __docs_rs = ["futures-util"] [dependencies] -tokio = { version = "1.22.0", path = "../tokio", features = ["sync"] } +tokio = { version = "1.28.0", path = "../tokio", features = ["sync"] } bytes = "1.0.0" futures-core = "0.3.0" futures-sink = "0.3.0" diff --git a/tokio-util/src/task/join_map.rs b/tokio-util/src/task/join_map.rs index c6bf5bc24..c9aed537d 100644 --- a/tokio-util/src/task/join_map.rs +++ b/tokio-util/src/task/join_map.rs @@ -316,6 +316,60 @@ where self.insert(key, task); } + /// Spawn the blocking code on the blocking threadpool and store it in this `JoinMap` with the provided + /// key. + /// + /// If a task previously existed in the `JoinMap` for this key, that task + /// will be cancelled and replaced with the new one. The previous task will + /// be removed from the `JoinMap`; a subsequent call to [`join_next`] will + /// *not* return a cancelled [`JoinError`] for that task. + /// + /// Note that blocking tasks cannot be cancelled after execution starts. + /// Replaced blocking tasks will still run to completion if the task has begun + /// to execute when it is replaced. A blocking task which is replaced before + /// it has been scheduled on a blocking worker thread will be cancelled. + /// + /// # Panics + /// + /// This method panics if called outside of a Tokio runtime. + /// + /// [`join_next`]: Self::join_next + #[track_caller] + pub fn spawn_blocking(&mut self, key: K, f: F) + where + F: FnOnce() -> V, + F: Send + 'static, + V: Send, + { + let task = self.tasks.spawn_blocking(f); + self.insert(key, task) + } + + /// Spawn the blocking code on the blocking threadpool of the provided runtime and store it in this + /// `JoinMap` with the provided key. + /// + /// If a task previously existed in the `JoinMap` for this key, that task + /// will be cancelled and replaced with the new one. The previous task will + /// be removed from the `JoinMap`; a subsequent call to [`join_next`] will + /// *not* return a cancelled [`JoinError`] for that task. + /// + /// Note that blocking tasks cannot be cancelled after execution starts. + /// Replaced blocking tasks will still run to completion if the task has begun + /// to execute when it is replaced. A blocking task which is replaced before + /// it has been scheduled on a blocking worker thread will be cancelled. + /// + /// [`join_next`]: Self::join_next + #[track_caller] + pub fn spawn_blocking_on(&mut self, key: K, f: F, handle: &Handle) + where + F: FnOnce() -> V, + F: Send + 'static, + V: Send, + { + let task = self.tasks.spawn_blocking_on(f, handle); + self.insert(key, task); + } + /// Spawn the provided task on the current [`LocalSet`] and store it in this /// `JoinMap` with the provided key. /// From 78bf8a9e5e72430274467cebce4cc37096508c18 Mon Sep 17 00:00:00 2001 From: wjjiang Date: Sun, 25 Jun 2023 16:40:49 +0800 Subject: [PATCH 2/4] sync: replace Poll::Ready with Ready (#5815) --- tokio/src/sync/oneshot.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tokio/src/sync/oneshot.rs b/tokio/src/sync/oneshot.rs index 0ff3ea281..af3cc854f 100644 --- a/tokio/src/sync/oneshot.rs +++ b/tokio/src/sync/oneshot.rs @@ -801,7 +801,7 @@ impl Sender { if state.is_closed() { coop.made_progress(); - return Poll::Ready(()); + return Ready(()); } if state.is_tx_task_set() { From 6d25a00145d0dfed10c93eb37704fde88967341c Mon Sep 17 00:00:00 2001 From: icedrocket <114203630+icedrocket@users.noreply.github.com> Date: Sun, 25 Jun 2023 17:54:14 +0900 Subject: [PATCH 3/4] fs: update cfg attr in `fs::read_dir` (#5806) --- tokio/src/fs/read_dir.rs | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/tokio/src/fs/read_dir.rs b/tokio/src/fs/read_dir.rs index a144a8637..def735b3c 100644 --- a/tokio/src/fs/read_dir.rs +++ b/tokio/src/fs/read_dir.rs @@ -139,7 +139,9 @@ impl ReadDir { target_os = "solaris", target_os = "illumos", target_os = "haiku", - target_os = "vxworks" + target_os = "vxworks", + target_os = "nto", + target_os = "vita", )))] file_type: std.file_type().ok(), std: Arc::new(std), @@ -200,7 +202,9 @@ pub struct DirEntry { target_os = "solaris", target_os = "illumos", target_os = "haiku", - target_os = "vxworks" + target_os = "vxworks", + target_os = "nto", + target_os = "vita", )))] file_type: Option, std: Arc, @@ -331,7 +335,9 @@ impl DirEntry { target_os = "solaris", target_os = "illumos", target_os = "haiku", - target_os = "vxworks" + target_os = "vxworks", + target_os = "nto", + target_os = "vita", )))] if let Some(file_type) = self.file_type { return Ok(file_type); From 910a1e2fcf8ebafd41c2841144c3a1037af7dc40 Mon Sep 17 00:00:00 2001 From: Dhruv Vats Date: Sun, 25 Jun 2023 16:34:35 +0530 Subject: [PATCH 4/4] io: fix `futures_io::AsyncSeek` implementaion for `Compat` (#5783) --- tokio-util/Cargo.toml | 1 + tokio-util/src/compat.rs | 2 ++ tokio-util/tests/compat.rs | 43 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 46 insertions(+) create mode 100644 tokio-util/tests/compat.rs diff --git a/tokio-util/Cargo.toml b/tokio-util/Cargo.toml index f16099d78..04971c6b1 100644 --- a/tokio-util/Cargo.toml +++ b/tokio-util/Cargo.toml @@ -56,6 +56,7 @@ async-stream = "0.3.0" futures = "0.3.0" futures-test = "0.3.5" parking_lot = "0.12.0" +tempfile = "3.1.0" [package.metadata.docs.rs] all-features = true diff --git a/tokio-util/src/compat.rs b/tokio-util/src/compat.rs index 6a8802d96..6e19b2bd2 100644 --- a/tokio-util/src/compat.rs +++ b/tokio-util/src/compat.rs @@ -227,6 +227,8 @@ impl futures_io::AsyncSeek for Compat { pos: io::SeekFrom, ) -> Poll> { if self.seek_pos != Some(pos) { + // Ensure previous seeks have finished before starting a new one + ready!(self.as_mut().project().inner.poll_complete(cx))?; self.as_mut().project().inner.start_seek(pos)?; *self.as_mut().project().seek_pos = Some(pos); } diff --git a/tokio-util/tests/compat.rs b/tokio-util/tests/compat.rs new file mode 100644 index 000000000..278ebfcfb --- /dev/null +++ b/tokio-util/tests/compat.rs @@ -0,0 +1,43 @@ +#![cfg(all(feature = "compat"))] +#![cfg(not(target_os = "wasi"))] // WASI does not support all fs operations +#![warn(rust_2018_idioms)] + +use futures_io::SeekFrom; +use futures_util::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt}; +use tempfile::NamedTempFile; +use tokio::fs::OpenOptions; +use tokio_util::compat::TokioAsyncWriteCompatExt; + +#[tokio::test] +async fn compat_file_seek() -> futures_util::io::Result<()> { + let temp_file = NamedTempFile::new()?; + let mut file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .open(temp_file) + .await? + .compat_write(); + + file.write_all(&[0, 1, 2, 3, 4, 5]).await?; + file.write_all(&[6, 7]).await?; + + assert_eq!(file.stream_position().await?, 8); + + // Modify elements at position 2. + assert_eq!(file.seek(SeekFrom::Start(2)).await?, 2); + file.write_all(&[8, 9]).await?; + + file.flush().await?; + + // Verify we still have 8 elements. + assert_eq!(file.seek(SeekFrom::End(0)).await?, 8); + // Seek back to the start of the file to read and verify contents. + file.seek(SeekFrom::Start(0)).await?; + + let mut buf = Vec::new(); + let num_bytes = file.read_to_end(&mut buf).await?; + assert_eq!(&buf[..num_bytes], &[0, 1, 8, 9, 4, 5, 6, 7]); + + Ok(()) +}