using System;

using System.Collections.Generic;

using System.Linq;

using System.Threading;

using System.Threading.Tasks;

​

public class Program

{

    public static async Task Main()

    {

        var rateLimiter = new RateLimiter(3, TimeSpan.FromSeconds(3));

        var task1 = Task.Run(async () =>

        {

            for (var i = 0; i < 5; i++)

            {

                await rateLimiter.WaitAsync();

                Console.WriteLine($"{Thread.CurrentThread.ManagedThreadId} : {DateTime.Now}");

            }

        }

​

        );

        var task2 = Task.Run(async () =>

        {

            for (var i = 0; i < 5; i++)

            {

                await rateLimiter.WaitAsync();

                Console.WriteLine($"{Thread.CurrentThread.ManagedThreadId} : {DateTime.Now}");

            }

        }

​

        );

        await Task.WhenAll(task1, task2);

    }

​

    public class RateLimiter

    {

        private readonly CancellationToken _cancellationToken;

        private readonly int _count;

        private readonly Queue<DateTime> _completed = new Queue<DateTime>();

        private readonly Queue<TaskCompletionSource<bool>> _queue = new Queue<TaskCompletionSource<bool>>();

        private readonly object _sync = new object ();

        private readonly TimeSpan _timeSpan;

        private int _isProcessing;

        public RateLimiter(int count, TimeSpan timeSpan, CancellationToken cancellationToken = default)

        {

            _count = count;

            _timeSpan = timeSpan;

            _cancellationToken = cancellationToken;

        }

​

        private void Cleanup()

        {

            lock (_sync)

            {

                _completed.Clear();

                while (_cancellationToken.IsCancellationRequested && _queue.Any())

                    _queue.Dequeue().TrySetCanceled();

                _queue.Clear();

                Interlocked.Exchange(ref _isProcessing, 0);

            }

        }

​

        private async Task ProcessAsync()

        {

            try

            {

                while (true)

                {

                    _cancellationToken.ThrowIfCancellationRequested();

                    var time = DateTime.Now - _timeSpan;

                    TimeSpan delay;

                    lock (_sync)

                    {

                        while (_completed.Any() && _completed.Peek() < time)

                            _completed.Dequeue();

                        while (_completed.Count < _count && _queue.Any())

                        {

                            _queue.Dequeue().TrySetResult(true);

                            _completed.Enqueue(DateTime.Now);

                        }

​

                        if (!_queue.Any() || !_completed.Any())

                        {

                            Cleanup();

                            break;

                        }

​

                        delay = (_completed.Peek() - time) + TimeSpan.FromMilliseconds(20);

                    }

​

                    if (delay.Ticks > 0)

                        await Task.Delay(delay, _cancellationToken);

                }

            }

            catch (OperationCanceledException)

            {

                Cleanup();

            }

        }

​

        public ValueTask WaitAsync()

        {

            _cancellationToken.ThrowIfCancellationRequested();

            lock (_sync)

            {

                if (_completed.Count < _count && !_queue.Any())

                {

                    _completed.Enqueue(DateTime.Now);

                    return new ValueTask();

                }

​

                try

                {

                    var tcs = new TaskCompletionSource<bool>();

                    _queue.Enqueue(tcs);

                    return new ValueTask(tcs.Task);

                }

                finally

                {

                    if (Interlocked.CompareExchange(ref _isProcessing, 1, 0) == 0)

                        _ = ProcessAsync();

                }

            }

        }

    }

}