C ++中的线程安全ExpiringDeque数据结构

Question

我正在尝试创建一个数据结构 ExpiringDeque。它应该有点类似于 std::deque。假设我只需要 push_back()、size() 和 pop_front()。数据结构需要每 T 秒自动过期最多 N 个第一个元素。

这个数据结构需要在内部管理自己的队列和过期线程。 如何以线程安全的方式编写它？这是我想出的一个例子，这看起来合理吗？我错过了什么？

#include <algorithm>
#include <atomic>
#include <cassert>
#include <deque>
#include <mutex>
#include <thread>
#include <unistd.h>
#include <iostream>

template <typename T>
class ExpiringDeque {
 public:
  ExpiringDeque(int n, int t) : numElements_(n), interval_(t), running_(true), items_({}) {
    expiringThread_ = std::thread{[&] () {
      using namespace std::chrono_literals;
      int waitCounter = 0;
      while (true) {
        if (!running_) {
          return;
        }
        
        std::this_thread::sleep_for(1s);
        if (waitCounter++ < interval_) {
          continue;
        }
        
        std::lock_guard<std::mutex> guard(mutex_);
        waitCounter = 0;
        int numToErase = std::min(numElements_, static_cast<int>(items_.size()));
        std::cout << "Erasing " << numToErase << " elements\n";
        items_.erase(items_.begin(), items_.begin() + numToErase);
      }
    }};
  }

  ~ExpiringDeque() {
     running_ = false;
     expiringThread_.join();
   }

  T pop_front() {
    if (items_.size() == 0) {
      throw std::out_of_range("Empty deque");
    }
    std::lock_guard<std::mutex> guard(mutex_);
    T item = items_.front();
    items_.pop_front();
    return item;
  }
  
  int size() {
    std::lock_guard<std::mutex> guard(mutex_);
    return items_.size();
  }

  void push_back(T item) {
    std::lock_guard<std::mutex> guard(mutex_);
    items_.push_back(item);
  }

 private:
  int numElements_;
  int interval_;
  
  std::atomic<bool> running_;
  std::thread expiringThread_;

  std::mutex mutex_;
  std::deque<T> items_;
};

int main() {
    ExpiringDeque<int> ed(10, 3);
    ed.push_back(1);
    ed.push_back(2);
    ed.push_back(3);
    
    assert(ed.size() == 3);
    assert(ed.pop_front() == 1);
    assert(ed.size() == 2);
    
    // wait for expiration
    sleep(5);
    
    assert(ed.size() == 0);
    ed.push_back(10);
    assert(ed.size() == 1);
    assert(ed.pop_front() == 10);
    
    return 0;
}

Answer 1

您可以通过使用条件变量避免ExpiringDeque 的析构函数中不必要的等待。我还会使用带有谓词的std::condition_variable::wait_for 来检查running_ 标志。这将确保您等待超时或通知，以较早者为准。您避免以这种方式使用waitCounter 和continue。

您应该做的另一件事是在检查pop_front() 中的双端队列大小之前锁定互斥锁，否则它不是线程安全的。

这是您的代码的更新版本：

template <typename T>
class ExpiringDeque {
public:
    ExpiringDeque(int n, int t) : numElements_(n), interval_(t), running_(true), items_({}), cv_() {
        expiringThread_ = std::thread{ [&]() {
          using namespace std::chrono_literals;

          while (true) {
            //Wait for timeout or notification
            std::unique_lock<std::mutex> lk(mutex_);
            cv_.wait_for(lk, interval_ * 1s, [&] { return !running_; });

            if (!running_)
                return;

            //Mutex is locked already - no need to lock again
            int numToErase = std::min(numElements_, static_cast<int>(items_.size()));
            std::cout << "Erasing " << numToErase << " elements\n";
            items_.erase(items_.begin(), items_.begin() + numToErase);
          }
        } };
    }

    ~ExpiringDeque() {
        //Set flag and notify worker thread
        {
            std::lock_guard<std::mutex> lk(mutex_);
            running_ = false;
        }
        cv_.notify_one();
        expiringThread_.join();
    }

    T pop_front() {
        std::lock_guard<std::mutex> guard(mutex_);
        if (items_.size() == 0) {
            throw std::out_of_range("Empty deque");
        }
        T item = items_.front();
        items_.pop_front();
        return item;
    }

    ...

private:
    int numElements_;
    int interval_;

    bool running_;
    std::thread expiringThread_;

    std::mutex mutex_;
    std::deque<T> items_;
    std::condition_variable cv_;
};

您可以将running_ 标志设为普通bool，因为std::condition_variable::wait_for 会自动检查超时或通知。