Limiting number of API requests started per second using Parallel.ForEach

Question

I am working on improving some of my code to increase efficiency. In the original code I was limiting the number of threads allowed to be 5, and if I had already 5 active threads I would wait until one finished before starting another one. Now I want to modify this code to allow any number of threads, but I want to be able to make sure that only 5 threads get started every second. For example:

• Second 0 - 5 new threads
• Second 1 - 5 new threads
• Second 2 - 5 new threads ...

Original Code (cleanseDictionary contains usually thousands of items):

        ConcurrentDictionary<long, APIResponse> cleanseDictionary = new ConcurrentDictionary<long, APIResponse>();
        ConcurrentBag<int> itemsinsec = new ConcurrentBag<int>();
        ConcurrentDictionary<long, string> resourceDictionary = new ConcurrentDictionary<long, string>();
        DateTime start = DateTime.Now;

        Parallel.ForEach(resourceDictionary, new ParallelOptions { MaxDegreeOfParallelism = 5 }, row =>
        {
            lock (itemsinsec)
            {
                ThrottleAPIRequests(itemsinsec, start);

                itemsinsec.Add(1);
            }

            cleanseDictionary.TryAdd(row.Key, _helper.MakeAPIRequest(string.Format("/endpoint?{0}", row.Value)));
        });


    private static void ThrottleAPIRequests(ConcurrentBag<int> itemsinsec, DateTime start)
    {
        if ((start - DateTime.Now).Milliseconds < 10001 && itemsinsec.Count > 4)
        {
            System.Threading.Thread.Sleep(1000 - (start - DateTime.Now).Milliseconds);
            start = DateTime.Now;
            itemsinsec = new ConcurrentBag<int>();
        }
    }

My first thought was increase the MaxDegreeofParallelism to something much higher and then have a helper method that will limit only 5 threads in a second, but I am not sure if that is the best way to do it and if it is, I would probably need a lock around that step?

Thanks in advance!

EDIT I am actually looking for a way to throttle the API Requests rather than the actual threads. I was thinking they were one in the same.

Edit 2: My requirements are to send over 5 API requests every second

Answer 1

"Parallel.ForEach" from the MS website

may run in parallel

If you want any degree of fine control over how the threads are managed, this is not the way.
How about creating your own helper class where you can queue jobs with a group id, allows you to wait for all jobs of group id X to complete, and it spawns extra threads as and when required?

Answer 2

For me the best solution is:

using System;
using System.Collections.Concurrent;
using System.Threading.Tasks;

namespace SomeNamespace
{
    public class RequestLimiter : IRequestLimiter
    {
        private readonly ConcurrentQueue<DateTime> _requestTimes;
        private readonly TimeSpan _timeSpan;

        private readonly object _locker = new object();

        public RequestLimiter()
        {
            _timeSpan = TimeSpan.FromSeconds(1);
            _requestTimes = new ConcurrentQueue<DateTime>();
        }

        public TResult Run<TResult>(int requestsOnSecond, Func<TResult> function)
        {
            WaitUntilRequestCanBeMade(requestsOnSecond).Wait();
            return function();
        }

        private Task WaitUntilRequestCanBeMade(int requestsOnSecond)
        {
            return Task.Factory.StartNew(() =>
            {
                while (!TryEnqueueRequest(requestsOnSecond).Result) ;
            });
        }

        private Task SynchronizeQueue()
        {
            return Task.Factory.StartNew(() =>
            {
                _requestTimes.TryPeek(out var first);

                while (_requestTimes.Count > 0 && (first.Add(_timeSpan) < DateTime.UtcNow))
                    _requestTimes.TryDequeue(out _);
            });
        }

        private Task<bool> TryEnqueueRequest(int requestsOnSecond)
        {
            lock (_locker)
            {
                SynchronizeQueue().Wait();
                if (_requestTimes.Count < requestsOnSecond)
                {
                    _requestTimes.Enqueue(DateTime.UtcNow);
                    return Task.FromResult(true);
                }
                return Task.FromResult(false);
            }
        }
    }
}

Answer 3

I want to be able to send over 5 API request every second

That's really easy:

while (true) {
 await Task.Delay(TimeSpan.FromSeconds(1));
 await Task.WhenAll(Enumerable.Range(0, 5).Select(_ => RunRequestAsync()));
}

Maybe not the best approach since there will be a burst of requests. This is not continuous.

Also, there is timing skew. One iteration takes more than 1 second. This can be solved with a few lines of time logic.