[API Proposal]: Add an IAsyncEnumerable<T>.ToEnumerable extension method.

[davidfowl]

EDIT: See #60106 (comment) for an API proposal.

Background and motivation

Attempting to adapt an IAsyncEnumerable<T> to an IEnumerable<T>, Blocking collection seems to be the ideal type. It can be used to generate an enumerable that waits until the IAsyncEnumerable<T> produces items:

public static IEnumerable<T> ToEnumerable<T>(this IAsyncEnumerable<T> asyncEnumerable)
{
    var list = new BlockingCollection<T>();

    async Task AsyncIterate()
    {
        await foreach (var item in asyncEnumerable)
        {
            list.Add(item);
        }

        list.CompleteAdding();
    }

    _ = AsyncIterate();

    return list.GetConsumingEnumerable();
}

This works well but there's no way to propagate exceptions to the consuming enumerable that may happen when enumerating the IAsyncEnumerable<T>. I'd like to write something like this:

public static IEnumerable<T> ToEnumerable<T>(this IAsyncEnumerable<T> asyncEnumerable)
{
    var list = new BlockingCollection<T>();

    async Task AsyncIterate()
    {
        try
        {
            await foreach (var item in asyncEnumerable)
            {
                list.Add(item);
            }
            list.CompleteAdding();
        }
        catch(Exception ex)
        {
            list.CompleteAdding(ex);
        }
    }

    _ = AsyncIterate();

    return list.GetConsumingEnumerable();
}

CompleteAdding with an exception would propagate it to the caller and it would throw on synchronous enumeration. The only way to throw an exception today AFAIK is to use a cancellation token but that won't propagate the original exception.

API Proposal

namespace System.Collections.Concurrent
{
    public class BlockingCollection<T> : IEnumerable<T>
    {
        public void CompleteAdding(Exception error);
    }
}

API Usage

public static IEnumerable<T> ToEnumerable<T>(this IAsyncEnumerable<T> asyncEnumerable)
{
    var list = new BlockingCollection<T>();

    async Task AsyncIterate()
    {
        try
        {
            await foreach (var item in asyncEnumerable)
            {
                list.Add(item);
            }
            list.CompleteAdding();
        }
        catch(Exception ex)
        {
            list.CompleteAdding(ex);
        }
    }

    _ = AsyncIterate();

    return list.GetConsumingEnumerable();
}

Risks

None that I'm aware of.

Tagging subscribers to this area: @dotnet/area-system-collections
See info in area-owners.md if you want to be subscribed.

Issue Details

---

Background and motivation

Attempting to adapt an IAsyncEnumerable<T> to an IEnumerable<T>, Blocking collection seems to be the ideal type. It can be used to generate an enumerable that waits until the IAsyncEnumerable<T> produces items:

public static IEnumerable<T> ToEnumerable<T>(this IAsyncEnumerable<T> asyncEnumerable)
{
    var list = new BlockingCollection<T>();

    async Task AsyncIterate()
    {
        await foreach (var item in asyncEnumerable)
        {
            list.Add(item);
        }

        list.CompleteAdding();
    }

    _ = AsyncIterate();

    return list.GetConsumingEnumerable();
}

This works well but there's no way to propagate exceptions to the consuming enumerable that may happen when enumerating the IAsyncEnumerable<T>. I'd like to write something like this:

public static IEnumerable<T> ToEnumerable<T>(this IAsyncEnumerable<T> asyncEnumerable)
{
    var list = new BlockingCollection<T>();

    async Task AsyncIterate()
    {
        try
        {
            await foreach (var item in asyncEnumerable)
            {
                list.Add(item);
            }
            list.CompleteAdding();
        }
        catch(Exception ex)
        {
            list.CompleteAdding(ex);
        }
    }

    _ = AsyncIterate();

    return list.GetConsumingEnumerable();
}

CompleteAdding with an exception would propagate it to the caller and it would throw on synchronous enumeration. The only way to throw an exception today AFAIK is to use a cancellation token but that won't propagate the original exception.

API Proposal

namespace System.Collections.Concurrent
{
    public class BlockingCollection<T> : IEnumerable<T>
    {
        public void CompleteAdding(Exception error);
    }
}

API Usage

public static IEnumerable<T> ToEnumerable<T>(this IAsyncEnumerable<T> asyncEnumerable)
{
    var list = new BlockingCollection<T>();

    async Task AsyncIterate()
    {
        try
        {
            await foreach (var item in asyncEnumerable)
            {
                list.Add(item);
            }
            list.CompleteAdding();
        }
        catch(Exception ex)
        {
            list.CompleteAdding(ex);
        }
    }

    _ = AsyncIterate();

    return list.GetConsumingEnumerable();
}

Risks

None that I'm aware of.

Author:	davidfowl
Assignees:	-
Labels:	api-suggestion, area-System.Collections, untriaged
Milestone:	-

[davidfowl]

cc @stephentoub @tarekgh

[svick]

If some loss of performance was acceptable here, then I think a simple modification of the code would make this work:

public static IEnumerable<T> ToEnumerable<T>(this IAsyncEnumerable<T> asyncEnumerable)
{
    var list = new BlockingCollection<T>();

    async Task AsyncIterate()
    {
        try
        {
            await foreach (var item in asyncEnumerable)
            {
                list.Add(item);
            }
        }
        finally
        {
            list.CompleteAdding();
        }
    }

    var task = AsyncIterate();

    foreach (var item in list.GetConsumingEnumerable())
    {
        yield return item;
    }
    
    task.GetAwaiter().GetResult();
}

Does this make the proposal less useful?

---

Also, I don't like the suggested naming: CompleteAdding(Exception) does something quite different from CompleteAdding(), so I think it should have a different name. Inspiration could be taken from TPL Dataflow, where the corresponding methods are called Complete() and Fault(Exception).

[eiriktsarpalis]

I believe @svick's alternative has the added advantage of delaying async enumeration until blocking enumeration has started. Instead of retrofitting BlockingCollection<T>, wouldn't it perhaps make more sense to add bespoke extension methods for IAsyncEnumerable? For example:

namespace System.Collections.Generic
{

    public static class CollectionExtensions
    {
         public static IEnumerable<T> ToBlockingEnumerable<T>(this IAsyncEnumerable<T> source, CancellationToken cancellationToken = default);
         public static ValueTask<List<T>> ToListAsync<T>(this IAsyncEnumerable<T> source, CancellationToken cancellationToken = default);    
    }
}

[davidfowl]

@eiriktsarpalis sure, I wasn't sure about our tolerance for adding extensions to IAsyncEnumerable since we didn't add any to date 😄

[stephentoub]

I'd be more inclined to add a ToEnumerable extension method than to add a CompleteAdding(Exception) to BlockingCollection. We've avoided adding a full set of LINQ methods for IAsyncEnumerable, instead deferring to https://github.com/dotnet/reactive for that, but for helpers for going between core types, e.g. IAsyncEnumerable to IEnumerable, I'm ok with it. (And, just to be pedantic, we actually have added IAsyncEnumerable extension methods 😄 : https://github.com/dotnet/runtime/blob/6195d98cad42bc7ab70729e6c970e2c420b29db1/src/libraries/System.Private.CoreLib/src/System/Threading/Tasks/TaskAsyncEnumerableExtensions.cs)

I don't know, though, that BlockingCollection should be used in the implementation. If you really want these semantics, you can do what @svick has done, but this does effectively change the iteration semantics to be that once you start enumerating, background operations are filling up the collection concurrently with consumption of data, and that's generally not how either enumerables nor async enumerables work. On top of that, BlockingCollection, in large part due to its design of being able to wrap arbitrary collections and delegate to them for a variety of functionality, isn't terribly efficient, in particular from an allocation perspective when data isn't available. If we wanted to add a ToEnumerable method, I'd prefer to see us just doing something like:

public static IEnumerable<T> ToEnumerable<T>(this IAsyncEnumerable<T> source, CancellationToken cancellationToken = default)
{
    IAsyncEnumerator<T> e = source.GetAsyncEnumerator(cancellationToken);
    try
    {
        while (true)
        {
            ValueTask<bool> moveNext = e.MoveNextAsync();
            if (moveNext.IsCompletedSuccessfully ? moveNext.Result : moveNext.AsTask().GetAwaiter().GetResult())
            {
                yield return e.Current;
            }
            else break;            
        }
    }
    finally
    {
        e.DisposeAsync().AsTask().GetAwaiter().GetResult();
    }
}

That keeps everything in the world of "nothing is in flight while not interacting with the enumerator", avoids needing to expose some kind of bounded buffering policy on how many items a backing collection would enable, etc.

We'd also want to consider whether we really want to encourage such blocking behaviors by exposing this as an extension method... might be worth forcing someone to call it via a static just to add a tiny bit more friction. Or something along those lines. Or maybe @eiriktsarpalis's naming suggestion would help.

[davidfowl]

If some loss of performance was acceptable here, then I think a simple modification of the code would make this work:

Somewhat, but I can think of scenarios that are the synchronous version of Channel<T> that would like some form of exception propagation to the consumer without wrapping the enumerator. That's a little cumbersome.

[davidfowl]

That keeps everything in the world of "nothing is in flight while not interacting with the enumerator", avoids needing to expose some kind of bounded buffering policy on how many items a backing collection would enable, etc.

The most obvious implementation is usually the best one 😄.

I don't think it avoids a need to add CompleteAdding with an exception, or some way to push the exception to the caller, unless we're saying that BlockingCollection isn't something we want to modify. If that's the case, what's the blocking version of a Channel<T>?

[stephentoub]

If that's the case, what's the blocking version of a Channel?

channelReader.ReadAllAsync().ToEnumerable()
😄

[davidfowl]

Hmmm, performance wise, is it the same?

[theodorzoulias]

@davidfowl wouldn't the proposal have side effects on other members of the class? Like the BlockingCollection<T>.Take method for example. What if a consumer was blocked on the Take, while the CompleteAdding(ex) was invoked from another thread?

[davidfowl]

Then Take would throw an exception? What happens when CompletedAdding is called from another thread while Take was blocked?