Proposal: Public APIs for C# 8 async streams

[stephentoub]

We're at a point where we should start exposing the necessary library support for C# 8 async streams, aka async enumerables, aka async iterators. While we may find ourselves wanting to make further changes to these APIs, we've locked on what we plan to deliver initially, and we can adapt as necessary based on feedback.

namespace System
{
    public interface IAsyncDisposable
    {
        ValueTask DisposeAsync();
    }
}

namespace System.Collections.Generic
{
    public interface IAsyncEnumerable<out T>
    {
        IAsyncEnumerator<T> GetAsyncEnumerator();
    }

    public interface IAsyncEnumerator<out T> : IAsyncDisposable
    {
        ValueTask<bool> MoveNextAsync();
        T Current { get; }
    }
}

namespace System.Runtime.CompilerServices
{
    public struct AsyncIteratorMethodBuilder
    {
        public static AsyncIteratorMethodBuilder Create();

        public void MoveNext<TStateMachine>(ref TStateMachine stateMachine)
            where TStateMachine : IAsyncStateMachine;

        public void AwaitOnCompleted<TAwaiter, TStateMachine>(
            ref TAwaiter awaiter, ref TStateMachine stateMachine)
            where TAwaiter : INotifyCompletion
            where TStateMachine : IAsyncStateMachine;

        public void AwaitUnsafeOnCompleted<TAwaiter, TStateMachine>(
            ref TAwaiter awaiter, ref TStateMachine stateMachine)
            where TAwaiter : ICriticalNotifyCompletion
            where TStateMachine : IAsyncStateMachine;

        public void Complete();
    }
}

Assembly:
We need to decide where we put these. My expectation is .NET Core we'd have all of this in System.Private.CoreLib, but we could also put it in System.Threading.Tasks.Extensions a new assembly. If we want to OOB this, it could be in both, with the .NET Core S.T.T.Extensions the new assembly just type forwarding to S.P.CoreLib as happens today for ValueTask.

A few notes:

• Alternative WaitForNextAsync+TryGetNext current design. We spent a lot of time exploring this attractive alternative. It has perf advantages, but also non-trivial additional complexity when using the APIs directly rather than through compiler support (e.g. foreach await). We also have a design where we could light-up with the alternative API should we find that the perf benefits are desirable enough to introduce a second interface. As such, we're going with the simpler, more familiar, and easier to work with MoveNextAsync+Current design.
• Thread safety. While these APIs are for asynchrony, that doesn't imply an instance of an enumerator can be used concurrently. Async enumerators are explicitly not thread safe, in terms of there being no way to correctly used MoveNextAsync+Current in an atomic fashion (without higher-level locking that provided the thread safety), but also in terms of accessing MoveNextAsync again before having consumed a previous calls ValueTask; doing so is erroneous and has undefined behavior. So, too, is accessing DisposeAsync after having called MoveNextAsync but without having consumed its ValueTask.
• ValueTask instead of Task. The original design called for MoveNextAsync to return Task<bool> and DisposeAsync to return Task. That works well when MoveNextAsync and DisposeAsync complete synchronously, but when they complete asynchronously, it requires allocation of another object to represent the eventual completion of the async operation. By instead returning ValueTask<bool> and ValueTask, an implementation can choose to implement IValueTaskSource and reuse the same object repeatedly for one call after another; in this fashion, for example, the compiler-generated type that's returned from an iterator can serve as the enumerable, as the enumerator, and as the promise for every asynchronously-completing MoveNextAsync and DisposeAsync call made on that enumerator, such that the whole async enumerable mechanism can incur overhead of a single allocation.
• Cancellation. Cancellation is provided external to the interfaces. If you want operations on the enumerable to be cancelable, you need to pass a CancellationToken to the thing creating the enumerable, such that the token can be embedded into the enumerable and used in its operation. You can of course choose to cancel awaits by awaiting something that itself represents both the MoveNextAsync ValueTask<bool> and a CancellationToken, but that would only cancel the await, not the underlying operation being awaited. As for IAsyncDisposable, while in theory it makes sense that anything async can be canceled, disposal is about cleanup, closing things out, freeing resources, etc., which is generally not something that should be canceled; cleanup is still important for work that's canceled. The same CancellationToken that caused the actual work to be canceled would typically be the same token passed to DisposeAsync, making DisposeAsync worthless because cancellation of the work would cause DisposeAsync to be a nop. If someone wants to avoid being blocked waiting for disposal, they can avoid waiting on the resulting ValueTask, or wait on it only for some period of time.
• Naming. Async is used in the various method names (e.g. DisposeAsync) even though the type also includes Async so that a type might implement both the synchronous and asynchronous counterparts and easily differentiate them.
• AsyncIteratorMethodBuilder. The compiler could get away with using the existing AsyncTaskMethodBuilder or AsyncVoidMethodBuilder types, but these both have negative impact that can be avoided by using a new, specially-designed type. AsyncTaskMethodBuilder allocates a Task to represent the async method, but that Task goes unused in an iterator. AsyncVoidMethodBuilder interacts with SynchronizationContext, because async void methods need to do so (e.g. calling OperationStarted and OperationCompleted on the current SynchronizationContext if there is one). And some of the methods are poorly named, e.g. Start makes sense when talking about starting an async method, but not when talking about iterating with an iterator. As such, we introduce a new type tailored to async iterators. Create just returns a builder that the compiler can use, which is likely just default(AsyncIteratorMethodBuilder), but the method might also do additional optional work, like tracing. MoveNext pushes the state machine forward, effectively just calling stateMachine.MoveNext(), but doing so with the appropriate handling of ExecutionContext. AwaitOnCompleted and AwaitUnsafeOnCompleted are exactly what they are on the existing builders. And Complete just serves to notify the builder that the iterator has finished iterating: technically this isn't necessary, and it may just be a nop, but it gives us a hook to be able to do things like tracing/logging should we choose to do so. (We could decide not to include this.)
• Implementing sync and async interfaces. It's fine for a single type to implement both the sync and async counterparts, e.g. IEnumerable<T> and IAsyncEnumerable<T>. When consuming manually, the developer can easily distinguish which is being used based on naming (e.g. GetEnumerator vs GetAsyncEnumerator), and when consuming via the compiler, the compiler will provide syntax for differentiation (e.g. foreach vs foreach await).

What else will we want?
I've opened several additional issues to cover related support we'll want to consider:

• Extensions for ConfigureAwait (https://github.com/dotnet/corefx/issues/32684). The compiler will support pattern-based foreach await in addition to binding to the interface, and we can use that to enable the implicit awaits on MoveNextAsync and DisposeAsync to be done using ConfigureAwait(false) by having an extension method like public static ConfiguredAsyncEnumerable<T> ConfigureAwait<T>(this IAsyncEnumerable<T> enumerable, bool continueOnCapturedContext), where the returned ConfiguredAsyncEnumerable<T> will propagate that through to a ConfiguredAsyncEnumerator<T>, and its MoveNextAsync and DisposeAsync will return ConfiguredValueTaskAwaitables.
• ManualResetValueTaskSource (https://github.com/dotnet/corefx/issues/32664). If MoveNextAsync and DisposeAsync were defined to return Tasks, then the compiler could use TaskCompletionSource<T> to implement those async operations. But as it's returning ValueTask, and as we're doing so to enable object reuse, the compiler will be using its own implementation of IValueTaskSource. To greatly simplify that and to encapsulate all of the relevant logic, we should productize the ManualResetValueTaskSource/ManualResetValueTaskSourceLogic helper type from https://github.com/dotnet/corefx/blob/master/src/Common/tests/System/Threading/Tasks/Sources/ManualResetValueTaskSource.cs.
• IAsyncDisposable implementations (https://github.com/dotnet/corefx/issues/32665). With IAsyncDisposable exposed, we'll want to implement on a variety of types in coreclr/corefx where the type could benefit from having an asynchronous disposable capability in addition to an existing synchronous ability.

cc: @jcouv, @MadsTorgersen, @jaredpar, @terrajobst, @tarekgh, @kouvel

[tarekgh]

would be nice if you include some sample code for the usage.

[stephentoub]

e.g. (subject to decisions around language syntax)

static async IAsyncEnumerable<TResult> Zip<TFirst, TSecond, TResult>(
    IAsyncEnumerable<TFirst> first, IAsyncEnumerable<TSecond> second, Func<TFirst, TSecond, TResult> resultSelector)
{
    using await (IAsyncEnumerator<TFirst> e1 = first.GetAsyncEnumerator())
    using await (IAsyncEnumerator<TSecond> e2 = second.GetAsyncEnumerator())
    {
        while (await e1.MoveNextAsync() && await e2.MoveNextAsync())
            yield return resultSelector(e1.Current, e2.Current);
    }
}

[jnm2]

Too late to include these APIs in .NET Standard 2.1?

[clairernovotny]

@stephentoub Did we determine what was happening with Ix.NET Async from dotnet/reactive? Especially with the Impl here:

dotnet/reactive#423

Are you taking that code and putting it here in corefx? What about the existing package, etc? We should likely have a discussion on this?

[stephentoub]

Are you taking that code and putting it here in corefx? What about the existing package, etc? We should likely have a discussion on this?

When we've (you, me, @MadsTorgersen, etc.) talked about this previously, my understanding of the plan was that it would continue to live in reactive, and it would be updated based on the actual APIs shipping in the platform. I don't believe anything has changed there. Has it?

[jaredpar]

@jnm2

Too late to include these APIs in .NET Standard 2.1?

Yes. Particularly because the associated language feature isn't complete yet, and isn't intended to be complete until the .NET Core 3.0 time frame. Need to ship the API + language feature at the same time so they can evolve together.

[clairernovotny]

@stephentoub That's fine with me, it just wasn't clear if that was still the case since it has been a while. Once the core interfaces/compiler work is somewhere we can pull from a private feed, we can update that branch to use that interface and match the specs. Also happy to give direct commit access to any member on the team that needs it.

[stephentoub]

Thanks, @onovotny. @MadsTorgersen, this is all still correct, right?

[jcouv]

@stephentoub shared a code sample above using high-level primitives.
Here's a link to a lowered version of such an async-iterator method: https://gist.github.com/jcouv/ae7800985e3a8700bb84c6650d25bb69
This is still work-in-progress, but it shows how the new builder type would be called.

public class C
{
    public static async System.Collections.Generic.IAsyncEnumerable<int> M()
    {
        await System.Threading.Tasks.Task.CompletedTask;
        yield return 4;
    }
}

[akarnokd]

I have a couple of questions:

1. Do I have to defend against reentrant MoveNextAsync invocations? I.e., when the ValueTask<bool> completes and the continuation would call MoveNextAsync again?

2. DisposeAsync() must be called after MoveNextAsync's ValueTask<bool> completes and before the next MoveNextAsync would be invoked, right? If I wanted to implement a Timeout operator (with time) over this, how can I stop the main IAsyncEnumerator safely while the current ValueTask<bool> is not yet complete?

3. Cancellation must be external (via CancellationToken for example). How can I inject it to a specific instance returned by IAsyncEnumerable.GetAsyncEnumerator? In other words, how can I cancel individual runs of the IAsyncEnumerable?

4. If the cancellation has to be established on the source generator, how can I compose LINQ-style operators over such source and carry the cancellation support along with it?

5. Is it mandatory to call (and await) DisposeAsync() when the MoveNextAsync's ValueTask<bool> completes with false, indicating no more values?

6. If the ValueTask<bool> completes with false, should I defend against another invocation of MoveNextAsync and if so, what should I do: a) keep returning a false ValueTask, b) throw an invalid operation exception, c) have the ValueTask failing with the exception?

[jkotas]

we could also put it in System.Threading.Tasks.Extensions. If we want to OOB this, it could be in both, with the .NET Core S.T.T.Extensions just type forwarding to S.P.CoreLib as happens today for ValueTask.

System.Threading.Tasks.Extensions shipped inbox on at least one platform. IIRC, the rules for OOBing that we have agreed on some time ago say that you cannot add to the OOB assembly once it shipped inbox on at least one platform.

[stephentoub]

System.Threading.Tasks.Extensions shipped inbox on at least one platform. IIRC, the rules for OOBing that we have agreed on some time ago say that you cannot add to the OOB assembly once it shipped inbox on at least one platform.

Hmm. I had it in my head that you couldn't modify/augment APIs of any types that had shipped in such assemblies, but I hadn't internalized that you couldn't add additional unrelated types to them, either. What's the cause of that? @weshaggard, @terrajobst, is that actually the case such that if we wanted to go partial OOB here we'd need to introduce a new assembly?

[stephentoub]

Do I have to defend against reentrant MoveNextAsync invocations? I.e., when the ValueTask completes and the continuation would call MoveNextAsync again?

You need to expect that code will run synchronously as part of the completion of the ValueTask, so whatever code is completing the ValueTask needs to be ok with arbitrary code running as part of that, including another invocation of MoveNextAsync. If it's not ok with that, it needs to move the completion to a place where it is ok with that, which could include queueing it. For example, TaskCompletionSource<T> can be constructed to RunContinuationsAsynchronously, so if you were handing back a ValueTask that wrapped a TCS Task, you could set that option and then it wouldn't matter where you called {Try}Set{Result/Exception/Canceled}, at the expense of the continuation getting queued. I expect/hope that ManualResetValueTaskSource{Logic} will provide a similar option in some fashion.

DisposeAsync() must be called after MoveNextAsync's ValueTask completes and before the next MoveNextAsync would be invoked, right?

Yes. An implementation of the interfaces is not guaranteed to be thread-safe, and most will not be. So you should not make a second asynchronous call on the object while a first is still in progress.

If I wanted to implement a Timeout operator (with time) over this, how can I stop the main IAsyncEnumerator safely while the current ValueTask is not yet complete?

You can't:
https://blogs.msdn.microsoft.com/pfxteam/2012/10/05/how-do-i-cancel-non-cancelable-async-operations/
And see my comments on Cancellation in my original post. You can cancel or timeout your action of await'ing, but you can't influence the implementation you're awaiting unless you own it as well or unless it provides a means for doing so, such as accepting a CancellationToken into whatever method/ctor created the async method. If you cancel/timeout your await, then you shouldn't interact with that object again, and just let it get GC'd and any objects appropriately finalized.

Cancellation must be external (via CancellationToken for example). How can I inject it to a specific instance returned by IAsyncEnumerable.GetAsyncEnumerator? In other words, how can I cancel individual runs of the IAsyncEnumerable?

You can't. It'd be part of the enumerable, or else you can choose to cancel between awaits on your own. The alternative is to take a CancellationToken into MoveNextAsync, which does not compose well, in particular with language support, and which can add a lot of cost.

If the cancellation has to be established on the source generator, how can I compose LINQ-style operators over such source and carry the cancellation support along with it?

The cancellation would be built into the source. Any LINQ operations over it would implicitly inherit that as part of their awaiting on that source. If you wanted to further inject cancellation into individual query operators, then those query operators should accept a token as well, e.g. .FooAsync(..., token).BarAsync(..., token).

There's a whole discussion of cancellation at https://github.com/dotnet/csharplang/blob/master/proposals/async-streams.md#cancellation.

Is it mandatory to call (and await) DisposeAsync() when the MoveNextAsync's ValueTask completes with false, indicating no more values?

It's never mandatory to call Dispose, and it's never mandatory to call DisposeAsync. But it's a good idea to. And just as you'd want to call Dispose on an IEnumerator<T> today after its MoveNext returns false, you'd want to call DisposeAsync on an IAsyncEnumerator<T> after its MoveNextAsync hands back false.

If the ValueTask completes with false, should I defend against another invocation of MoveNextAsyncand if so, what should I do: a) keep returning a false ValueTask, b) throw an invalid operation exception, c) have the ValueTask failing with the exception?

Yes, it's valid (though uncommon) to do so. With IEnumerator<T>, calling MoveNext after you've already reached the end of the iteration is documented to return false. The same will be true for MoveNextAsync.

[jkotas]

I hadn't internalized that you couldn't add additional unrelated types to them, either. What's the cause of that?

The OOB would need to target a netstandard. If it does not target a netstandard, it is platform specific and we do not need unified type identity. It can be one type identity for netfx and different one for netcoreapp, etc. You can ship OOB like that, but it has limited value.

If the OOB targets netstandard, it should better work on all platforms that support netstandard, including .NET Core 2.1 where this shipped inbox. The problem is which implementation are you going to pick when somebody references the OOB in .NET Core 2.1. It cannot be the inbox implementation because of it is missing the new methods; and it cannot be the OOB implementation because of it would introduce second copy of the inbox types.

[stephentoub]

Thanks, @jkotas. That makes sense. We've made a bit of a mess, haven't we ;)

Ok, so if we decide we need to go OOB for platforms other than .NET Core 3, we'll need to introduce a new assembly.