using System;

using System.Collections.Generic;

using System.Data;

using System.Linq;

using System.Threading;

using System.Threading.Tasks;

public class Program

{

    public static void Main()

    {

        DataTable dt = ProcessAllAsync(Enumerable.Range(0, 10).Select(i => new InputObject { x = i })).Result;

        foreach (DataRow row in dt.Rows)

        {

            Console.WriteLine(row[0].ToString());   

        }

        Console.WriteLine("Done.");

    }

    struct InputObject

    {

        public int x;

    }

    struct FinalObject

    {

        public int x;

    }

    static FinalObject Process(InputObject o)

    {

        // Simulate synchronous work.

        Thread.Sleep(50);

        return new FinalObject { x = o.x };

    }

    static async Task<DataTable> ProcessAllAsync(IEnumerable<InputObject> allData)

    {

        DataTable table = CreateTable();

        int maxDegreeOfParallelism = Environment.ProcessorCount;

        await ParallelForEachAsync(

            allData,

            maxDegreeOfParallelism,

            // Loop body: these Tasks will run in parallel, up to {maxDegreeOfParallelism} at any given time.

            async dataObj =>

            {

                FinalObject o = await Task.Run(() => Process(dataObj)).ConfigureAwait(false); // Thread pool processing.

                await Task.Delay(50).ConfigureAwait(false); // Artificial throttling.

                return o;

            },

            // Completion handler: these will be executed one at a time, and can safely mutate shared state.

            moveObj => table.Rows.Add(moveObj.x)

        );

        return table;

    }

    static async Task ParallelForEachAsync<TInput, TResult>(IEnumerable<TInput> input,

                                                            int maxDegreeOfParallelism,

                                                            Func<TInput, Task<TResult>> body,

                                                            Action<TResult> onCompleted)

    {

        Queue<TInput> queue = new Queue<TInput>(input);

        if (queue.Count == 0) {

            return;

        }

        List<Task<TResult>> tasksInFlight = new List<Task<TResult>>(maxDegreeOfParallelism);

        do

        {

            while (tasksInFlight.Count < maxDegreeOfParallelism && queue.Count != 0)

            {

                TInput item = queue.Dequeue();

                Task<TResult> task = body(item);

                tasksInFlight.Add(task);

            }

            Task<TResult> completedTask = await Task.WhenAny(tasksInFlight).ConfigureAwait(false);

            tasksInFlight.Remove(completedTask);

            TResult result = completedTask.GetAwaiter().GetResult(); // We know the task has completed. No need for await.

            onCompleted(result);

        }

        while (queue.Count != 0 || tasksInFlight.Count != 0);

    }

    static DataTable CreateTable()

    {

        DataTable dt = new DataTable();

        dt.Columns.Add("x", typeof(int));

        return dt;

    }

}