5
\$\begingroup\$

I want a rate-limiter that permits, at most, N operations over T seconds. I found a sample implementation here but it makes an assumption that operations complete in a predictable, consistent amount of time. I don't want to make that assumption.

My attempt to solve this simplifies the approach in the linked article by requiring the caller to signal when their operation completes. A timer fires every T seconds, releasing a SlimSemaphore the number of completed operations since the timer previously fired.

I'm looking for feedback on potential bugs, areas for improvement, and ways to make this more efficient.

using System;
using System.Threading;
using System.Threading.Tasks;
using System.Timers;
using Timer = System.Timers.Timer;

public class RateLimiter : IDisposable
{
    // Semaphore used to count and limit the number of occurrences per unit time.
    private readonly SemaphoreSlim semaphore;

    // Timer used to trigger exiting the semaphore.
    private readonly Timer batchTimer;

    // Whether this instance is disposed.
    private bool isDisposed;

    private int countCompleted;

    /// <summary>
    /// Number of occurrences allowed per unit of time.
    /// </summary>
    public int Occurrences { get; private set; }

    /// <summary>
    /// The length of the time unit, in milliseconds.
    /// </summary>
    public int TimeUnitMilliseconds { get; private set; }

    public RateLimiter(int occurrences, TimeSpan timeUnit)
    {
        // Check the arguments.
        if (occurrences <= 0)
        {
            throw new ArgumentOutOfRangeException(nameof(occurrences), "Number of occurrences must be a positive integer");
        }

        if (timeUnit != timeUnit.Duration())
        {
            throw new ArgumentOutOfRangeException(nameof(timeUnit), "Time unit must be a positive span of time");
        }

        if (timeUnit >= TimeSpan.FromMilliseconds(uint.MaxValue))
        {
            throw new ArgumentOutOfRangeException(nameof(timeUnit), "Time unit must be less than 2^32 milliseconds");
        }

        this.Occurrences = occurrences;
        this.TimeUnitMilliseconds = (int)timeUnit.TotalMilliseconds;
        Interlocked.Exchange(ref this.countCompleted, 0);

        // Create the semaphore, with the number of occurrences as the maximum count.
        this.semaphore = new SemaphoreSlim(this.Occurrences, this.Occurrences);

        // Create a timer to exit the semaphore. Use the time unit as the original
        // interval length because that's the earliest we will need to exit the semaphore.
        this.batchTimer = new Timer(this.TimeUnitMilliseconds);
        this.batchTimer.Elapsed += this.OnBatchTimerElapsed;
        this.batchTimer.AutoReset = true;
        this.batchTimer.Enabled = true;
    }

    private void OnBatchTimerElapsed(object sender, ElapsedEventArgs e)
    {
        this.semaphore.Release(Interlocked.Exchange(ref this.countCompleted, 0));
    }

    public void Completed()
    {
        Interlocked.Increment(ref this.countCompleted);
    }

    /// <summary>
    /// Blocks the current thread until allowed to proceed, or until the specified timeout elapses.
    /// </summary>
    /// <param name="millisecondsTimeout">Number of milliseconds to wait, or -1 to wait indefinitely.</param>
    /// <returns>true if the thread is allowed to proceed, or false if timed out</returns>
    public async Task<bool> WaitToProceed(int millisecondsTimeout)
    {
        // Check the arguments.
        if (millisecondsTimeout < -1)
        {
            throw new ArgumentOutOfRangeException("millisecondsTimeout");
        }

        this.CheckDisposed();

        // Block until we can enter the semaphore or until the timeout expires.
        return await this.semaphore.WaitAsync(millisecondsTimeout);
    }

    /// <summary>
    /// Blocks the current thread indefinitely, until allowed to proceed.
    /// </summary>
    public async Task<bool> WaitToProceed()
    {
        return await this.WaitToProceed(Timeout.Infinite);
    }

    // Throws an ObjectDisposedException if this object is disposed.
    private void CheckDisposed()
    {
        if (this.isDisposed)
        {
            throw new ObjectDisposedException("RateLimiter is already disposed");
        }
    }

    /// <summary>
    /// Releases unmanaged resources held by an instance of this class.
    /// </summary>
    public void Dispose()
    {
        this.Dispose(true);
        GC.SuppressFinalize(this);
    }

    /// <summary>
    /// Releases unmanaged resources held by an instance of this class.
    /// </summary>
    /// <param name="isDisposing">Whether this object is being disposed.</param>
    protected virtual void Dispose(bool isDisposing)
    {
        if (this.isDisposed)
        {
            if (isDisposing)
            {
                // The semaphore and timer both implement IDisposable and 
                // therefore must be disposed.
                this.semaphore.Dispose();
                this.batchTimer.Dispose();

                this.isDisposed = true;
            }
        }
    }
}

Sample Usage:

var tasks = new List<Task>();

// Can process up to 40 operations per 10 seconds.
var rateLimiter = new RateLimiter(40, TimeSpan.FromSeconds(10));

for (int i = 0; i < 90; i++)
{
    tasks.Add(Task.Run(async () =>
    {
        await rateLimiter.WaitToProceed();
        // do stuff
        await Task.Delay(2000);
        rateLimiter.Completed();
    }));
}

await Task.WhenAll(tasks);
\$\endgroup\$
  • \$\begingroup\$ One potential problem is that the user doesn't place the Completed function in a finally block. So if the // do stuff portion throws, a semaphore count is never released. \$\endgroup\$ – Craig Jun 8 at 3:29
5
\$\begingroup\$

IMO the use of this. before members in the methods reduces readability and is unnecessary.


Occurrences and TimeUnitMilliseconds should be immutable and only settable in the constructor so make it private int Occurrences {get;} or private readonly int Occurrences Alternatively you can make it public settable to allow for dynamic configuration?


protected virtual void Dispose(bool isDisposing)
{
  if (this.isDisposed)
  {
    if (isDisposing)
    {

I think the if (this.isDisposed) should be if (!isDisposed)


  if (timeUnit >= TimeSpan.FromMilliseconds(uint.MaxValue))
  {
    throw new ArgumentOutOfRangeException(nameof(timeUnit), "Time unit must be less than 2^32 milliseconds");
  }
  Occurrences = occurrences;
  TimeUnitMilliseconds = (int)timeUnit.TotalMilliseconds;

Here you test timeUnit against uint.MaxValue and after that the value is cast to int. If int.MaxValue < timeUnit < uint.MaxValue the test is passed but the cast will result in a negative TimeUnitMilliseconds.


Your test case doesn't dispose the RateLimiter instance.


semaphore.Release(Interlocked.Exchange(ref countCompleted, 0));

If countCompleted == 0 this will fail. It can happen if no actions haven't terminated before the timer ticks.


GC.SuppressFinalize(this);

This has no meaning if the class doesn't have a finalizer (~RateLimiter()) GC.SuppressFinalize(Object) Method


A way to avoid that the users fail to call Completed() is to let RateLimiter handle the execution of the methods it self in a pattern like:

public class RateLimiter : IDisposable
{
  async public Task Run(Func<Task> func)
  {
    try
    {
      await WaitToProceed();
      await func();
    }
    finally
    {
      Completed();
    }
  }

  ....
}

And then make WaitToProceed() and Completed() as private members.

You'll of course need to provide serveral overloads of the above, that take a timeout value and/or cancellation token and/or have return values etc. And the class should maybe be renamed to something more suitable.

\$\endgroup\$
  • \$\begingroup\$ In your last point, if I use a Func, will exceptions thrown from func() bubble up to the caller of Run(). \$\endgroup\$ – Craig Jun 8 at 16:15
  • \$\begingroup\$ @Craig: Yes it will, but you can also catch exceptions inside Run() if you want to know about them there - and optionally rethrow them to be handled by the caller. All exceptions will be collected by the system and delivered to the caller in one AggregateException \$\endgroup\$ – Henrik Hansen Jun 8 at 16:28

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.