c++的queue在多线程下崩溃原因分析

杨永贞

发布于 2022-06-30 15:46:41

1.1K0

发布于 2022-06-30 15:46:41

这是个难找的bug，c++的bug真是防不胜防。若不是单点调试，在生产环境中可真不好找。以下是我排查此bug的一个过程记录，留作备忘，在以后的使用过程中要小心避坑。

问题产生

我们知道c++的queue和map等数据结构是线程并发不安全的，为此我们常封装实现了线程安全的priority_queue，姑且叫做 thread_safe::priority_queue。（关于c++并发编程这块儿推荐经典书籍《C++并发编程实战》）。本以为封装后就可以放心在多线程中使用了，结果崩溃了，且还是偶发的。

先看以下示例：

#include <iostream>
#include <queue>
using namespace std;

int main()
{
    cout<<"Hello World"<<endl;

    priority_queue<int> pqueue;
     //pushing value in pqueue.
    cout<<pqueue.top();
    cout<<"Hello World End"<<endl;
    return 0;
}

再看以下示例：

#include <iostream>
#include <queue>
using namespace std;

int main()
{
   std::cout<<"Hello World\n";
    pqueue.push(10);
    std::cout<<pqueue.top();
    pqueue.pop();
    pqueue.pop();
    if(pqueue.empty()){
        std::cout<<"\n pqueue is empty\n";
    }else{
        std::cout<<"\n pqueue is not empty\n";
    }
    std::cout<<pqueue.top();
    std::cout<<"Hello World End\n";
    return 0;
}

提出几个问题。这两个示例分别会输出什么？

做下测试会发现，第一个示例直接就崩了，第二个会输出 pqueue is not empty，有点儿跟想象的不一样。可能你回说这样的测试无意义吧，正常使用中，连基本的queue是否是empty都不判断吗？

这也是本次bug的导火索。请看在多线程中的示例：

thread_safe::priority_queue<int> priorityQueue_;

void task_A() {
    LOGGING_DEBUG(" task_A start...");
    while (true) {
        // 1
        if (priorityQueue_.empty()) {
            LOGGING_WARN("priorityQueue is empty.");
            return;
        }
        // 2
        auto taskPtr = priorityQueue_.top();
        if (taskPtr) {

        } else {
            LOGGING_ERROR("taskPtr is nullptr.");
            priorityQueue_.pop();
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    }
}

在多线程环境下，即便有priorityQueue_.empty()的判断，但是也已经失去了意义。因为并发的情况下，执行到2时，能保证priorityQueue_非空？可能它已经是empty了。

以下是本次测试thread_safe_queue的实现：

namespace thread_safe
{
template <class T, class Container = std::deque<T>>
class queue
{
public:
  explicit queue(const Container &ctnr = Container()) : storage(ctnr) {}
  bool empty() const
  {
    std::lock_guard<std::mutex> lock(mutex);
    return storage.empty();
  }

  size_t size() const
  {
    std::lock_guard<std::mutex> lock(mutex);
    return storage.size();
  }

  T &back()
  {
    std::lock_guard<std::mutex> lock(mutex);
    return storage.back();
  }
  const T &back() const
  {
    std::lock_guard<std::mutex> lock(mutex);
    return storage.back();
  }

  T &front()
  {
    std::lock_guard<std::mutex> lock(mutex);
    return storage.front();
  }
  const T &front() const
  {
    std::lock_guard<std::mutex> lock(mutex);
    return storage.front();
  }

  void push(const T &u)
  {
    std::lock_guard<std::mutex> lock(mutex);
    storage.push(u);
  }

  void pop()
  {
    std::lock_guard<std::mutex> lock(mutex);
    storage.pop();
  }

private:
  std::queue<T, Container> storage;
  mutable std::mutex mutex;
};

template <class T, class Container = std::vector<T>, class Compare = std::less<typename Container::value_type>>
class priority_queue
{
public:
  explicit priority_queue(const Compare &x = Compare(), const Container &y = Container()) : storage(x, y) {}
  template <class InputIterator>
  priority_queue(InputIterator first, InputIterator last, const Compare &x = Compare(),
                 const Container &y = Container())
      : storage(first, last, x, y)
  {
  }

  bool empty() const
  {
    std::lock_guard<std::mutex> lock(mutex);
    return storage.empty();
  }

  size_t size() const
  {
    std::lock_guard<std::mutex> lock(mutex);
    return storage.size();
  }

  //  T &top(void)
  //  {
  //    std::lock_guard<std::mutex> lock(mutex);
  //    return storage.top();
  //  }
  const T &top() const
  {
    std::lock_guard<std::mutex> lock(mutex);
    return storage.top();
  }

  void push(const T &u)
  {
    std::lock_guard<std::mutex> lock(mutex);
    storage.push(u);
  }

  void pop()
  {
    std::lock_guard<std::mutex> lock(mutex);
    if (!storage.empty())
      storage.pop();
  }

private:
  std::priority_queue<T, Container, Compare> storage;
  mutable std::mutex mutex;
};

}  // namespace thread_safe

可以基于此封装在多线程中测试验证下。