Lock-free implementation of the token bucket algorithm in C++11.
This project was created by Erik Rigtorp <[email protected]>.
Lock-free implementation of the token bucket algorithm in C++
License: MIT License
Lock-free implementation of the token bucket algorithm in C++11.
This project was created by Erik Rigtorp <[email protected]>.
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.
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();
}
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;
}
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
?
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
}
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
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.