using System;

using System.Threading.Tasks;

using System.Collections.Generic;

using System.Threading;

public class Program

{

    public async static Task Main()

    {

        var processor = new Processor();

        var cancelSource = new CancellationTokenSource();

        var numberOfInstances = 2;

        await processor.StartAsync(cancelSource.Token, numberOfInstances);

        await Task.Delay(2000, cancelSource.Token);

        cancelSource.Cancel();

    }

}

​

public class Processor

{

    private int _maxRegenerationTasksPerMinute = 5;

    private int recivedCount;

    private DateTime lastRegenerationPeriod;

    private static int usingDelay = 0;

    public async Task StartAsync(CancellationToken cancellationToken, int concurrentInstances)

    {

        await Concurrent.Run(ProcessSessionAsync, cancellationToken, concurrentInstances);

    }

    private async Task ProcessSessionAsync(string guid, CancellationToken cancellationToken)

    {

        Console.WriteLine($"{guid} Starting check for delay");

        await DelayForMaxRegenerationTasksPerMinutes(guid, cancellationToken);

        Console.WriteLine($"{guid} Starting actual processing");

        await Task.Delay(500, cancellationToken);

    }

    private async Task<bool> DelayForMaxRegenerationTasksPerMinutes(string guid, CancellationToken cancellationToken)

    {

        Console.WriteLine($"{guid} Pre-exchange usingDelay to 1 UsingDelay: {usingDelay}");

        if (0 == Interlocked.Exchange(ref usingDelay, 1))

        {

            Console.WriteLine($"{guid} Post-exchange usingDelay to 1 UsingDelay: {usingDelay}");

            if (DateTime.UtcNow > lastRegenerationPeriod)

            {

                Console.WriteLine($"{guid} Pre-Set LastRegenerationPeriod: {lastRegenerationPeriod}");

                lastRegenerationPeriod = DateTime.UtcNow.AddMinutes(1);

                Console.WriteLine($"{guid} Post-Set LastRegenerationPeriod: {lastRegenerationPeriod}");

                Console.WriteLine($"{guid} Pre-Set ReceivedCount: {recivedCount}");

                recivedCount = 1;

                Console.WriteLine($"{guid} Post-Set ReceivedCount: {recivedCount}");

            }

            else

            {

                Console.WriteLine($"{guid} ReceivedCount: {recivedCount}, MaxTasksPerMinute: {_maxRegenerationTasksPerMinute}");

                if (recivedCount < _maxRegenerationTasksPerMinute)

                {

                    Console.WriteLine($"{guid} Pre-Increment ReceivedCount: {recivedCount}");

                    recivedCount++;

                    Console.WriteLine($"{guid} Post-Increment ReceivedCount: {recivedCount}");

                }

                else

                {

                    var waitMilisecond = (int)(lastRegenerationPeriod - DateTime.UtcNow).TotalMilliseconds;

                    Console.WriteLine($"{guid} Delaying for {waitMilisecond}");

                    await Task.Delay(waitMilisecond, cancellationToken);

                }

            }

            Console.WriteLine($"{guid} Pre-exchange usingDelay to 0 UsingDelay: {usingDelay}");

            Interlocked.Exchange(ref usingDelay, 0);

            Console.WriteLine($"{guid} Post-exchange usingDelay to 0 UsingDelay: {usingDelay}");

        }

​

        return true;

    }

}

​

public static class Concurrent

{

    public static async Task Run(Func<string, CancellationToken, Task> method,

                                 CancellationToken cancellationToken,

                                 int concurrentInstances)

    {

        async Task<bool> WrappedAction(string guid, CancellationToken token)

        {

            await method(guid, cancellationToken);

            return true;

        }

​

        await Run(WrappedAction, cancellationToken, concurrentInstances);

    }

    private static async Task Run(Func<string, CancellationToken, Task<bool>> method,

                                  CancellationToken cancellationToken,

                                  int concurrentInstances)

    {

        List<Task> instances = new List<Task>();

​

        while (!cancellationToken.IsCancellationRequested)

        {

            instances.RemoveAll(task => task.IsCompleted);

​

            var desiredInstances = concurrentInstances;

            if (desiredInstances < 0)

                desiredInstances = 0;

​

            while (instances.Count < desiredInstances)

            {

                Console.WriteLine($"Instance Count: {instances.Count}, Desired Instances: {desiredInstances}");

                instances.Add(CreateMethodTask(method, cancellationToken));

            }

​

            if (instances.Count == 0)

                instances.Add(Task.Delay(TimeSpan.FromSeconds(10), cancellationToken));

​

            //Wait for any task to complete, then loop back.  Note that WhenAny suppresses exceptions.

            await Task.WhenAny(instances);

        }

​

        //Waits for all tasks to finish, suppressing any exceptions.

        await Task.WhenAny(Task.WhenAll(instances));

    }

    private static Task CreateMethodTask(Func<string, CancellationToken, Task<bool>> method,

                                         CancellationToken cancellationToken)

    {

        // Unique guid so we can correlate logs...

        var guid = Guid.NewGuid().ToString();

        Console.WriteLine($"{guid} Creating Task");

        return Task.Run(async () =>

                        {

                            var isWorkCompleted = await method(guid, cancellationToken);

                            if (isWorkCompleted)

                                await Task.Delay(TimeSpan.Zero, cancellationToken);

                        }, cancellationToken);

    }

}

​