turn the futures-unordered footgun into a status quo story

[nikomatsakis]

Closes #131

[nikomatsakis]

Hey @farnz, does everything in here read true to you? I tried to take what you said during that writing session last week and write it out.

Also @Darksonn, I'd appreciate you making sure I don't have any factual inaccuracies.

[Darksonn]

In general, this is a really nasty footgun in the design of Stream. Besides the placement of stream::iter, the story seems correct to me. Another way to fix it would be to write it as:

async fn do_work(database: &Database) {
    let work = do_select(database, FIND_WORK_QUERY).await?;

    stream::iter(work)
        .map(|item| async move {
            let work_item = do_select(database, work_from_item(item)).await;
            process_work_item(database, work_item).await;
        })
        .buffered(5)
        .for_each(|()| std::future::ready(()))
        .await;
}

The above could be further rewritten to use for_each_concurrent. With this method you process the work items immediately, but they are included in the upper limit of five tasks.

To not count them in the upper limit of five, a further way to write it would be:

async fn do_work(database: &Database) {
    let work = do_select(database, FIND_WORK_QUERY).await?;

    let selects = stream::iter(work)
        .map(|item| do_select(database, work_from_item(item)))
        .buffered(5)
        .fuse();
    tokio::pin!(selects);

    let mut results = FuturesUnordered::new();

    loop {
        tokio::select! {
            Some(work_item) = selects.next() => {
                results.push(process_work_item(database, work_item));
            },
            Some(()) = results.next() => { /* do nothing */ },
            else => break,
        }
    }
}

I'm not sure if there is a simpler way to write the above.

[nikomatsakis]

Ah, that's interesting. I had not considered the select! version. I'll include that in the FAQ I think. I imagine there are probably utilities that aim to do something similar. I was also thinking about join, but for that, you'd need a future that 'drains' a stream completely, and I guess maybe one doesn't exist?

I was imagining something like:

join!(
    selects.map(|work_item| async move { results.push(process_work_item(database, work_item)) }).complete(),
    results.complete(),
);

where complete is a method like

impl<T: Stream> StreamExt for T {
    async fn complete(self) {
        while let Some(_) = self.next().await { }
    }
}

[Darksonn]

I mean, my select! is essentially that solution except that by cancelling the results.next() call when a new work item becomes available, you gain the mutable access to results that you need to push a future into it.

but for that, you'd need a future that 'drains' a stream completely, and I guess maybe one doesn't exist?

You can do that with .for_each(|()| std::future::ready(())).await.

[Darksonn]

It's true that the spawn used here would not have fixed her problem, but spawning would have fixed her problem if she had put the spawn around do_select rather than around process_work_item. (Ignoring any issues regarding 'static)

[nikomatsakis]

Yes, I should note that in the FAQ,

[farnz]

The fix we use most commonly is described below - we spawn inside do_select to ensure that the work done there can complete.

[nikomatsakis]

I mean, my select! is essentially that solution except that by cancelling the results.next() call when a new work item becomes available, you gain the mutable access to results that you need to push a future into it.

Right, I know they're equivalent, I was just playing around with what it might look like to not "open code" it. But you're right about the borrow checker interactions, good point.

[farnz]

Otherwise, LGTM. Thanks for writing this for me!

[farnz]

The fix we use most commonly is described below - we spawn inside do_select to ensure that the work done there can complete.

[nikomatsakis]

I incorporated @farnz's feedback, which I think really improved the story, thanks again for that. It's much more precise now.