From IEnumerable> to IAsyncEnumerable by yield returning inside a Parallel.ForEach/Parallel.ForEachAsync gives error CS1621

Question

In a .NET 6 project, I have to call a web API which is offset paginated (page/per page) and I would like to make the n calls parallel as far as possible.

This is the method which calls the API one time with the given page number:

private Task<ApiResponse> CallApiAsync(int page,
    CancellationToken cancellationToken = default)
{
    return GetFromJsonAsync<ApiResponse>($"...&page={page}", cancellationToken)
        .ConfigureAwait(false);
}

What I actually need is a forward only streamable iterator of all the API calls from page 1 to page n, so given this requirement, I thought IAsyncEnumerable was the right API to use so I could fire the API calls in parallel and access each API response as soon as one was ready, without needing all of them to be finished.

So I came up with the following code:

public async IAsyncEnumerable<ApiResponse> CallApiEnumerableAsync(int perPage,
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    int numProducts = GetNumberOfProducts(perPage);

    int numCalls = MathExtensions.CeilDiv(numProducts, perPage);

    var pages = Enumerable.Range(1, numCalls);

    Parallel.ForEach(pages, async page => {
        yield return await CallApiAsync(page, cancellationToken).ConfigureAwait(false);
    });

    yield break;
}

But I get the following error at yield: CS1621 - The yield statement cannot be used inside an anonymous method or lambda expression.
Is there a way to achieve the result I would like to get?
Feel free to ask questions if I wasn't clear enough!

Answer 1

The most readily available tool that you can use for this purpose is a TransformBlock<TInput,TOutput> from the TPL Dataflow library. This component is natively available in .NET Core and later, and it is essentially a processor/projector/transformer with two queues (input and output). You specify the processing function, then you configure the options according to your needs, then you feed it with data, and finally you retrieve the processed output:

public async IAsyncEnumerable<ApiResponse> CallApiEnumerableAsync(int perPage,
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    int numProducts = GetNumberOfProducts(perPage);
    int numCalls = MathExtensions.CeilDiv(numProducts, perPage);
    var pages = Enumerable.Range(1, numCalls);

    TransformBlock<int, ApiResponse> block = new(async page =>
    {
        return await CallApiAsync(page, cancellationToken);
    }, new ExecutionDataflowBlockOptions()
    {
        CancellationToken = cancellationToken,
        MaxDegreeOfParallelism = 10, // Configurable, the default is 1
        EnsureOrdered = true, // This is the default
    });

    // Feed the block with input data
    foreach (var page in pages) block.Post(page);
    block.Complete();

    // Emit the output data as they become available
    while (await block.OutputAvailableAsync())
        while (block.TryReceive(out var item))
            yield return item;

    // Propagate possible exception (including cancellation)
    await block.Completion;
}

This simple implementation initiates the TransformBlock when the resulting IAsyncEnumerable<ApiResponse> is enumerated, and it won't stop until either all the processing is completed, or the cancellationToken is canceled. The processing is not driven by the enumeration of the resulting sequence. It won't even stop if the client code simply abandons the enumeration, by breaking the await foreach loop. If you want to include this functionality (graceful termination), you will have to add a try-finally block, and an internal linked CancellationTokenSource as shown here. The yielding loop should be placed inside the try, and the cancellation of the linked CancellationTokenSource should be placed inside the finally.

Answer 2

You can't return from Parallel.ForEach/Parallel.ForEachAsync to the "outside" of the enclosing function, you provide the function which will be executed on every iteration and that's it.

One approach is to just loop through the data using SemaphoreSlim to limit number of concurrently executing tasks.

async IAsyncEnumerable<ApiResponse> CallApiEnumerableAsync(int perPage,
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    int numProducts = GetNumberOfProducts(perPage);
    int numCalls = MathExtensions.CeilDiv(numProducts, perPage);
    
    var linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
    var degreeOfParallelism = 2;
    var semaphoreSlim = new SemaphoreSlim(degreeOfParallelism);

    var tasks = Enumerable.Range(1, numCalls)
        .Select(async page =>
        {
            var linkedToken = linkedTokenSource.Token;
            await semaphoreSlim.WaitAsync(linkedToken);
            linkedToken.ThrowIfCancellationRequested();
            try
            {
                return await CallApiAsync(page, linkedToken);
            }
            catch
            {
                linkedTokenSource.Cancel();
                throw;
            }
            finally
            {
                semaphoreSlim.Release();
            }
        })
        .ToList();
    
    foreach (var page in tasks)
    {
        yield return await page;
    }
}

Answer 3

Parallel.ForEach is designed for in-memory CPU-bound operations, not for asynchronous IO.

It seems you can just use a simple for loop

public async IAsyncEnumerable<ApiResponse> CallApiEnumerableAsync(int perPage,
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    int numProducts = GetNumberOfProducts(perPage);

    int numCalls = MathExtensions.CeilDiv(numProducts, perPage);
    for (var page = 1; page <= numCalls; page++)
    {
        yield return await CallApiAsync(page, cancellationToken).ConfigureAwait(false);
    }
}

---

If you want to parallelize it (and I'm not sure that's actually a good idea with an API), you can perhaps use Tasks.

public async IAsyncEnumerable<ApiResponse> CallApiEnumerableAsync(int perPage,
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    int numProducts = GetNumberOfProducts(perPage);

    int numCalls = MathExtensions.CeilDiv(numProducts, perPage);
    
    var tasks = Enumerable.Range(1, numCalls)
        .Select(i => CallApiAsync(page, cancellationToken))
        .ToList();

    while(true)
    {
        var completed = await Task.WhenAny(tasks, cancellationToken).ConfigureAwait(false);
        tasks.Remove(completed);
        yield return completed.Result;
        if(tasks.Count == 0)
            break;
    }
}

This is a simplified version, and I may have left out some error handling, I'll leave you to improve that. You may want to use TransformBlock or similar in order to limit the concurrency, alternatively put a certain amount of Tasks into a list and wait for each one to finish on a one-out-one-in basis, using await Task.WhenAny.

Answer 4

If you want to spin tasks in parallel, calling await CallApiAsync(...) in a loop won't cut it. You have to use Task.Run(...). In it's simplest form:

var tasks = new List<Task>(pages.Length);
foreach (int page in pages)
{
    tasks.Add(
        Task.Run(() => CallApiAsync(page, cancellationToken).ConfigureAwait(false));
    );
}
await Task.WhenAll(tasks);

If you need to handle results per API call:

// within foreach loop:
Task.Run(async () => {
    var results = await CallApiAsync(page, cancellationToken).ConfigureAwait(false);
    // do something with `results`
});