rigtorp / tokenbucket

Lock-free implementation of the token bucket algorithm in C++

TokenBucket.h

Lock-free implementation of the token bucket algorithm in C++11.

About

This project was created by Erik Rigtorp <[email protected]>.

tokenbucket's Issues

Not working as expected ? allowing more tokens .

Hi ,

I am trying your implementation keeping the tokens per sec to 500 . the number of tokens allowed seems to be more than 500 around 512ish ...
Can you please help me if me going wrong somwher in understanding.

also it does work as expected with smaller token size of say 40 or 100

#include
#include
#include
#include
#include

class TokenBucket {
public:
TokenBucket() {}

TokenBucket(const uint64_t rate, const uint64_t burstSize) {
    timePerToken_ = 1000000 / rate;
    timePerBurst_ = burstSize * timePerToken_;
}

TokenBucket(const TokenBucket &other) {
    timePerToken_ = other.timePerToken_.load();
    timePerBurst_ = other.timePerBurst_.load();
}

TokenBucket &operator=(const TokenBucket &other) {
    timePerToken_ = other.timePerToken_.load();
    timePerBurst_ = other.timePerBurst_.load();
    return *this;
}

bool consume(const uint64_t tokens)
{
    const uint64_t now =
            std::chrono::duration_cast<std::chrono::microseconds>(
                    std::chrono::steady_clock::now().time_since_epoch())
                    .count();
    const uint64_t timeNeeded =
            tokens * timePerToken_.load(std::memory_order_relaxed);
    const uint64_t minTime =
            now - timePerBurst_.load(std::memory_order_relaxed);
    uint64_t oldTime = time_.load(std::memory_order_relaxed);
    uint64_t newTime = oldTime;

    if (minTime > oldTime)
    {
        newTime = minTime;
    }

    for (;;) {
        newTime += timeNeeded;
        if (newTime > now) {
            return false;
        }
        if (time_.compare_exchange_weak(oldTime, newTime,
                                        std::memory_order_relaxed,
                                        std::memory_order_relaxed)) {
            return true;
        }
        newTime = oldTime;
    }

    return false;
}

private:
std::atomic<uint64_t> time_ = {0};
std::atomic<uint64_t> timePerToken_ = {0};
std::atomic<uint64_t> timePerBurst_ = {0};
};

int main()
{

uint64_t rate = 500;
uint64_t bs = 500;
TokenBucket t(rate, bs);


uint64_t i{0};
while(true)
{
    if (t.consume(1))
    {
        std::cout << ++i << std::endl;
    }
    else
    {
        i = 0;
        std::cout << "cant consume "<< std::endl;
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}
return 0;

}

Output :
1
2
......
506
507
508
509
510
511
512
513
514
515
516
517
518
cant consume

Thanks in advance.

Unnecessary `atomic`s for member variables

As it stands, there's no need for timePerToken_ and timePerBurst_ to be declared as std::atomic<uint64_t> because there are no concurrent modifications to either variable. The consume function modifies time_, but only reads from the other two aforementioned variables. Can they become simply uint64_t?

newTime updated multiple times?

Not sure if it's an issue or not, but definitely a question... :)

If the 'compare_exchange_weak' call in the code below fails, the newTime would be incremented by timeNeeded multiple times. Is that the intent?

    for (;;) {
      newTime += timeNeeded;
      if (newTime > now) {
        return false;
      }
      if (time_.compare_exchange_weak(oldTime, newTime,
                                      std::memory_order_relaxed,
                                      std::memory_order_relaxed)) {
        return true;
      }
      newTime = oldTime;
    }

copy constructor problem

The copy constructor doesn't initialize time_. This causes a problem if the memory of the new object happens to contain a time in the future. I added the line below, and my problems went away.

TokenBucket(const TokenBucket &other) {
time_ = other.time_.load(); // added this line
timePerToken_ = other.timePerToken_.load();
timePerBurst_ = other.timePerBurst_.load();
}

example

Hi,

I try to use TokenBucket to limit the output rate of an UDP send... and I fail
Could you give us an example of how to use that class?
If I want to limit at X Mbps an UDP flow @ MTU octets per packet, how can I do?

thanks

Not considering minTime < oldTime condition if compare_exchange_weak fails

Before entering the loop the condition is handled for the case when the previous run (oldTime) is older than the earliest possible bucked (minTime) of the buffer window.

However, for the case that compare_exchange_weak fails and newTime is reset to oldTime back again and the loop is re-run, setting it to old time is wrong as it does ignore that case?

    if (minTime > oldTime) {
      newTime = minTime; // <--- newTime changed
    }

    for (;;) {
      newTime += timeNeeded;
      if (newTime > now) {
        return false;
      }
      if (time_.compare_exchange_weak(oldTime, newTime,
                                      std::memory_order_relaxed,
                                      std::memory_order_relaxed)) {
        return true;
      }
      newTime = oldTime; // <-- newTime reset to oldTime, not considering the eventually required shift to minTime
    }