Parallel tasks

CSharp · Console

/Program.cs

using System;
using System.Diagnostics;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Reflection;


namespace Console2017
{

    class ConsoleTimer : IDisposable
    {
        private readonly Stopwatch _stopwatch;

        public ConsoleTimer()
        {
            _stopwatch = new Stopwatch();
            _stopwatch.Start();
        }

        public void Dispose()
        {
            _stopwatch.Stop();
            Console.WriteLine("Duration: {0}ms", _stopwatch.ElapsedMilliseconds);
        }
    }

	
   public class Program
    {

        public static async Task<IEnumerable<TResult>> ExecTasksInParallelAsync1<TSource, TResult>(IEnumerable<TSource> source, Func<TSource, Task<TResult>> task, int minDegreeOfParallelism = 1, int maxDegreeOfParallelism = 1)
        {
            var allTasks = new List<Task<TResult>>();

            using (var throttler = new SemaphoreSlim(minDegreeOfParallelism, maxDegreeOfParallelism))
            {
                foreach (var element in source)
                {
                    // do an async wait until we can schedule again
                    await throttler.WaitAsync();

                    Func<Task<TResult>> func = async () =>
                    {
                        try
                        {
                            return await task(element);
                        }
                        finally
                        {
                            throttler.Release();
                        }
                    };

                    //Option 1
                    allTasks.Add(Task.Run(func));
                    //Option 2
                    //allTasks.Add(func.Invoke());
                }

                return await Task.WhenAll(allTasks);
            }
        }

        public static async Task<IEnumerable<TResult>> ExecTasksInParallelAsync2<TSource, TResult>(IEnumerable<TSource> source, Func<TSource, Task<TResult>> task, int minDegreeOfParallelism = 1, int maxDegreeOfParallelism = 1)
        {
            var allTasks = new List<Task<TResult>>();

            using (var throttler = new SemaphoreSlim(minDegreeOfParallelism, maxDegreeOfParallelism))
            {
                foreach (var element in source)
                {
                    // do an async wait until we can schedule again
                    await throttler.WaitAsync();

                    Func<Task<TResult>> func = async () =>
                    {
                        try
                        {
                            return await task(element);
                        }
                        finally
                        {
                            throttler.Release();
                        }
                    };

                    //Option 1
                    //allTasks.Add(Task.Run(func));
                    //Option 2
                    allTasks.Add(func.Invoke());
                }

                return await Task.WhenAll(allTasks);
            }
        }

        public static async Task<IEnumerable<TResult>> ExecTasksInParallelAsync3<TSource, TResult>(IEnumerable<TSource> source, Func<TSource, Task<TResult>> task, int minDegreeOfParallelism = 1, int maxDegreeOfParallelism = 1)
        {
            return source.AsParallel().WithDegreeOfParallelism(maxDegreeOfParallelism)
                .Select(s => task(s).GetAwaiter().GetResult());
        }


        public static void Main()
        {
            MainAsync().GetAwaiter().GetResult();
            Console.WriteLine("Press ENTER");
            Console.ReadLine();
        }
	   
	   public static async Task<string> DoSomethingToAString(string input)
	   {
		   await Task.Delay(1);
		   return input.ToUpper();
	   }

        public static async Task MainAsync()
        {
            var list = Enumerable.Range(1, 10000).Select(i => string.Format("Testing({0})...", i)).ToList();
            for (var run = 1; run<= 3; run++)
            {
                Console.WriteLine("Run {0}", run);
                Console.Write("Option 1: ");
                using (new ConsoleTimer())
                {
                    for (int trial = 0; trial < 10; trial++)
                    {
                        var output = await Program.ExecTasksInParallelAsync1<string, string>(list, DoSomethingToAString, 1, 4);
						output.ToList();
                    }
                }

                Console.Write("Option 2: ");
                using (new ConsoleTimer())
                {
                    for (int trial = 0; trial < 10; trial++)
                    {
                        var output = await Program.ExecTasksInParallelAsync2<string, string>(list, DoSomethingToAString, 1, 4);
                        output.ToList();
                    }
                }

                Console.Write("Option 3 :");
                using (new ConsoleTimer())
                {
                    for (int trial = 0; trial < 10; trial++)
                    {
                        var output = await Program.ExecTasksInParallelAsync3<string, string>(list, DoSomethingToAString, 1, 4);
                        output.ToList();
                    }
                }
            }
        }
    }
}