Semaphore slim to handle throttling per time period

Question

I have a requirement from a client, to call their API, however, due to the throttling limit, we can only make 100 API calls in a minute. I am using SemaphoreSlim to handle that, Here is my code.

 async Task<List<IRestResponse>> GetAllResponses(List<string> locationApiCalls) 
 {
     var semaphoreSlim = new SemaphoreSlim(initialCount: 100, maxCount: 100);
     var failedResponses = new ConcurrentBag<IReadOnlyCollection<IRestResponse>>();
     var passedResponses = new ConcurrentBag<IReadOnlyCollection<IRestResponse>>();

     var tasks = locationApiCalls.Select(async locationApiCall =>
     {
          await semaphoreSlim.WaitAsync();
          try
          {
              var response = await RestApi.GetResponseAsync(locationApi);
              if (response.IsSuccessful)
              {
                  passedResponses.Add((IReadOnlyCollection<IRestResponse>)response);
              }
              else
              {
                 failedResponses.Add((IReadOnlyCollection<IRestResponse>)response);
              }
           }
           finally
           {
               semaphoreSlim.Release();
           }
     });

     await Task.WhenAll(tasks);
     var passedResponsesList = passedResponses.SelectMany(x => x).ToList();
}

However this line

var passedResponsesList = passedResponses.SelectMany(x => x).ToList();

never gets executed and I see Lots of failedResponses as well, I guess I have to add Task.Delay (for 1 minute) somewhere in the code as well.

Restricting queries to 100 _concurrent_ is not the same as restricting queries to 100 _per minute_. Without a proper [mcve], I don't see a way to debug your code, but even if it were possible to address the tasks that don't complete or fail, you'd still have the problem that you don't seem to have implementing time-based throttling properly. — Peter Duniho, May 17 '21 at 20:43
How many instances of your app run at once? Is this a console app? Web app? — mjwills, May 17 '21 at 21:53
@mjwills, this is a console application. It runs only one instance at a time. — Pankaj, May 17 '21 at 23:14
You could use the `RateLimiter` class found in [this](https://stackoverflow.com/questions/65825673/partition-how-to-add-a-wait-after-every-partition/65829971#65829971) answer. — Theodor Zoulias, May 18 '21 at 04:42
As a side note, the `ConcurrentBag` is not appropriate for the usage. A `ConcurrentQueue` would serve you better, because it preserves the order of the inserted items. The `ConcurrentBag` is a [highly specialized](https://stackoverflow.com/questions/15400133/when-to-use-blockingcollection-and-when-concurrentbag-instead-of-listt/64823123#64823123) class. It's not a thread-safe `List`. — Theodor Zoulias, May 18 '21 at 07:13
Time calculation is totally wrong. Statement {stopwatch.Elapsed - earliest} will calculate the time between time when an item was added to queue and read from queue. In the most cases it will be really small value. — Basil Kosovan, Mar 12 '22 at 22:38

Douglas · Accepted Answer · 2021-05-17T22:26:48.050

You need to keep track of the time when each of the previous 100 requests was executed. In the sample implementation below, the ConcurrentQueue<TimeSpan> records the relative completion time of each of these previous 100 requests. By dequeuing the first (and hence earliest) time from this queue, you can check how much time has passed since 100 requests ago. If it's been less than a minute, then the next request needs to wait for the remainder of the minute before it can be executed.

async Task<List<IRestResponse>> GetAllResponses(List<string> locationApiCalls)
{
    var semaphoreSlim = new SemaphoreSlim(initialCount: 100, maxCount: 100);
    var total = 0;
    var stopwatch = Stopwatch.StartNew();
    var completionTimes = new ConcurrentQueue<TimeSpan>();

    var failedResponses = new ConcurrentBag<IReadOnlyCollection<IRestResponse>>();
    var passedResponses = new ConcurrentBag<IReadOnlyCollection<IRestResponse>>();

    var tasks = locationApiCalls.Select(async locationApiCall =>
    {
        await semaphoreSlim.WaitAsync();

        if (Interlocked.Increment(ref total) > 100)
        {
            completionTimes.TryDequeue(out var earliest);
            var elapsed = stopwatch.Elapsed - earliest;
            var delay = TimeSpan.FromSeconds(60) - elapsed;
            if (delay > TimeSpan.Zero)
                await Task.Delay(delay);
        }

        try
        {
            var response = await RestApi.GetResponseAsync(locationApi);
            if (response.IsSuccessful)
            {
                passedResponses.Add((IReadOnlyCollection<IRestResponse>)response);
            }
            else
            {
                failedResponses.Add((IReadOnlyCollection<IRestResponse>)response);
            }
        }
        finally
        {
            completionTimes.Enqueue(stopwatch.Elapsed);
            semaphoreSlim.Release();
        }
    });

    await Task.WhenAll(tasks);
    var passedResponsesList = passedResponses.SelectMany(x => x).ToList();
}

If you're calling this method from the UI thread of a WinForms or WPF application, remember to add ConfigureAwait(false) to its await statements.

You should probably pay attention to the return value of the `completionTimes.TryDequeue` method. I assume that a `false` return value would reveal a bug/race-condition in your algorithm. I would suggest to add a `Debug.Assert(dequeued);` after this line, in order to (hopefully) notice such a bug during the development of the app. — Theodor Zoulias, May 18 '21 at 07:24
@Douglas, Thank you. I can now see all of the responses as passed ones (I checked by printing an index in with a Console.WriteLine ), However, My Code is still not returning value. It is a Console application, I tried adding ConfigureAwait(false) to await statrements, still no Luck. — Pankaj, May 18 '21 at 11:36
@TheodorZoulias: I agree with adding the defensive check, even if it would never fail under the above implementation. I left it out above just to keep the code simple. — Douglas, May 18 '21 at 17:54
@Pankaj: Do you have a return statement in your actual method? — Douglas, May 18 '21 at 17:56

Semaphore slim to handle throttling per time period

1 Answers1