具有并發性的漏桶演算法

試圖模擬一個場景，其中多個執行緒正在創建流量以填充存盤桶和一個執行緒以指定的速率泄漏存盤桶。但是，代碼正在陷入僵局。你能看看這段代碼嗎？如果您發現任何錯誤和我應該添加的最佳修改，請告訴我。

代碼

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

開/關：

 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

uj5u.com熱心網友回復：

顯示的代碼中有多個基本錯誤。

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;

桶中的漏水將等到桶中至少有 1 滴水，然后再漏 20 滴水。因此，如果桶中已經有五滴水，那么桶中的水將是負的十五滴。這顯然是沒有意義的，所以這將是需要修復的第二個錯誤，然后才能正常作業。

這里可能還有第三個錯誤。桶被定義為具有一定的容量。然而，在我上面的一個例子中，水桶的水滴最終超過了其規定的容量。這也沒有加起來。

標籤：

上一篇：當在守護執行緒中運行的任何模塊發生例外時如何銷毀tkintermainloop

下一篇：web::block()因“`NonNull<pq_sys::pg_conn>`無法在執行緒之間安全共享”而失敗