具有并发性的漏桶算法

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

在显示的代码中有多个基本错误。

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

leak()将等待桶的填充率至少为0，然后从桶中减去泄漏率。然后它就会完成。就这样。不会再这样了。泄漏的螺纹将不复存在。它会变成一根线。它将永远怀念峡湾。一旦水桶漏了一次，它的漏水孔就被堵住了，它就变成了一个完全防漏的水桶。

 new LeakyBucket(30, 20);

水桶的容量为30桶，漏水率为20桶。

lb->add(rand()%40);

它被打了十次电话，加在0到39滴水之间的任何地方。

那么，假设我们第一次把20滴水倒进水桶里。漏水线会苏醒过来，把那20滴水拿出来，赚到它应得的退休金。

但是等等，我们还有九种水要补充！

第二个打给add()的电话将下降25滴水。第三次尝试增加30滴水。水桶现在容量过大了。对add()的第四个调用现在将永远阻塞，因为正如我们刚才看到的，这个桶现在是完全防泄漏的。

这是第一个错误:桶漏一次，然后就不会再漏了。

            _cond.wait(lk, [this]() {
                return filled > 0;
            });

            filled -= leakRate;

水桶里的漏水会等到水桶里至少有一滴水，然后漏掉20滴水。所以，如果五滴水已经在水桶里了，水桶，在所有这些之后，会有负十五滴水。这显然是没有意义的，因此这将是第二个需要修复的错误，然后才能正常工作。

这里可能还有第三只虫子。桶被定义为具有一定的容量。然而，在我的一个例子中，在上面，水桶最终有更多的水滴超过其规定的容量。这也不算什么。