New `JoinDeque` as FIFO-ordered analog of `JoinSet`

[nickkuk]

Is your feature request related to a problem? Please describe.
Several times people noted the analogies between tokio::task::JoinSet and futures::stream::FuturesUnordered and futures::StreamExt::buffer_unordered method. The JoinSet structure is considered more convenient to use and allows users to implement the same things as its counterparts from the futures crate.

At the same time, there is no analog for futures::stream::FuturesOrdered and futures::StreamExt::buffered in tokio.

Describe the solution you'd like
I'm suggesting to add tokio::task::JoinDeque data structure. Potential implementation can be found here.

Describe alternatives you've considered
Leave it for tokio users to implement themselves.

Additional context
None.

[ADD-SP]

I think an ordered version of Joinset has valid usecase. And I prefer to put the ordered version under tokio_util::task instead of tokio::task because it is not highly coupled with the async runtime itself.

[hawkw]

I'm not really sure if I understand the benefit of something like this over just putting all the spawned tasks in a VecDeque and awaiting them in order, like this:

let mut tasks = VecDeque::new();
for _ in 0..TASKS {
    tasks.push_back(tokio::spawn(async { ... });
}

while let Some(task) = tasks.pop_front() {
    task.await.unwrap(); // ...or do stuff with it
}

A JoinSet- or FuturesUnordered-like primitive is better than just putting all the tasks in a Vec, and looping over it and polling all of them to find the first task to complete, because it doesn't require O(n) polling every time you try to find the next completed future. And, doing that with async/.await syntax is difficult because there isn't an easy way to poll a Future once and keep going if it hasn't completed. So having a primitive for waiting for the first of a set of tasks to complete is very important.

However, if the goal is to wait for tasks to complete in FIFO order, you're not polling all the tasks, you're just polling the next one, so you don't need a more efficient way to determine which one has completed. So, you can just stick them in a normal deque and await them one by one. The reason that futures::stream::FuturesOrdered and futures::StreamExt::buffered exist isn't because having a primitive for awaiting tasks in FIFO sequence is necessary, it's because they allow you to turn that into a Stream, so that you can use it with Stream combinators or other functions that expect Stream implementations.

JoinSet isn't a Stream, since the tokio main crate doesn't depend on the futures crate for the Stream trait; instead, when AsyncIterator stabilizes in std, Tokio will use that trait instead. So, a JoinDeque type structure in the tokio crate wouldn't be able to provide the main value that the futures::stream::FuturesOrdered type provides at present, and therefore doesn't really accomplish anything beyond what you can get by just putting the tasks' JoinHandles in a queue and popping from it in order.

[hawkw]

One possible reason we might actually want to have a JoinDeque-like API in tokio proper someday is that, many moons ago, there was talk about eventually trying to change the idle_notified_set module used by JoinSet so that the linked list pointers are inlined into the task's allocation, rather than requiring a separate allocation for each task's entry in the lists. This would be a way to reduce allocations while using JoinSet that's only possible with tight integration with the runtime, and it's something that a JoinDeque-type API could also benefit from if it was part of tokio. But, we haven't done that, so there's no benefit for it today.

[nickkuk]

@hawkw Thank you! You are completely right, I overcomplicated things.

But now I'm thinking of more general variant of futures::StreamExt::buffered with two constraints: on total number of pending values and on number of futures that are actually executing (where first number can be larger than second). I believe that for such feature I have to poll all futures at once to run new one when some other finishes.

Anyways, such structure seems to be too special for my task, so maybe I should close this issue.

[ADD-SP]

I was persuaded by @hawkw, JoinDeque doesn't make many differences compared to VecDeque<JoinHandle>.

[ADD-SP]

But now I'm thinking of more general variant of futures::StreamExt::buffered with two constraints: on total number of pending values and on number of futures that are actually executing (where first number can be larger than second).

If I Understand Correctly, are you going to do concurrency control by futures::StreamExt::buffered?

[nickkuk]

No, I just tried to implement the similar functional as futures::StreamExt::buffered but in Tokio. I see now that just VecDeque<JoinHandle> is enough for ordering and dynamical control over maximal amount of pending values.

Also I realized that I can use tokio::sync::Semaphore to limit the number of outgoing requests being sent at the same time and it seems to be compatible with VecDeque<JoinHandle> buffering. So, I don't even need to use JoinSet.

[Darksonn]

I could see the use as an utility since VecDeque doesn't give you the cancel-on-drop behavior, and people are unlikely to implement that part correctly if they do it on their own.

[nickkuk]

I could see the use as an utility since VecDeque doesn't give you the cancel-on-drop behavior, and people are unlikely to implement that part correctly if they do it on their own.

Yeah, if we want add such structure with FIFO ordering and cancel-on-drop behavior, do you see more lightweight implementation than via JoinSet and maintaining ids, e.g., maybe by using some tokio internals?