具有并发性的漏桶算法

Question

试图模拟多个线程正在创建流量以填充桶的场景&一个以指定速率泄漏桶的线程。你能检查一下这段代码吗？

#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

Answer 1

避免在名称

中引入下划线

下划线的某些用法由标准保留。，我建议您避免所有名字中的前导下划线。

另外，为什么只有一些成员变量有一个前导下划线？如果你要使用它们，要保持一致。

在每一行的末尾添加一个换行符，

打印时，将换行符添加到每一行的末尾而不是开始处。这可以确保接下来发生的一切(可能是错误消息，或者程序结束时shell的提示)从自己的行开始。

利用std::min()

如果要限制某些值，则不要使用三元表达式，而是使用std::min()。例如：

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

这就避免了重复自己的事情。

避免需要有符号整数

当前的填充级别、最大容量和泄漏容量都应该是非负数.考虑使用unsigned ints为他们。如果结果可能为负值，则还应避免减去值。相反，要找到一种以非负值来完成所有事情的方法，就像在leak()中那样：

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

不必要地使用new

没有理由使用new在runLeakyBucketAlgorithm()中创建LeakyBucket的实例。您可以在堆栈上声明它：

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

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

对漏件

的重新设计

把事情添加到桶里做得很好。然而，泄漏部分目前并不是很有用，它还要求用户自己创建一个线程。我会用两种方式重新设计：

添加了一个在泄漏

时调用的回调函数

如果LeakyBucket的用户不必知道任何关于线程的信息，但仍然能够告诉桶泄漏时应该做什么，那就太好了。可以将线程管理移动到LeakyBucket的构造函数和析构函数，并将回调函数作为参数添加到构造函数中。这样，你就可以像这样使用它：

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

让leak() return知道

泄露了多少

与其让leak()成为一个必须在自己的线程中运行的无限循环，不如考虑使leak()类似于add()，而不是告诉它要添加多少，让它return知道自上次调用leak()以来泄漏了多少。这样，来电者就可以决定睡多长时间，如果有的话：

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));
    }
});

精确定时

std::this_thread::sleep_for()并不保证它会在指定的持续时间内完全休眠。即使是这样，您在leak()循环中执行的所有其他操作也需要一些时间。因此，实际上，您的泄漏速度比指定的要慢一些。应该可以获得当前时间并将其与循环的最后一次迭代之间的时间进行比较，这样可以更准确地说明时间的推移。当然，如果要实现返回自上次调用以来泄漏量的leak()版本，则需要执行类似的操作。