I have been looking for the best approach to creating an extension method that would give me the ability to select in a linq query using async/await with a max degree of parallelism. I'm aware that this can be achieved with the use of different libraries but wanted to create one without. Using postings I found online I made a few changes to convert a ForEachAsync to the following:

public static async Task<IEnumerable<TOut>> SelectAsync<TIn, TOut>(this IEnumerable<TIn> source, 
        Func<TIn, Task<TOut>> body, 
        int maxDegreeOfParallelism)
    var bag = new ConcurrentBag<TOut>();

    var selectAsync = Partitioner.Create(source)
        .Select(async partition => 
            using (partition)
                while (partition.MoveNext())
                    var item = await body(partition.Current);       

    await Task

    return bag;

Here is the unit test:

public async Task SelectAsync_EnsureMaxConcurrency_SuccessfulOnChecksAsync()
    // Arrange
    const int itemsMax = 10;
    const int dop = 5;
    const int delayInMs = 1000;

    var itemsToProcess = Enumerable
        .Range(1, itemsMax);

    // Act
    var started = DateTime.Now;

    var result = await itemsToProcess
        .SelectAsync(async item =>
            await Task.Delay(1000);
            return item;
        }, dop);

    var elapsed = (DateTime.Now - started)

    const double expected = (double)itemsMax / dop * delayInMs;

    // Assert
    Debug.WriteLine($"Elapsed:{elapsed} Expected:{expected}");

    Assert.AreEqual(itemsMax, result.Count());
    Assert.IsTrue(elapsed > expected);
  • 1
    \$\begingroup\$ Thread count (and thus max parallelism) isn't decided in code, it's decided in the environment. The same compiled application will run on different machines with different specs (and thus potentially different thread counts). Your application will always inherently receive however many threads the environment has deemed appropriate (or less if you simply don't have enough going on to warrant all the available threads).. For direct control, you'd need to use threads, not tasks. But this question feels more like an XY problem or a misunderstanding of asynchronicity. \$\endgroup\$
    – Flater
    Feb 7, 2020 at 23:36
  • \$\begingroup\$ @Flater I appreciate your input. This isn't an XY problem nor is it a misunderstanding of asynchronicity. What I'm attempting to accomplish is much like what is described in this SO answer stackoverflow.com/a/25877042/5121114 rather than simply return a Task, I want to return a Task<T> - moreover I understand that I don't have direct control with Tasks but I'm trying to throttle threads requested from the thread pool as described point 6 of this async/await antipattern blog posted by Mark Heath markheath.net/post/async-antipatterns \$\endgroup\$
    – TheRock
    Feb 8, 2020 at 5:31


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Browse other questions tagged or ask your own question.