Tasks not running concurrently by custom LINQ operator

Question

I am attempting to create a concurrent version of SelectAwait (and others) present as part of System.Linq.Async which provides extension methods to IAsyncEnumerable. This is the code that I am using:

private async IAsyncEnumerable<TOut> SelectParallelAsync<T, TOut>(
    this IAsyncEnumerable<T> enumerable, Func<T, Task<TOut>> predicate)
{
    var sem = new SemaphoreSlim(1, 10);
    
    var retVal = enumerable.Select(item => {
        var task = Task.Run(async () => {
            await sem.WaitAsync();
            var retVal = await predicate(item);
            sem.Release();

            return retVal;
        });

        return task;
    });

    await foreach (var item in retVal)
        yield return await item;
}

Enumerable is a simple enumerable from 0-1000. The code is being called as

.SelectParallelAsync(async i =>
{
    Console.WriteLine($"In Select : {i}");
    await Task.Delay(1000);
    return i + 5;
});

I was expecting all the tasks to get started immediately and being run 10 at a time. However, they get triggered one after another. Is there any way I can achieve something like this? Much appreciated.

EDIT: I am using semaphore instead of Parallel.ForEach or .AsParallel().WithMaxDegreeOfParallelism because I want to share this semaphore between multiple methods. Furthermore, PLINQ is not exactly very extendable and I can't add my own extension methods to it.

EDIT 2 : Added my own solution for sake of completion.

Answer 1

The enumeration of the source IAsyncEnumerable<T> enumerable is driven by the enumeration of the resulting AsyncEnumerable<TOut>. When the consumer of the resulting sequence requests the first TOut element of the sequence, at that point a T value will be requested from the source IAsyncEnumerable<T> enumerable. Then the value will be projected to a Task<TOut>, then this task will be awaited, and finally the result of the task will returned to the consumer. Everything happens sequentially. There is no concurrency. There is no internal activity before the consumer asks for an element, and after the element has been delivered to the consumer.

Adding concurrency to a LINQ operator is much more involved than it might appear from the first glance. It means that when the consumer asks for the first element, 10 tasks must start at once. And when any of these tasks completes, another task must start automatically in its place, without the consumer asking for it. And there must be a limit to how many tasks can be stored internally, that have not yet been requested by the consumer. And no more tasks should be started when this limit is reached, until the consumer takes one and creates an empty slot. And you must think what to do with the internal mechanism that actively starts the tasks and watches for their completion, in case the consumer decides that it had enough, and won't request any more elements (by exiting the consuming loop). And you must also think what to do with the stored tasks, in case the one that is about to be delivered to the consumer has failed. And what if more than one tasks have failed? And what to do in case the enumeration is canceled with a CancellationToken?

Doing all these correctly using only primitive tools like TaskCompletionSources and SemaphoreSlims, without using higher-level tools like the Channel<T>, is extremely difficult. If you are not familiar with the Channel<T>, my advice is to spend some time and familiarize yourself with it. It's a quite simple mechanism. If you know anything about the BlockingCollection<T> class, the Channel<T> is an asynchronous version of it.

In another question I have posted an AwaitResults method, that could be used to implement the SelectParallelAsync operator quite easily:

private IAsyncEnumerable<TOut> SelectParallelAsync<T, TOut>(
    this IAsyncEnumerable<T> enumerable, Func<T, Task<TOut>> predicate)
{
    return enumerable
        .Select(item => predicate(item))
        .AwaitResults(maxConcurrency: 10);
}

You could study that implementation, and change it to fit your needs.

Answer 2

For the sake of completion, I'd also like to add why the code in the question wasn't working and what I did to fix it.

The original created differed execution and when we did an await foreach, it would wait for each of the item in the enumerable to be completed before it can operate upon them (in this case, just return). As a result, the tasks were just getting fired one by one instead of being fired simultaneously. Another issue was my lack of understanding of how the semaphore works. I should have been using SemaphoreSlim(10) or something like that for multiple "threads" to access that code block at once.

To fix this, we need to start all the tasks at once and replace the semaphore piece. Adding a ToListAsync pretty much solves the first issue. It simply starts all the tasks at once. The final code which I went with looks something like this

public static async IAsyncEnumerable<TOut> SelectParallelAsync<TIn, TOut>(
this IAsyncEnumerable<TIn> enumerable, ParallelAsyncOptions options, Func<TIn, ValueTask<TOut>> predicate)
{
    var sem = new SemaphoreSlim(options.MaxDegreeOfParallelism, options.MaxDegreeOfParallelism);
    var retVal = await enumerable.Select(item => {
        return Task.Run(async () => {
            await sem.WaitAsync();
            var retVal = await predicate(item);
            sem.Release();

            return retVal;
        });
    }).ToListAsync();

    foreach (var item in retVal)
        yield return await item;    
}

I'd also like to add that the the other solution using channel works pretty well as well. A similar solution can however be achieved using a ConcurrentQueue using a "Complete" variable to correctly handle when the loop exits. The solution I proposed doesn't require the completion flag because that's inherently a part of IEnumerable or IAsyncEnumerable. Simply exiting the enumeration block completes the enumeration as well.

EDIT : Had a conversation (in comments) with the author of the other answer and he pointed out faults in this implementation. His implementation supports true asynchronous enumerable way better than mine does.