tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-24 00:00:11 +02:00
Code Issues Packages Projects Releases Wiki Activity

joinset: rename join_one to join_next (#4755)

Browse Source
This commit is contained in:
David Barsky
2022-06-09 22:41:25 +02:00
committed by GitHub
parent 340c4dc3b2
commit 08d49532b4
6 changed files with 89 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio-util/src/task/join_map.rs
+32 -25
View File
@@ -50,9 +50,9 @@ use tokio::task::{AbortHandle, Id, JoinError, JoinSet, LocalSet};
///
/// let mut seen = [false; 10];
///
/// // When a task completes, `join_one` returns the task's key along
/// // When a task completes, `join_next` returns the task's key along
/// // with its output.
/// while let Some((key, res)) = map.join_one().await {
/// while let Some((key, res)) = map.join_next().await {
/// seen[key] = true;
/// assert!(res.is_ok(), "task {} completed successfully!", key);
/// }
@@ -82,7 +82,7 @@ use tokio::task::{AbortHandle, Id, JoinError, JoinSet, LocalSet};
/// // provided key.
/// assert!(aborted);
///
/// while let Some((key, res)) = map.join_one().await {
/// while let Some((key, res)) = map.join_next().await {
/// if key == "goodbye world" {
/// // The aborted task should complete with a cancelled `JoinError`.
/// assert!(res.unwrap_err().is_cancelled());
@@ -277,14 +277,14 @@ where
///
/// 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_one`] will
/// be removed from the `JoinMap`; a subsequent call to [`join_next`] will
/// *not* return a cancelled [`JoinError`] for that task.
///
/// # Panics
///
/// This method panics if called outside of a Tokio runtime.
///
/// [`join_one`]: Self::join_one
/// [`join_next`]: Self::join_next
#[track_caller]
pub fn spawn<F>(&mut self, key: K, task: F)
where
@@ -301,10 +301,10 @@ where
///
/// 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_one`] will
/// be removed from the `JoinMap`; a subsequent call to [`join_next`] will
/// *not* return a cancelled [`JoinError`] for that task.
///
/// [`join_one`]: Self::join_one
/// [`join_next`]: Self::join_next
#[track_caller]
pub fn spawn_on<F>(&mut self, key: K, task: F, handle: &Handle)
where
@@ -321,7 +321,7 @@ where
///
/// 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_one`] will
/// be removed from the `JoinMap`; a subsequent call to [`join_next`] will
/// *not* return a cancelled [`JoinError`] for that task.
///
/// # Panics
@@ -329,7 +329,7 @@ where
/// This method panics if it is called outside of a `LocalSet`.
///
/// [`LocalSet`]: tokio::task::LocalSet
/// [`join_one`]: Self::join_one
/// [`join_next`]: Self::join_next
#[track_caller]
pub fn spawn_local<F>(&mut self, key: K, task: F)
where
@@ -345,11 +345,11 @@ where
///
/// 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_one`] will
/// be removed from the `JoinMap`; a subsequent call to [`join_next`] will
/// *not* return a cancelled [`JoinError`] for that task.
///
/// [`LocalSet`]: tokio::task::LocalSet
/// [`join_one`]: Self::join_one
/// [`join_next`]: Self::join_next
#[track_caller]
pub fn spawn_local_on<F>(&mut self, key: K, task: F, local_set: &LocalSet)
where
@@ -399,7 +399,7 @@ where
///
/// # Cancel Safety
///
/// This method is cancel safe. If `join_one` is used as the event in a [`tokio::select!`]
/// This method is cancel safe. If `join_next` is used as the event in a [`tokio::select!`]
/// statement and some other branch completes first, it is guaranteed that no tasks were
/// removed from this `JoinMap`.
///
@@ -416,8 +416,9 @@ where
/// * `None` if the `JoinMap` is empty.
///
/// [`tokio::select!`]: tokio::select
pub async fn join_one(&mut self) -> Option<(K, Result<V, JoinError>)> {
let (res, id) = match self.tasks.join_one_with_id().await {
#[doc(alias = "join_one")]
pub async fn join_next(&mut self) -> Option<(K, Result<V, JoinError>)> {
let (res, id) = match self.tasks.join_next_with_id().await {
Some(Ok((id, output))) => (Ok(output), id),
Some(Err(e)) => {
let id = e.id();
@@ -429,19 +430,25 @@ where
Some((key, res))
}
#[doc(hidden)]
#[deprecated(since = "0.7.4", note = "renamed to `JoinMap::join_next`.")]
pub async fn join_one(&mut self) -> Option<(K, Result<V, JoinError>)> {
self.join_next().await
}
/// Aborts all tasks and waits for them to finish shutting down.
///
/// Calling this method is equivalent to calling [`abort_all`] and then calling [`join_one`] in
/// Calling this method is equivalent to calling [`abort_all`] and then calling [`join_next`] in
/// a loop until it returns `None`.
///
/// This method ignores any panics in the tasks shutting down. When this call returns, the
/// `JoinMap` will be empty.
///
/// [`abort_all`]: fn@Self::abort_all
/// [`join_one`]: fn@Self::join_one
/// [`join_next`]: fn@Self::join_next
pub async fn shutdown(&mut self) {
self.abort_all();
while self.join_one().await.is_some() {}
while self.join_next().await.is_some() {}
}
/// Abort the task corresponding to the provided `key`.
@@ -467,7 +474,7 @@ where
/// // Look up the "goodbye world" task in the map and abort it.
/// map.abort("goodbye world");
///
/// while let Some((key, res)) = map.join_one().await {
/// while let Some((key, res)) = map.join_next().await {
/// if key == "goodbye world" {
/// // The aborted task should complete with a cancelled `JoinError`.
/// assert!(res.unwrap_err().is_cancelled());
@@ -548,7 +555,7 @@ where
/// map.abort_matching(|key| key.starts_with("goodbye"));
///
/// let mut seen = 0;
/// while let Some((key, res)) = map.join_one().await {
/// while let Some((key, res)) = map.join_next().await {
/// seen += 1;
/// if key.starts_with("goodbye") {
/// // The aborted task should complete with a cancelled `JoinError`.
@@ -567,7 +574,7 @@ where
pub fn abort_matching(&mut self, mut predicate: impl FnMut(&K) -> bool) {
// Note: this method iterates over the tasks and keys *without* removing
// any entries, so that the keys from aborted tasks can still be
// returned when calling `join_one` in the future.
// returned when calling `join_next` in the future.
for (Key { ref key, .. }, task) in &self.tasks_by_key {
if predicate(key) {
task.abort();
@@ -578,9 +585,9 @@ where
/// Returns `true` if this `JoinMap` contains a task for the provided key.
///
/// If the task has completed, but its output hasn't yet been consumed by a
/// call to [`join_one`], this method will still return `true`.
/// call to [`join_next`], this method will still return `true`.
///
/// [`join_one`]: fn@Self::join_one
/// [`join_next`]: fn@Self::join_next
pub fn contains_key<Q: ?Sized>(&self, key: &Q) -> bool
where
Q: Hash + Eq,
@@ -593,9 +600,9 @@ where
/// [task ID].
///
/// If the task has completed, but its output hasn't yet been consumed by a
/// call to [`join_one`], this method will still return `true`.
/// call to [`join_next`], this method will still return `true`.
///
/// [`join_one`]: fn@Self::join_one
/// [`join_next`]: fn@Self::join_next
/// [task ID]: tokio::task::Id
pub fn contains_task(&self, task: &Id) -> bool {
self.get_by_id(task).is_some()
@@ -739,7 +746,7 @@ where
/// Aborts all tasks on this `JoinMap`.
///
/// This does not remove the tasks from the `JoinMap`. To wait for the tasks to complete
/// cancellation, you should call `join_one` in a loop until the `JoinMap` is empty.
/// cancellation, you should call `join_next` in a loop until the `JoinMap` is empty.
pub fn abort_all(&mut self) {
self.tasks.abort_all()
}
tokio-util/tests/task_join_map.rs
+12 -12
View File
@@ -24,7 +24,7 @@ async fn test_with_sleep() {
map.detach_all();
assert_eq!(map.len(), 0);
assert!(matches!(map.join_one().await, None));
assert!(matches!(map.join_next().await, None));
for i in 0..10 {
map.spawn(i, async move {
@@ -35,7 +35,7 @@ async fn test_with_sleep() {
}
let mut seen = [false; 10];
while let Some((k, res)) = map.join_one().await {
while let Some((k, res)) = map.join_next().await {
seen[k] = true;
assert_eq!(res.expect("task should have completed successfully"), k);
}
@@ -43,7 +43,7 @@ async fn test_with_sleep() {
for was_seen in &seen {
assert!(was_seen);
}
assert!(matches!(map.join_one().await, None));
assert!(matches!(map.join_next().await, None));
// Do it again.
for i in 0..10 {
@@ -54,7 +54,7 @@ async fn test_with_sleep() {
}
let mut seen = [false; 10];
while let Some((k, res)) = map.join_one().await {
while let Some((k, res)) = map.join_next().await {
seen[k] = true;
assert_eq!(res.expect("task should have completed successfully"), k);
}
@@ -62,7 +62,7 @@ async fn test_with_sleep() {
for was_seen in &seen {
assert!(was_seen);
}
assert!(matches!(map.join_one().await, None));
assert!(matches!(map.join_next().await, None));
}
#[tokio::test]
@@ -101,7 +101,7 @@ async fn alternating() {
assert_eq!(map.len(), 2);
for i in 0..16 {
let (_, res) = map.join_one().await.unwrap();
let (_, res) = map.join_next().await.unwrap();
assert!(res.is_ok());
assert_eq!(map.len(), 1);
map.spawn(i, async {});
@@ -127,7 +127,7 @@ async fn abort_by_key() {
}
}
while let Some((key, res)) = map.join_one().await {
while let Some((key, res)) = map.join_next().await {
match res {
Ok(()) => {
num_completed += 1;
@@ -161,7 +161,7 @@ async fn abort_by_predicate() {
// abort odd-numbered tasks.
map.abort_matching(|key| key % 2 != 0);
while let Some((key, res)) = map.join_one().await {
while let Some((key, res)) = map.join_next().await {
match res {
Ok(()) => {
num_completed += 1;
@@ -190,12 +190,12 @@ fn runtime_gone() {
drop(rt);
}
let (key, res) = rt().block_on(map.join_one()).unwrap();
let (key, res) = rt().block_on(map.join_next()).unwrap();
assert_eq!(key, "key");
assert!(res.unwrap_err().is_cancelled());
}
// This ensures that `join_one` works correctly when the coop budget is
// This ensures that `join_next` works correctly when the coop budget is
// exhausted.
#[tokio::test(flavor = "current_thread")]
async fn join_map_coop() {
@@ -222,7 +222,7 @@ async fn join_map_coop() {
let mut count = 0;
let mut coop_count = 0;
loop {
match map.join_one().now_or_never() {
match map.join_next().now_or_never() {
Some(Some((key, Ok(i)))) => assert_eq!(key, i),
Some(Some((key, Err(err)))) => panic!("failed[{}]: {}", key, err),
None => {
@@ -260,7 +260,7 @@ async fn abort_all() {
let mut count = 0;
let mut seen = [false; 10];
while let Some((k, res)) = map.join_one().await {
while let Some((k, res)) = map.join_next().await {
seen[k] = true;
if let Err(err) = res {
assert!(err.is_cancelled());
tokio/src/runtime/tests/loom_join_set.rs
+5 -5
View File
@@ -16,13 +16,13 @@ fn test_join_set() {
assert_eq!(set.len(), 1);
set.spawn(async { () });
assert_eq!(set.len(), 2);
let () = set.join_one().await.unwrap().unwrap();
let () = set.join_next().await.unwrap().unwrap();
assert_eq!(set.len(), 1);
set.spawn(async { () });
assert_eq!(set.len(), 2);
let () = set.join_one().await.unwrap().unwrap();
let () = set.join_next().await.unwrap().unwrap();
assert_eq!(set.len(), 1);
let () = set.join_one().await.unwrap().unwrap();
let () = set.join_next().await.unwrap().unwrap();
assert_eq!(set.len(), 0);
set.spawn(async { () });
assert_eq!(set.len(), 1);
@@ -60,7 +60,7 @@ fn abort_all_during_completion() {
set.spawn(async { () });
set.abort_all();
match set.join_one().await {
match set.join_next().await {
Some(Ok(())) => complete_happened.store(true, SeqCst),
Some(Err(err)) if err.is_cancelled() => cancel_happened.store(true, SeqCst),
Some(Err(err)) => panic!("fail: {}", err),
@@ -69,7 +69,7 @@ fn abort_all_during_completion() {
}
}
assert!(matches!(set.join_one().await, None));
assert!(matches!(set.join_next().await, None));
});
drop(set);
tokio/src/task/join_set.rs
+26 -12
View File
@@ -43,7 +43,7 @@ use crate::util::IdleNotifiedSet;
/// }
///
/// let mut seen = [false; 10];
/// while let Some(res) = set.join_one().await {
/// while let Some(res) = set.join_next().await {
/// let idx = res.unwrap();
/// seen[idx] = true;
/// }
@@ -195,17 +195,24 @@ impl<T: 'static> JoinSet<T> {
abort
}
#[doc(hidden)]
#[deprecated(since = "1.20.0", note = "renamed to `JoinSet::join_next`.")]
pub async fn join_one(&mut self) -> Option<Result<T, JoinError>> {
self.join_next().await
}
/// Waits until one of the tasks in the set completes and returns its output.
///
/// Returns `None` if the set is empty.
///
/// # Cancel Safety
///
/// This method is cancel safe. If `join_one` is used as the event in a `tokio::select!`
/// This method is cancel safe. If `join_next` is used as the event in a `tokio::select!`
/// statement and some other branch completes first, it is guaranteed that no tasks were
/// removed from this `JoinSet`.
pub async fn join_one(&mut self) -> Option<Result<T, JoinError>> {
crate::future::poll_fn(|cx| self.poll_join_one(cx))
#[doc(alias = "join_one")]
pub async fn join_next(&mut self) -> Option<Result<T, JoinError>> {
crate::future::poll_fn(|cx| self.poll_join_next(cx))
.await
.map(|opt| opt.map(|(_, res)| res))
}
@@ -220,35 +227,42 @@ impl<T: 'static> JoinSet<T> {
///
/// # Cancel Safety
///
/// This method is cancel safe. If `join_one_with_id` is used as the event in a `tokio::select!`
/// This method is cancel safe. If `join_next_with_id` is used as the event in a `tokio::select!`
/// statement and some other branch completes first, it is guaranteed that no tasks were
/// removed from this `JoinSet`.
///
/// [task ID]: crate::task::Id
/// [`JoinError::id`]: fn@crate::task::JoinError::id
#[doc(alias = "join_one_with_id")]
pub async fn join_next_with_id(&mut self) -> Option<Result<(Id, T), JoinError>> {
crate::future::poll_fn(|cx| self.poll_join_next(cx)).await
}
#[doc(hidden)]
#[deprecated(since = "1.20.0", note = "renamed to `JoinSet::join_next_with_id`")]
pub async fn join_one_with_id(&mut self) -> Option<Result<(Id, T), JoinError>> {
crate::future::poll_fn(|cx| self.poll_join_one(cx)).await
self.join_next_with_id().await
}
/// Aborts all tasks and waits for them to finish shutting down.
///
/// Calling this method is equivalent to calling [`abort_all`] and then calling [`join_one`] in
/// Calling this method is equivalent to calling [`abort_all`] and then calling [`join_next`] in
/// a loop until it returns `None`.
///
/// This method ignores any panics in the tasks shutting down. When this call returns, the
/// `JoinSet` will be empty.
///
/// [`abort_all`]: fn@Self::abort_all
/// [`join_one`]: fn@Self::join_one
/// [`join_next`]: fn@Self::join_next
pub async fn shutdown(&mut self) {
self.abort_all();
while self.join_one().await.is_some() {}
while self.join_next().await.is_some() {}
}
/// Aborts all tasks on this `JoinSet`.
///
/// This does not remove the tasks from the `JoinSet`. To wait for the tasks to complete
/// cancellation, you should call `join_one` in a loop until the `JoinSet` is empty.
/// cancellation, you should call `join_next` in a loop until the `JoinSet` is empty.
pub fn abort_all(&mut self) {
self.inner.for_each(|jh| jh.abort());
}
@@ -267,7 +281,7 @@ impl<T: 'static> JoinSet<T> {
///
/// When the method returns `Poll::Pending`, the `Waker` in the provided `Context` is scheduled
/// to receive a wakeup when a task in the `JoinSet` completes. Note that on multiple calls to
/// `poll_join_one`, only the `Waker` from the `Context` passed to the most recent call is
/// `poll_join_next`, only the `Waker` from the `Context` passed to the most recent call is
/// scheduled to receive a wakeup.
///
/// # Returns
@@ -288,7 +302,7 @@ impl<T: 'static> JoinSet<T> {
///
/// [coop budget]: crate::task#cooperative-scheduling
/// [task ID]: crate::task::Id
fn poll_join_one(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<(Id, T), JoinError>>> {
fn poll_join_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<(Id, T), JoinError>>> {
// The call to `pop_notified` moves the entry to the `idle` list. It is moved back to
// the `notified` list if the waker is notified in the `poll` call below.
let mut entry = match self.inner.pop_notified(cx.waker()) {
tokio/tests/async_send_sync.rs
+3 -3
View File
@@ -454,11 +454,11 @@ assert_value!(tokio::task::JoinHandle<YY>: Send & Sync & Unpin);
#[cfg(tokio_unstable)]
mod unstable {
use super::*;
async_assert_fn!(tokio::task::JoinSet<Cell<u32>>::join_one(_): Send & Sync & !Unpin);
async_assert_fn!(tokio::task::JoinSet<Cell<u32>>::join_next(_): Send & Sync & !Unpin);
async_assert_fn!(tokio::task::JoinSet<Cell<u32>>::shutdown(_): Send & Sync & !Unpin);
async_assert_fn!(tokio::task::JoinSet<Rc<u32>>::join_one(_): !Send & !Sync & !Unpin);
async_assert_fn!(tokio::task::JoinSet<Rc<u32>>::join_next(_): !Send & !Sync & !Unpin);
async_assert_fn!(tokio::task::JoinSet<Rc<u32>>::shutdown(_): !Send & !Sync & !Unpin);
async_assert_fn!(tokio::task::JoinSet<u32>::join_one(_): Send & Sync & !Unpin);
async_assert_fn!(tokio::task::JoinSet<u32>::join_next(_): Send & Sync & !Unpin);
async_assert_fn!(tokio::task::JoinSet<u32>::shutdown(_): Send & Sync & !Unpin);
assert_value!(tokio::task::JoinSet<YY>: Send & Sync & Unpin);
assert_value!(tokio::task::JoinSet<YN>: Send & Sync & Unpin);
tokio/tests/task_join_set.rs
+11 -11
View File
@@ -24,7 +24,7 @@ async fn test_with_sleep() {
set.detach_all();
assert_eq!(set.len(), 0);
assert!(matches!(set.join_one().await, None));
assert!(matches!(set.join_next().await, None));
for i in 0..10 {
set.spawn(async move {
@@ -35,14 +35,14 @@ async fn test_with_sleep() {
}
let mut seen = [false; 10];
while let Some(res) = set.join_one().await.transpose().unwrap() {
while let Some(res) = set.join_next().await.transpose().unwrap() {
seen[res] = true;
}
for was_seen in &seen {
assert!(was_seen);
}
assert!(matches!(set.join_one().await, None));
assert!(matches!(set.join_next().await, None));
// Do it again.
for i in 0..10 {
@@ -53,14 +53,14 @@ async fn test_with_sleep() {
}
let mut seen = [false; 10];
while let Some(res) = set.join_one().await.transpose().unwrap() {
while let Some(res) = set.join_next().await.transpose().unwrap() {
seen[res] = true;
}
for was_seen in &seen {
assert!(was_seen);
}
assert!(matches!(set.join_one().await, None));
assert!(matches!(set.join_next().await, None));
}
#[tokio::test]
@@ -99,7 +99,7 @@ async fn alternating() {
assert_eq!(set.len(), 2);
for _ in 0..16 {
let () = set.join_one().await.unwrap().unwrap();
let () = set.join_next().await.unwrap().unwrap();
assert_eq!(set.len(), 1);
set.spawn(async {});
assert_eq!(set.len(), 2);
@@ -123,7 +123,7 @@ async fn abort_tasks() {
}
}
loop {
match set.join_one().await {
match set.join_next().await {
Some(Ok(res)) => {
num_completed += 1;
assert_eq!(res % 2, 0);
@@ -150,13 +150,13 @@ fn runtime_gone() {
}
assert!(rt()
.block_on(set.join_one())
.block_on(set.join_next())
.unwrap()
.unwrap_err()
.is_cancelled());
}
// This ensures that `join_one` works correctly when the coop budget is
// This ensures that `join_next` works correctly when the coop budget is
// exhausted.
#[tokio::test(flavor = "current_thread")]
async fn join_set_coop() {
@@ -182,7 +182,7 @@ async fn join_set_coop() {
let mut count = 0;
let mut coop_count = 0;
loop {
match set.join_one().now_or_never() {
match set.join_next().now_or_never() {
Some(Some(Ok(()))) => {}
Some(Some(Err(err))) => panic!("failed: {}", err),
None => {
@@ -219,7 +219,7 @@ async fn abort_all() {
assert_eq!(set.len(), 10);
let mut count = 0;
while let Some(res) = set.join_one().await {
while let Some(res) = set.join_next().await {
if let Err(err) = res {
assert!(err.is_cancelled());
}