Discussion: async loops

[alrz]

Moved from dotnet/roslyn#8014

async loops

Currently await in loops, blocks the iteration until the awaitable returns. async on a loop is to indicate that it may continue the iteration after an await is reached.

// waits in each iteration
async Task SaveAllAsync(Foo[] data) {
    foreach (var item in data) {
        await item.SaveAsync();
    }
}

// this won't get blocked and will wait in parallel
async Task SaveAllAsync(Foo[] data) {
    async foreach (var item in data) {
        await item.SaveAsync();
    } 
}

This behaves like a list of running tasks and a Task.WhenAll but it provides language support for processing tasks as they complete.

Note that in each iteration, we don't move to the next item until we get to an await, for the sake of fairness, we will do so even if the task has already been completed.

No iteration would run in parallel but might be in waiting state at the same time — a completed task might be waiting for another iteration to get to an await or end of the loop block, that way we won't need any lock in the body of the loop.

Similar to async on methods which would be redundant without any await in the method body, async on a loop have no effect without any await in the loop body, so in that case a warning would be useful.

An async loop may be used with yield return, e.g.

async IObservable<T> ProcessAllAsync(Foo[] data) {
    async foreach (var foo in data) {
        yield return await foo.ProcessAsync();
    }
}

I've used an IObservable<T> because if we want to return an IAsyncEnumerable<T> we will need to cache the results as they complete so it would be preferable to use a blocking loop in that case.

The async modifier can be added regardless of the source collection being an IAsyncEnumerable<T>.

foreach (var item in seq) // blocking iteration of a regular sequence
async foreach (var item in seq) // non-blocking iteration of a regular sequence
foreach await (var item in asyncSeq) // blocking iteration of an async sequence
async foreach await (var item in asyncSeq) // non-blocking iteration of an async sequence

For example, imagine we are iterating an async stream of integers, (example by @HaloFour)

IAsyncEnumerable<int> ids = LookupIdsAsync();
async foreach await (int id in ids) {
    Console.WriteLine($"Looking up user ID {id}.");
    User user = await LookupUserAsync(id);
    Console.WriteLine($"Found user {user.Name} for ID {id}.");
}
Console.WriteLine("Finished looking up users.");

The possible output might be the following since the multiple calls to LookupUserAsync could be dispatched concurrently:

Looking up user ID 1.
Looking up user ID 2.
Found user Bar for ID 2.
Found user Foo for ID 1.
Looking up user ID 3.
Found user Baz for ID 3.
Finished looking up users.

Whereas with an blocking foreach it would be the following since the async method would have to complete the loop body before iterating to the next element in the sequence:

Looking up user ID 1.
Found user Foo for ID 1.
Looking up user ID 2.
Found user Bar for ID 2.
Looking up user ID 3.
Found user Baz for ID 3.
Finished looking up users.

When we are waiting for MoveNextAsync we won't go further and if any other tasks from previous iterations have been completed we will continue the loop body. So in case of IAsycEnumerable<T> it'll be possible for a "Found user" message to be printed prior to a "Looking up user" message.

Other loops can be made async as well since this doesn't depend on iterating a particular collection,

async while (cond)
async do stmt; while (cond);
async for (int i = 0; cond; ++i)

It might be preferable to make a fresh variable in each iteration for async for just like foreach.

[benaadams]

Still think await is in the wrong place

foreach (var item in seq) // blocking iteration of a regular sequence
async foreach (var item in seq) // non-blocking iteration of a regular sequence
foreach (await var item in asyncSeq) // blocking iteration of an async sequence
async foreach (await var item in asyncSeq) // non-blocking iteration of an async sequence

So for the last two it should be

foreach (var item await in asyncSeq) // blocking iteration of an async sequence
async foreach (var item await in asyncSeq) // non-blocking iteration of an async sequence

With

async foreach (var item await in asyncSeq) 

being task returning

var task = async foreach (var item await in asyncSeq) 

and

foreach (var item await in asyncSeq)

being semantic sugar for

await async foreach (var item await in asyncSeq)

Await after the in should also be allowed (as it is currently) for the async return of the enumerable so; you could do something a bit yukky like:

var task = async foreach (var item await in await asyncSeq)

[alrz]

@benaadams

That's not part of this proposal, I've borrowed it from a design note, you may discuss it in #43.

[HaloFour]

I can't help but think that this could be accomplished through a library function. Something like:

public static Task ThenEach<T>(this IAsyncEnumerable<T> stream, Func<T, Task> action) { ... }

I don't doubt that there would be some challenges in implementation, particularly in keeping it "single threaded". But nothing insurmountable. Of course that wouldn't work with arbitrary task-alikes.

[jnm2]

Arbitrary tasklikes will be solved with SAwaitable.

[aluanhaddad]

@jnm2, @HaloFour this killer feature here, at least for me, is yield return await. Do shapes cover that use case? If so then that is great. If not, then is there another issue for that?

[alrz]

Note that yield return await will be possible once we have async streams, just that it would return an IAsyncEnumerable<T>. As you can see in the example above, if you use an async loop to yield values, you'd have an IObservable<T> instead.

[aluanhaddad]

@alrz thanks for clarifying.
So, to make sure I understand this correctly, given that Async Streams are implemented, the following will become possible

async IAsyncEnumerable<HttpResponseMessage> DownloadAsync(IEnumerable<string> urls)
{
    foreach(var url in urls)
    {
        yield return await new HttpClient().GetAsync(url);
    }
}

and that an intent of the async loops proposed here is to enable

async IObservable<HttpResponseMessage> DownloadAsync(IEnumerable<string> urls)
{
    async foreach(var url in urls)
    {
        yield return await new HttpClient().GetAsync(url);
    }
}

[omariom]

foreach is a loop.
If iterations of a loop don't happen sequentially is it a loop at all?
May be a new name?

[yaakov-h]

If iterations of a loop occur in parallel, is it still a loop?

[alrz]

this is not a parallel loop, just that it uses await as a break point to switch between iterations.

[aluanhaddad]

@alrz

I've used an IObservable because if we want to return an IAsyncEnumerable we will need to cache the results as they complete so it would be preferable to use a blocking loop in that case.

Thinking about this further I'm not sure I buy that argument. The async prefix paired with the IObservable<T> return type strikes me as rather subtle. I realize that they mean very different things but they don't emphasize it. I can definitely imagine fiddling with swapping the async in and out while toggling between IAsyncEnumerable and IObservable having started out simply wanting to write a non-blocking iterator and being unsure, or not particularly aware of, the distinction. This is exacerbated by the fact that both types have dozens of semi analogous methods.

[alrz]

@aluanhaddad

That's a valid concern. As an alternative to caching, I think we can do a non-blocking iteration but wait for MoveNextAsync once we are yielding the value to avoid caching, so that it would be possible to swap async in an iterator method without requiring the return type to be changed. That still does affect how we are running the loop, though it wouldn't be visible from the call-site point-of-view.

[alrz]

@HaloFour I shall defer to your own comments on that,

Agreed, where the composition is simple and doesn't branch. But once you introduce conditional logic, error handling, looping, disposable resources, etc., the ability to neatly compose a pipeline using LINQ/Rx/callbacks becomes an absolute mess and where something like this would be useful. [..] While PFX and Task.WhenAll do easily allow synchronization on multiple tasks, they couldn't allow yielding the processed results as they are completed very easily.

[omariom]

@alrz
Regardless of the language support it would be nice to have it as a library solution in CoreFx.
Async.Foreach? Async.Interleaved?

[HaloFour]

@alrz

Indeed, I'm a bit on the fence about it. I believe that the functionality itself would be quite useful, the question is whether or not it's something that can be done reasonably via a library vs. a language feature. The tricky part is definitely trying to keep the coroutine handling single-threaded, which could probably be done by hijacking/wrapping the synchronization context. Otherwise a library couldn't really handle the continutation of the lambda action argument.

[jcouv]

Relates to https://github.com/stephentoub/csharplang/blob/master/proposals/async-streams.md