3
\$\begingroup\$

Trying to mimic a scenario where multiple threads are creating the traffic to fill the buckets & a thread which leaks the bucket at a specified rate. Could you review this code?

#include <iostream>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <atomic>
#include <chrono>

using namespace std;

class LeakyBucket {
public:
    LeakyBucket(int size, int rate) : maxCapacity(size), leakRate(rate), filled(0)  {}
    void add(int newDataSize) {
        unique_lock<mutex> lk(_mtx);
        _cond.wait(lk, [this](){
           return  filled<=maxCapacity;
        });


        filled = (filled+newDataSize) > maxCapacity ? maxCapacity:(filled+newDataSize);
        cout<<"\n Filled bucket with : "<<newDataSize;
        cout<<"\n Filled: "<<filled<<"\n ----------";
        _cond.notify_one();
    }

    void leak() {
        while(1) {
            {
                unique_lock<mutex> lk(_mtx);
            _cond.wait(lk, [this]() {
                return filled > 0 || _done;
            });
            if(_done)
                break;

            filled = (filled-leakRate<0) ? 0 : (filled-leakRate);
            cout << "\n Leaked bucket with leakRate";
            cout << "\n BucketFilledRemain: " << filled << "\n ----------";
            _cond.notify_one();
            }
            _sleep:
            this_thread::sleep_for(chrono::seconds(1));
        }
    }

    bool _done = false;
private:
    atomic<int> filled;
    int maxCapacity;
    int leakRate; // Per second
    mutex _mtx;
    condition_variable _cond;

};

void runLeakyBucketAlgorithm() {
    LeakyBucket *lb = new LeakyBucket(30, 20);

    thread t1(&LeakyBucket::leak, lb);
    thread t2([&](){
       for(int i=0; i<10; i++) {
           cout<<"\n launching thread: "<<i;
           lb->add(rand()%40);
       }
       this_thread::sleep_for(chrono::seconds(5));
       lb->_done = true;
    });
    if(t2.joinable()) {
       t2.join();
    }

    t1.join();
}

O/P

 launching thread: 0
 Filled bucket with : 7
 Filled: 7
 ----------
 launching thread: 1
 Filled bucket with : 9
 Filled: 16
 ----------
 launching thread: 2
 Leaked bucket with leakRate
 BucketFilledRemain: 0
 ----------
 Filled bucket with : 33
 Filled: 30
 ----------
 launching thread: 3
 Filled bucket with : 18
 Filled: 30
 ----------
 launching thread: 4
 Filled bucket with : 10
 Filled: 30
 ----------
 launching thread: 5
 Filled bucket with : 32
 Filled: 30
 ----------
 launching thread: 6
 Filled bucket with : 24
 Filled: 30
 ----------
 launching thread: 7
 Filled bucket with : 38
 Filled: 30
 ----------
 launching thread: 8
 Filled bucket with : 3
 Filled: 30
 ----------
 launching thread: 9
 Filled bucket with : 29
 Filled: 30
 ----------
 Leaked bucket with leakRate
 BucketFilledRemain: 10
 ----------
 Leaked bucket with leakRate
 BucketFilledRemain: 0
\$\endgroup\$

1 Answer 1

3
\$\begingroup\$

Avoid leading underscores in names

Some uses of underscores are reserved by the standard, I recommend just avoiding all leading underscores in names.

Also, why do only some member variables have a leading underscore? If you are going to use them, be consistent.

Add a newline at the end of each line

When printing, add the newline at the end of each line instead of at the start. This ensures whatever comes next (which might be an error message, or the shell's prompt if your program ends) starts on its own line.

Make use of std::min()

Instead of using a ternary expression, use std::min() if you want to limit some value. For example:

filled = std::min(filled + newDataSize, maxCapacity);

This avoids having to repeat yourself.

Avoid needing signed integers

The current fill level, the maximum capacity and the leak reate should all be non-negative numbers. Consider using unsigned ints for them. You should also then avoid subtracting values if the result could be negative. Instead, find a way to do everything with non-negative values, like so in leak():

filled -= std::min(leakRate, filled);

Unnecessary use of new

There is no reason to use new to create an instance of a LeakyBucket inside runLeakyBucketAlgorithm(). You can just declare it on the stack:

void runLeakyBucketAlgorithm() {
    LeakyBucket lb(30, 20);

    thread t1(&LeakyBucket::leak, &lb);
    ...
};

Redesigning the leaky part

Adding things to the bucket is done nicely. However, the leaking part is not very useful at the moment, and it also requires the user to create a thread themselves. I would redesign this in one of two ways:

Add a callback function that is called when leaking

It would be nice if the user of a LeakyBucket didn't have to know anything about threads, but still be able to tell the bucket what to do when it leaks. You can move thread management to the constructor and destructor of LeakyBucket, and add a callback function as a parameter to the constructor. This way, you could use it like so:

LeakyBucket lb(30, 20, [](unsigned int leaked) {
    std::cout << "Bucket leaked " << leaked << " units\n";
});

Let leak() return how much has leaked

Instead of letting leak() be an infinite loop that has to run in its own thread, consider making leak() similar to add(), but instead of telling it how much to add, have it return how much has leaked since the last call to leak(). This way, the caller can decide how long to sleep, if at all:

LeakyBucket lb(30, 20);

std::thread t1([&]() {
    for (int i = 0; i < 60; i++) {
         std::cout << "Bucket leaked " << lb.leak() << " units\n";
         std::this_thread::sleep_for(std::chrono::seconds(1));
    }
});

Accurate timing

std::this_thread::sleep_for() does not guarantee it will sleep exactly for the duration you specified. And even if it did, all the other operations you are doing inside the loop in leak() also take some time. So in effect, you are leaking a bit slower than you specified. It should be possible to get the current time and compare it to the time from the last iteration of the loop, and this way more accurately account for the passage of time. For sure you need to do something like this if you want to implement the version of leak() that returns how much has leaked since the last call.

\$\endgroup\$
4
  • \$\begingroup\$ Hello G. Sliepen, Thank you for the detailed answer, appreciate it. Could you please elaborate more on Add a callback function part? I am not able to get it completely \$\endgroup\$ Jan 24, 2022 at 18:31
  • \$\begingroup\$ @user8707488 What is it that is not clear exactly? \$\endgroup\$
    – G. Sliepen
    Jan 24, 2022 at 19:58
  • \$\begingroup\$ I didn't get how can I redesign the leaky part. I know you have already explained in detail, but trying to understand how can I implement Add a callback function that is called when leaking. \$\endgroup\$ Jan 24, 2022 at 21:43
  • \$\begingroup\$ @user8707488 Ah. You can use a variable of type std::function<void(unisgned int)> to hold any function that takes an unsigned int as parameter. Add a parameter in the constructor of LeakyBucket of that type and store it as a member variable: class LeakyBucket { public: LeakyBucket(..., std::function<void(unsigned int)> callback): ..., callback(callback) {} ... private: std::function<void(unsigned int)> callback; } Then inside the leaking thread, call callback(std::min(leaked, fillrate)). \$\endgroup\$
    – G. Sliepen
    Jan 24, 2022 at 21:53

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

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