我应该如何改进线程池以使其更加线程安全？

Question

我目前正在学习有关线程池的基础知识。以下是我根据网络上的一些示例编写的一些代码块：

SyncQueue.h

#ifndef SYNC_QUEUE_H
#define SYNC_QUEUE_H

#include <list>
#include <mutex>
#include <iostream>

template<typename T>
class SyncQueue {
public:
  SyncQueue();
  ~SyncQueue();
  SyncQueue(const SyncQueue&) = delete;
  SyncQueue& operator=(const SyncQueue &) = delete;
  void append(const T& data);
  T& get();
  unsigned long size();
  bool empty();
private:
  std::list<T> queue;
  std::mutex myMutex;
};
#endif

SyncQueue.cpp

#include "SyncQueue.h"

template<typename T>
SyncQueue<T>::SyncQueue():
  queue(),
  myMutex() {}

template<typename T>
SyncQueue<T>::~SyncQueue() {}

template<typename T>
void SyncQueue<T>::append(const T& data) {
  std::unique_lock<std::mutex> l(myMutex);
  queue.push_back(data);
}

template<typename T>
T& SyncQueue<T>::get() {
  std::unique_lock<std::mutex> l(myMutex);
  T& res = queue.front();
  queue.pop_front();
  return res;
}

template<typename T>
unsigned long SyncQueue<T>::size() {
  std::unique_lock<std::mutex> l(myMutex);
  return queue.size();
}

template<typename T>
bool SyncQueue<T>::empty() {
  std::unique_lock<std::mutex> l(myMutex);
  return queue.empty();
}

template class SyncQueue<std::function<void()>>;

ThreadPool.h

#ifndef THREAD_POOL_H
#define THREAD_POOL_H

#include <atomic>
#include <functional>
#include <mutex>
#include <thread>
#include <vector>
#include "SyncQueue.h"

class ThreadPool {
public:
  ThreadPool(unsigned long thrdAmount = 0);
  virtual ~ThreadPool();
  void appendTask(std::function<void()> func);
  unsigned long pendingTasks();
private:
  void runThread();
  unsigned int myThrdAmount;
  std::atomic<bool> done;
  SyncQueue<std::function<void()>> syncQueue;
  std::vector<std::thread> threads;
  std::condition_variable myCondVar;
  std::mutex myMutex;
};

#endif

ThreadPool.cpp

#include "ThreadPool.h"

ThreadPool::ThreadPool(unsigned long thrdAmount):
  myThrdAmount(0),
  done(false),
  syncQueue(),
  threads(),
  myCondVar(),
  myMutex() {
  if (thrdAmount > 0) {
    myThrdAmount = thrdAmount;
  } else {
    myThrdAmount = std::thread::hardware_concurrency();
  }
  for (unsigned int i = 0; i < myThrdAmount; i++) {
    threads.push_back(std::thread(&ThreadPool::runThread, this));
  }
}

ThreadPool::~ThreadPool() {
  done = true;
  myCondVar.notify_all();
  for (auto& thrd: threads) {
    if (thrd.joinable()) {
      thrd.join();
    }
  }
}

void ThreadPool::appendTask(std::function<void()> func) {
  syncQueue.append(func);
  {
    std::unique_lock<std::mutex> l(myMutex);
    myCondVar.notify_one();
  }
}

unsigned long ThreadPool::pendingTasks() {
  return syncQueue.size();
}

void ThreadPool::runThread() {
  while (!done) {
    if (syncQueue.empty()) {
      std::unique_lock<std::mutex> l(myMutex);
      myCondVar.wait(l);
      continue;
    }
    syncQueue.get()();
  }
}

main.cpp

#include <unistd.h>
#include <iostream>
#include "ThreadPool.h"

void print() {
  std::cout << "Hello World!" << std::endl;
}

int main(int argc, char const *argv[]) {
  ThreadPool p;
  for (int i = 0; i < 20; i++) {
    p.appendTask(print);
  }
  std::cout << "Pending: " << p.pendingTasks() << std::endl;
  sleep(5);
  for (int i = 0; i < 20; i++) {
    p.appendTask(print);
  }
  return 0;
}

尽管 SyncQueue 上的所有操作都被互斥锁锁定并且 ThreadPool 的条件变量也受到互斥锁的保护，但代码通常会导致未定义的行为。

也就是说，你能解释一下代码哪里缺乏线程安全吗？我应该如何改进它？

Answer 1

 void ThreadPool::appendTask(std::function<void()> func) {
  syncQueue.append(func);
  {
    std::unique_lock<std::mutex> l(myMutex);
    myCondVar.notify_one();
  }
}

void ThreadPool::runThread() {
  while (!done) {
    if (syncQueue.empty()) {
      std::unique_lock<std::mutex> l(myMutex);
      myCondVar.wait(l);
      continue;
    }
    syncQueue.get()();
  }
}

问题是 myMutex 实际上并没有保护任何东西。所以你的代码在等待队列时有一个灾难性的竞争条件。

考虑：

线程调用 runThread 发现 syncQueue 为空。
线程调用 appendTask 将作业添加到队列并调用 notify_one。没有要通知的线程。
线程调用 runThread 最终获得 myMutex 上的锁并等待条件变量，但队列不为空。

将您用于等待的条件变量与保护您正在等待的谓词的互斥体相关联是绝对重要的。条件变量的全部目的是允许您自动解锁谓词并等待没有竞争条件的信号。但是您将谓词隐藏在 syncQueue 中，破坏了条件变量的锁处理逻辑。

您可以通过在 myMutex 互斥锁的保护下对 syncQueue 进行所有调用来解决此竞争条件。但是让 syncQueue 可等待可能更有意义。不过，这可能会使关闭线程池变得更加困难。

我应该如何改进线程池以使其更加线程安全？

How should I improve a thread pool to make it more thread safe?

c++

queue

thread-safety

threadpool