add taskset api #4238
add taskset api #4238
Conversation
Noah-Kennedy
requested review from
carllerche,
Darksonn,
hawkw and
LucioFranco
This API simplifies managing the lifetimes of sets of tasks, allowing for joining in the style of FuturesUnordered or mass-cancellation.
Noah-Kennedy
force-pushed
the
noah/taskset-api
branch
from
1914c47
to
5241f45
Compare
Noah-Kennedy
added
A-tokio-stream
| @@ -43,6 +44,7 @@ futures-core = "0.3.0" | |||
| futures-sink = "0.3.0" | |||
| futures-io = { version = "0.3.0", optional = true } | |||
| futures-util = { version = "0.3.0", optional = true } | |||
| futures = { version = "0.3.0", optional = true } | |||
There was a problem hiding this comment.
It seems like you're only using this for futures::future::pending() in doc-tests, but you can just use std::future::pending() for those.
| /// assert_eq!(NUMS, joined.as_slice()); | ||
| /// # }); | ||
| /// ``` | ||
| pub async fn join_all(mut self) -> Vec<Result<T, JoinError>> { |
There was a problem hiding this comment.
I'm not a big fan of this return type - you don't get any #[must_use] warnings if you ignore the return value. Though I suppose you could put #[must_use] on the method itself.
There was a problem hiding this comment.
Though I suppose you could put
#[must_use]on the method itself.
@Darksonn Currently, this might not work as expected on async fn (rust-lang/rust#78149). A lint was added to rustc recently (playground), at least until #[must_use] is changed to apply to the future's output value instead of the impl Future returned by the function.
warning: `must_use` attribute on `async` functions applies to the anonymous `Future` returned by the function, not the value within
tokio-util/src/taskset.rs
Outdated
| let (t, idx, _tasks) = futures_util::future::select_all(self.tasks.iter_mut()).await; | ||
|
|
||
| self.tasks.remove(idx); |
There was a problem hiding this comment.
This has an O(n) cost per call, and that's too expensive. It would be a repeat of #2401 if anyone calls this in a loop. The same applies to the poll function (it's worse there).
There was a problem hiding this comment.
I scraped both together fairly quickly so that we could look at the API. I'll rewrite the internals of both.
There was a problem hiding this comment.
Pushed up a new one that isn't as atrociously awful, but getting this to not be O(n) will require a different underlying data structure. I'll look into this more tonight.
tokio-util/src/taskset.rs
Outdated
| /// # }); | ||
| /// ``` | ||
| pub async fn join_all(mut self) -> Vec<Result<T, JoinError>> { | ||
| let output = futures_util::future::join_all(self.tasks.iter_mut()).await; |
There was a problem hiding this comment.
This should just be a simple loop awaiting each join handle in order.
tokio-util/src/taskset.rs
Outdated
| for task in self.tasks.iter_mut() { | ||
| output.push(task.await) | ||
| } | ||
|
|
||
| output |
There was a problem hiding this comment.
Seems like it should be ok to add an self.tasks.clear() after the loop so the destructor doesn't have to abort everything again.
There was a problem hiding this comment.
Thanks for the catch, done.
clear taskset after joinall, so drop doesn't try and abort the completed tasks
|
@Darksonn what do we want to do about the O(n) |
|
You're probably going to need some sort of custom waker to get around it. |
|
@Darksonn FuturesUnordered uses several linked lists (I believe one of which is intrusive) to manage the stored tasks. Implementing a similar data structure is one option, although I don't think that implementing such a complicated set of datastructures for a single API is worthwhile. If we put this into tokio instead of tokio-util, would it be possible for us to reuse some of the internal APIs to simplify the implementation? |
|
Well, I'm pretty sure we could do it in a simpler way than what |
|
Closing since this was implemented in #4335 instead. |
Closes #3903