15
\$\begingroup\$

The idea of this class is that several threads are sending data over a network and each thread are sharing the same instance of this class and before sending N bytes over the network each thread is calling ThrottledWait(n).

My worry is that each thread might run on different core and get different value for DateTime.UtcNow.Ticks. I am not 100% sure it's thread-safe.

Also calling Thread.Sleep(ts) might sleep for longer than asked for and might cause traffic to not be smooth because of aliasing so we might want to do a thread.sleep() for less than the calculated amount and waste the remaining time checking DateTime.UtcNow.Ticks in a busy loop.

public class Throttler
{
    // Use this constant as average rate to disable throttling
    public const long NoLimit = -1;
    // Number of consumed tokens
    private long _consumedTokens;
    // timestamp of last refill time
    private long _lastRefillTime;
    // ticks per period
    private long _periodTicks;

    private double _averageRate;

    public long BurstSize
    {
        get;
        set;
    }

    public long AverageRate
    {
        get { return (long)_averageRate; }
        set { _averageRate = value; }
    }

    public TimeSpan Period
    {
        get
        {
            return new TimeSpan(_periodTicks);
        }
        set
        {
            _periodTicks = value.Ticks;
        }
    }
    public Throttler()
    {
        BurstSize = 1;
        AverageRate = NoLimit;
        Period = TimeSpan.FromSeconds(1);
    }

    /// <summary>
    /// Create a Throttler
    /// ex: To throttle to 1024 byte per seconds with burst of 200 byte use
    /// new Throttler(1024,TimeSpan.FromSeconds(1), 200);
    /// </summary>
    /// <param name="averageRate">The number of tokens to add to the bucket every interval. </param>
    /// <param name="period">Timespan of on interval.</param>
    /// <param name="burstSize"></param>
    public Throttler(long averageRate, TimeSpan period, long burstSize = 1)
    {
        BurstSize = burstSize;
        AverageRate = averageRate;
        Period = period;
    }

    public bool TryThrottledWait(long amount)
    {
        if (BurstSize <= 0 || _averageRate <= 0)
        { // Instead of throwing exception, we just let all the traffic go
            return true;
        }
        RefillToken();
        return ConsumeToken(amount);
    }

    private bool ConsumeToken(long amount)
    {
        while (true)
        {
            long currentLevel = System.Threading.Volatile.Read(ref _consumedTokens);
            if (currentLevel + amount > BurstSize)
            {
                return false; // not enough space for amount token
            }

            if (Interlocked.CompareExchange(ref _consumedTokens, currentLevel + amount, currentLevel) == currentLevel)
            {
                return true;
            }
        }
    }

    public void ThrottledWait(long amount)
    {
        while (true) {
            if (TryThrottledWait(amount))
            {
                break;
            }

            long refillTime = System.Threading.Volatile.Read(ref _lastRefillTime);
            long nextRefillTime = (long) (refillTime + (_periodTicks / _averageRate));
            long currentTimeTicks = DateTime.UtcNow.Ticks;
            long sleepTicks = Math.Max(nextRefillTime - currentTimeTicks, 0);
            TimeSpan ts = new TimeSpan(sleepTicks);
            Thread.Sleep(ts);
        } 
    }

    /// <summary>
    /// Compute elapsed time using DateTime.UtcNow.Ticks and refil token using _periodTicks and _averageRate
    /// </summary>
    private void RefillToken()
    {
        long currentTimeTicks = DateTime.UtcNow.Ticks;
        // Last refill time in  ticks unit
        long refillTime = System.Threading.Volatile.Read(ref _lastRefillTime);
        // Time delta in ticks unit
        long TicksDelta = currentTimeTicks - refillTime;
        long newTokens = (long)(TicksDelta * _averageRate / _periodTicks);
        if (newTokens > 0)
        {
            long newRefillTime = refillTime == 0
                ? currentTimeTicks
                : refillTime + (long)(newTokens * _periodTicks / _averageRate);

            if (Interlocked.CompareExchange(ref _lastRefillTime, newRefillTime, refillTime) == refillTime)
            {
                // Loop until we succeed in refilling "newTokens" tokens
                while (true)
                {
                    long currentLevel = System.Threading.Volatile.Read(ref _consumedTokens);
                    long adjustedLevel = (long)Math.Min(currentLevel, BurstSize); // In case burstSize decreased
                    long newLevel = (long) Math.Max(0, adjustedLevel - newTokens);
                    if (Interlocked.CompareExchange(ref _consumedTokens, newLevel, currentLevel) == currentLevel)
                    {
                        return;
                    }
                }
            }
        }
    }
}

To throttle to 1024 byte per seconds with burst of 200 byte we would do

var throttler = new Throttler(1024,TimeSpan.FromSeconds(1), 200);

Then each time we need to send some byte

void Sendbytes(byte[] byteArray) {
    throttler.ThrottledWait(byteArray.Length);
    ...
    // write the bytes
}
\$\endgroup\$
4
+100
\$\begingroup\$

It's hard to be sure if the implementation is correct because there's no high-level documentation to explain what the expected behaviour is. If I were assigned this as a maintenance programmer then I would look at

        long currentLevel = System.Threading.Volatile.Read(ref _consumedTokens);
        if (currentLevel + amount > BurstSize)
        {
            return false; // not enough space for amount token
        }

and assume there's a bug. (Surely bursts are about exceeding the long-term average, and in the example given of a long-term average of 1024 bytes per second with burst of 200 bytes we wouldn't be able to write 256 bytes per second if we tried to do it in one packet per second). The documentation needs to be a lot clearer if you want anyone else to use it, including yourself in six months' time.


What's going on with this?

        if (Interlocked.CompareExchange(ref _lastRefillTime, newRefillTime, refillTime) == refillTime)
        {
            // Loop until we succeed in refilling "newTokens" tokens
            while (true)
            {
                long currentLevel = System.Threading.Volatile.Read(ref _consumedTokens);
                long adjustedLevel = (long)Math.Min(currentLevel, BurstSize); // In case burstSize decreased
                long newLevel = (long) Math.Max(0, adjustedLevel - newTokens);
                if (Interlocked.CompareExchange(ref _consumedTokens, newLevel, currentLevel) == currentLevel)
                {
                    return;
                }
            }
        }

The outer if seems to be effectively "If no other thread is interlaced with us and has beaten us to the update"; then the inner while / if seems to assume that we're still interlaced with another thread which has also made it past the outer if.

Either the outer if is an effective guard, in which case the body can be simplified to a direct assignment; or it isn't, in which case it needs to be replaced with one.

Although I'm not quite sure what you're trying to do, I'm almost certain that Monitor.TryEnter would be a better way of doing it than Interlocked.CompareExchange.

\$\endgroup\$
  • \$\begingroup\$ your understanding of how Interlocked.CompareExchange is used is correct. Monitor.TryEnter would also work but was measured to be too slow for throttling because it try to take a lock. So the logic is if we successfully update the _lastRefillTime atomically we try to decrement newTokens to _consumedTokens atomically until we succeed \$\endgroup\$ – skyde Mar 22 '17 at 17:54
  • 1
    \$\begingroup\$ @skyde, if you've tried Monitor.TryEnter then fair enough, but add a comment saying that it was too slow so that the maintenance programmer knows not to try it. \$\endgroup\$ – Peter Taylor Mar 22 '17 at 21:03
  • \$\begingroup\$ I updated the code in the question. it would now support writing 256 bytes per second if we tried to do it in one packet per second \$\endgroup\$ – skyde Apr 7 '17 at 22:19
  • \$\begingroup\$ @skyde please refrain from doing this when someone answered your question. Instead you can i.e. post a new answer with the reviewed code \$\endgroup\$ – t3chb0t Apr 8 '17 at 15:20
6
\$\begingroup\$

Take this answer as general advice.

I see that the nesting seems to be a bit deeper than what people normally see. Most of the time it is not a problem. However, sometimes placing a lot of code in a deeply indented block might force the reader to scroll to the right to view the rest of the code. It's not significant but it can save a few seconds of scrolling for the person who reads your code (maybe even yourself).

There are a few ways to tackle this problem. I'll demonstrate with your code. Consider the following snippet:

    if (newTokens > 0)
    {
        long newRefillTime = refillTime == 0
            ? currentTimeTicks
            : refillTime + (long)(newTokens * _periodTicks / _averageRate);

        if (Interlocked.CompareExchange(ref _lastRefillTime, newRefillTime, refillTime) == refillTime)
        {
            // Loop until we succeed in refilling "newTokens" tokens
            while (true)
            {
                long currentLevel = System.Threading.Volatile.Read(ref _consumedTokens);
                long adjustedLevel = (long)Math.Min(currentLevel, BurstSize); // In case burstSize decreased
                long newLevel = (long) Math.Max(0, adjustedLevel - newTokens);
                if (Interlocked.CompareExchange(ref _consumedTokens, newLevel, currentLevel) == currentLevel)
                {
                    return;
                }
            }
        }
    }

The outermost if block doesn't have an else. One thing we can do is invert the condition and return.

        if (newTokens <= 0)
        {
            return;
        }
        // rest of the code

There is an immediate improvement. We can extend this further.

        if (Interlocked.CompareExchange(ref _lastRefillTime, newRefillTime, refillTime) != refillTime)
        {
            return;
        }
        // Loop until we succeed in refilling "newTokens" tokens
        while (true)
        {
            long currentLevel = System.Threading.Volatile.Read(ref _consumedTokens);
            long adjustedLevel = (long) Math.Min(currentLevel, BurstSize); // In case burstSize decreased
            long newLevel = (long) Math.Max(0, adjustedLevel - newTokens);
            if (Interlocked.CompareExchange(ref _consumedTokens, newLevel, currentLevel) == currentLevel)
            {
                return;
            }
        }

It is easiest for if statements that are near the end of a method and lack an else. For other cases, it might not be possible. I generally try to detect the cases that I don't need to handle and exit quickly, rather than check if I need to execute a piece of code. The idea is to keep on performing eliminations until we have the ideal conditions necessary for the main code to work.

For the while loop, you could use a do {...} while (...); loop. Lesser code and exactly what you intend to do anyway:

        // Loop until we succeed in refilling "newTokens" tokens
        do
        {
            long currentLevel = System.Threading.Volatile.Read(ref _consumedTokens);
            long adjustedLevel = (long) Math.Min(currentLevel, BurstSize); // In case burstSize decreased
            long newLevel = (long) Math.Max(0, adjustedLevel - newTokens);
        }
        while (Interlocked.CompareExchange(ref _consumedTokens, newLevel, currentLevel) != currentLevel);

Performance remains fairly the same. It might seem unnecessary but it helps prevent monstrosities like:

enter image description here

A little goes a long way.

\$\endgroup\$
3
\$\begingroup\$

You should only ever have one constructor that actually sets any fields.

public Throttler()
{
    BurstSize = 1;
    AverageRate = NoLimit;
    Period = TimeSpan.FromSeconds(1);
}

public Throttler(long averageRate, TimeSpan period, long burstSize = 1)
{
    BurstSize = burstSize;
    AverageRate = averageRate;
    Period = period;
}

Notice how BurstSize defaults to 1 in both cases, but you've done it differently? Inconsistencies that this can be a real killer. You can clean it up with some ctor chaining.

public Throttler()
    :this(NoLimit, TimeSpan.FromSeconds(1))
{ }

public Throttler(long averageRate, TimeSpan period)
    :this(averageRate, period, burstSize: 1)
{ }

public Throttler(long averageRate, TimeSpan period, long burstSize)
{
    BurstSize = burstSize;
    AverageRate = averageRate;
    Period = period;
}

I could review your implementation a bit more, but I have a more general concern. You seem to be concerned about multi-threading, yet you've exposed a synchronous API and are sleeping threads. I don't find it likely that this what you actually want, nor is manually dealing with threads very common since the introduction of the async/await pattern several versions ago. I would think deeply about whether you've actually solved your problem. You're forcing your entire application to stop and wait for TryThrottledWait() to run. This means nothing else can happen until it returns at an indeterminate time in the future. If there's a user at the other end, the app will appear frozen. If it's running headless on a server or in the background, then you're merely wasting resources that could be put to better use.

\$\endgroup\$
  • \$\begingroup\$ We could easily make an Async version of TryThrottledWait() that is using "await Task.Delay(ts);" instead of "Thread.Sleep(ts);" but all the logic of computing how long to wait for would be exactly the same and the class would still need to be threadsafe to correctly enforce the global limit. \$\endgroup\$ – skyde Mar 18 '17 at 23:39
1
\$\begingroup\$

I updated the code to solve some of the issue. For reference I am including the latest version below:

public class Throttler
{
    // Use this constant as average rate to disable throttling
    public const long NoLimit = -1;

    // Number of consumed tokens
    private long _consumedTokens;

    // timestamp of last refill time
    private long _lastRefillTime;

    // ticks per period
    private long _periodTicks;

    private double _averageRate;

    public long BurstSize
    {
        get;
        set;
    }

    public long AverageRate
    {
        get { return (long)_averageRate; }
        set { _averageRate = value; }
    }

    public TimeSpan Period
    {
        get
        {
            return new TimeSpan(_periodTicks);
        }
        set
        {
            _periodTicks = value.Ticks;
        }
    }

    public Throttler()
    {
        BurstSize = 1;
        AverageRate = NoLimit;
        Period = TimeSpan.FromSeconds(1);
    }

    /// <summary>
    /// Create a Throttler
    /// ex: To throttle to 1024 byte per seconds with burst of 200 byte use
    /// new Throttler(1024,TimeSpan.FromSeconds(1), 200);
    /// </summary>
    /// <param name="averageRate">The number of tokens to add to the bucket every interval. </param>
    /// <param name="period">Timespan of on interval.</param>
    /// <param name="burstSize"></param>
    public Throttler(long averageRate, TimeSpan period, long burstSize = 1)
    {
        BurstSize = burstSize;
        AverageRate = averageRate;
        Period = period;
    }

    public long TryThrottledWait(long amount)
    {
        if (BurstSize <= 0 || _averageRate <= 0)
        { // Instead of throwing exception, we just let all the traffic go
            return amount;
        }
        RefillToken();
        return ConsumeToken(amount);
    }

    // Return number of consummed token
    private long ConsumeToken(long amount)
    {
        while (true)
        {
            long currentLevel = Volatile.Read(ref _consumedTokens);
            long available = BurstSize - currentLevel;
            if (available == 0)
            {
                return 0;
            }
            long toConsume = amount;
            if (available < toConsume)
            {
                toConsume = available;
            }
            if (Interlocked.CompareExchange(ref _consumedTokens, currentLevel + toConsume, currentLevel) == currentLevel)
            {
                return toConsume;
            }
        }
    }

    /// <summary>
    /// Wait that works inside synchronous methods. 
    /// </summary>
    /// <param name="amount">number of tokens to remove</param>
    /// <returns>Returns once all Thread.Sleep have occurred</returns>
    public void ThrottledWait(long amount)
    {
        long remaining = amount;
        while (true)
        {
            remaining -= TryThrottledWait(remaining);
            if (remaining == 0)
            {
                break;
            }

            TimeSpan ts = GetSleepTime();
            Thread.Sleep(ts);
        }
    }

    /// <summary>
    /// Wait that works inside Async methods. 
    /// </summary>
    /// <param name="amount">number of tokens to remove</param>
    /// <returns>Returns once all Task.Delays have occurred</returns>
    public async Task ThrottledWaitAsync(long amount)
    {
        long remaining = amount;
        while (true)
        {
            remaining -= TryThrottledWait(remaining);
            if (remaining == 0)
            {
                break;
            }

            TimeSpan ts = GetSleepTime();
            await Task.Delay(ts).ConfigureAwait(false);
        }
    }

    /// <summary>
    /// Compute elapsed time using DateTime.UtcNow.Ticks and refil token using _periodTicks and _averageRate
    /// </summary>
    private void RefillToken()
    {
        long currentTimeTicks = DateTime.UtcNow.Ticks;
        // Last refill time in  ticks unit
        long refillTime = Volatile.Read(ref _lastRefillTime);
        // Time delta in ticks unit
        long TicksDelta = currentTimeTicks - refillTime;
        long newTokens = (long)(TicksDelta * _averageRate / _periodTicks);
        if (newTokens <= 0)
        {
            return;
        }
        long newRefillTime = refillTime == 0
                ? currentTimeTicks
                : refillTime + (long)(newTokens * _periodTicks / _averageRate);
        // Only try to refill newTokens If no other thread has beaten us to the update _lastRefillTime  
        if (Interlocked.CompareExchange(ref _lastRefillTime, newRefillTime, refillTime) != refillTime)
        {
            return;
        }
        // Loop until we succeed in refilling "newTokens" tokens
        // Its still possible for 2 thread to concurrently run the block below
        // This is why we need to make sure the refill is atomic
        while (true)
        {
            long currentLevel = Volatile.Read(ref _consumedTokens);
            long adjustedLevel = Math.Min(currentLevel, BurstSize); // In case burstSize decreased
            long newLevel = Math.Max(0, adjustedLevel - newTokens);
            if (Interlocked.CompareExchange(ref _consumedTokens, newLevel, currentLevel) == currentLevel)
            {
                return;
            }
        }
    }

    /// <summary>
    /// Get time to sleep until data can be sent again
    /// </summary>
    /// <returns>Timespan to wait</returns>
    private TimeSpan GetSleepTime()
    {
        long refillTime = Volatile.Read(ref _lastRefillTime);
        long nextRefillTime = (long)(refillTime + (_periodTicks / _averageRate));
        long currentTimeTicks = DateTime.UtcNow.Ticks;
        long sleepTicks = Math.Max(nextRefillTime - currentTimeTicks, 0);
        TimeSpan ts = new TimeSpan(sleepTicks);
        return ts;
    }
}
\$\endgroup\$

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.