Parallel helpers

ParallelHelpers.cs

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using MoreLinq;

namespace Payboard.Common
{
    public static class ParallelHelpers
    {
        /// <summary>
        ///     Iterate asynchronously in parallel over a data source, but only with a given degree of parallelization, so that we
        ///     don't DOS ourselves.
        /// </summary>
        public static async Task ForEachParallel<T>(this IEnumerable<T> list, Func<T, Task> action,
            int maxParallelization = 100)
        {
            // Create the execution block
            var options = new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = maxParallelization
            };
            var sendMessageBlock = new ActionBlock<T>(async item => { await action(item); }, options);

            // Send everything to the execution block and wait for it to finish
            list.ForEach(item => sendMessageBlock.Post(item));
            sendMessageBlock.Complete();
            await sendMessageBlock.Completion;
        }

        /// <summary>
        ///     Select asynchronously parallel from a data source, but only with a given degree of parallelization, so that we
        ///     don't DOS ourselves.
        /// </summary>
        public static async Task<List<TResult>> SelectParallel<TSource, TResult>(this IEnumerable<TSource> list,
            Func<TSource, Task<TResult>> mapFunc,
            int maxParallelization = 100)
        {
            // Create the execution block
            var options = new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = maxParallelization
            };
            var results = new List<TResult>();
            var sendMessageBlock = new ActionBlock<TSource>(async item =>
            {
                var result = await mapFunc(item);
                lock (results)
                {
                    results.Add(result);
                }
            }, options);

            // Send everything to the execution block and wait for it to finish
            list.ForEach(cu => sendMessageBlock.Post(cu));
            sendMessageBlock.Complete();
            await sendMessageBlock.Completion;
            return results;
        }
    }
}