C++ Thread Pool

Thread Pool

쓰레드 풀은 쓰레드의 작업 대기소 개념이다. 여러 Thread가 대기를 하고 있고 작업을 넣어주면 알아서 스레드중 하나가 받아서 처리하고 다시 대기상태로 돌아오는 것이다.

• 이는 쉽게 queue를 이용하는 것이고 producer-consumer 이랑 기본 아이디어는 동일하다고 볼 수 있다.

설계 1: 컨테이너 만들기

• Thread Pool을 만들기 위해서는 Worker(작업하는 스레드들)들을 보관할 컨테이너가 필요하다.

// 총 Worker 쓰레드의 개수.
size_t num_threads_;
// Worker 쓰레드를 보관하는 벡터.
std::vector<std::thread> worker_threads_;

• size_t: 메모리, 객체등의 크기를 표현하기 위한 정수 타입. sizeof의 결과 타입이다. 크기가 가질 수 있는 최대 객체 크기만큼 표현이 가능하다.

// 할일들을 보관하는 job 큐.
std::queue<std::function<void()>> jobs_;

• 전체 thread 컨테이너를 만들었으면 이제 작업들을 넣을 컨테이너도 만든다.
• 일단은 전체 흐름을 이해하기 위해 작업은 void를 반환하고 인자가 없는걸로 고정하고 진행하겠다. (추후에 template을 사용해 다 대응해야함.)

std::condition_variable cv_job_q_;
std::mutex m_job_q_;

• 위에서 작업 컨테이너를 queue를 이용했는데, 큐는 멀티 스레드 환경에서 안전하지 않기 때문에 이를 보호할 mutex와 cv를 선언한다.

// 모든 쓰레드 종료
bool stop_all;

• 그리고 모든 thread를 종료할 수 있는 bool 타입을 하나 만든다.

설계 2: Thread Pool 만들기

ThreadPool::ThreadPool(size_t num_threads)
    : num_threads_(num_threads), stop_all(false) {
  worker_threads_.reserve(num_threads_);
  for (size_t i = 0; i < num_threads_; ++i) {
    worker_threads_.emplace_back([this]() { this->WorkerThread(); });
  }
}

• .reserve(): 미리 메모리 공간을 미리 만들어 두는 것이다. -> 이걸 안하면 emplace_back을 하면서 기존 메모리 공간을 넘어서면 다시 다 해제하고 새로운 공간을 할당하는 식으로 성능을 잡아먹게 된다. +) 안정성도 좀 떨어짐.
• 입력받은 만큼 WorkerThread()를 실행하게 한다.

설계 3: 스레드가 작업 가져가게 하기

void ThreadPool::WorkerThread() {
  while (true) {
    std::unique_lock<std::mutex> lock(m_job_q_);
    cv_job_q_.wait(lock, [this]() { return !this->jobs_.empty() || stop_all; });
    if (stop_all && this->jobs_.empty()) {
      return;
    }

    // 맨 앞의 job 을 뺀다.
    std::function<void()> job = std::move(jobs_.front());
    jobs_.pop();
    lock.unlock();

    // 해당 job 을 수행한다 :)
    job();
  }
}

• 그렇다면 WorkerThread()에서는 무슨 일을 수행해야 할까?
• 일단 작업이 있는지 확인하고 있다면 수행하면 될 것이다.
• 락을 걸고 푼다. (wait 상태) -> 누군가 호출하면 lock을 건다는 뜻.
• jobs_가 비워져 있지 않거나 stop_all이 true가 될때까지 대기한다.
• stop_all이 true고 작업이 없으면 종료한다. (작업 중지)
• move()를 통해 jobs_의 맨 앞 작업을 꺼낸다.
• 락을 풀고 작업을 수행한다.

설계 4: 작업 추가하기

void ThreadPool::EnqueueJob(std::function<void()> job) {
  if (stop_all) {
    throw std::runtime_error("ThreadPool 사용 중지됨");
  }
  {
    std::lock_guard<std::mutex> lock(m_job_q_);
    jobs_.push(std::move(job));
  }
  cv_job_q_.notify_one();
}

• 작업은 쉽게 jobs_에 밀어넣으면 된다.
• 이때 stop_all이라면 작업을 넣지 않는다.
• lock_guard<std::mutex>를 이용해 중간 블록에서만 락을 운용한다.
• m_job_q_ 락을 거는 이유는 이미 작업 처리중일 수 있기 때문이다.
• push가 성공적으로 끝나면 notify_one()을 통해 wait중인 스레드중 하나를 깨워 동작시킨다.

설계 5: 소멸자

ThreadPool::~ThreadPool() {
  stop_all = true;
  cv_job_q_.notify_all();

  for (auto& t : worker_threads_) {
    t.join();
  }
}

• 마지막으로 소멸자는 워커 스레드 전체를 깨우게 해서 return시키고 join()을 통해 안전하게 대기한다.

전체 Thread Pool 코드

#include <chrono>
#include <condition_variable>
#include <cstdio>
#include <functional>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>

namespace ThreadPool {
class ThreadPool {
 public:
  ThreadPool(size_t num_threads);
  ~ThreadPool();

  // job 을 추가한다.
  void EnqueueJob(std::function<void()> job);

 private:
  // 총 Worker 쓰레드의 개수.
  size_t num_threads_;
  // Worker 쓰레드를 보관하는 벡터.
  std::vector<std::thread> worker_threads_;
  // 할일들을 보관하는 job 큐.
  std::queue<std::function<void()>> jobs_;
  // 위의 job 큐를 위한 cv 와 m.
  std::condition_variable cv_job_q_;
  std::mutex m_job_q_;

  // 모든 쓰레드 종료
  bool stop_all;

  // Worker 쓰레드
  void WorkerThread();
};

ThreadPool::ThreadPool(size_t num_threads)
    : num_threads_(num_threads), stop_all(false) {
  worker_threads_.reserve(num_threads_);
  for (size_t i = 0; i < num_threads_; ++i) {
    worker_threads_.emplace_back([this]() { this->WorkerThread(); });
  }
}

void ThreadPool::WorkerThread() {
  while (true) {
    std::unique_lock<std::mutex> lock(m_job_q_);
    cv_job_q_.wait(lock, [this]() { return !this->jobs_.empty() || stop_all; });
    if (stop_all && this->jobs_.empty()) {
      return;
    }

    // 맨 앞의 job 을 뺀다.
    std::function<void()> job = std::move(jobs_.front());
    jobs_.pop();
    lock.unlock();

    // 해당 job 을 수행한다 :)
    job();
  }
}

ThreadPool::~ThreadPool() {
  stop_all = true;
  cv_job_q_.notify_all();

  for (auto& t : worker_threads_) {
    t.join();
  }
}

void ThreadPool::EnqueueJob(std::function<void()> job) {
  if (stop_all) {
    throw std::runtime_error("ThreadPool 사용 중지됨");
  }
  {
    std::lock_guard<std::mutex> lock(m_job_q_);
    jobs_.push(std::move(job));
  }
  cv_job_q_.notify_one();
}

}  // namespace ThreadPool

void work(int t, int id) {
  printf("%d start \n", id);
  std::this_thread::sleep_for(std::chrono::seconds(t));
  printf("%d end after %ds\n", id, t);
}

int main() {
  ThreadPool::ThreadPool pool(3);

  for (int i = 0; i < 10; i++) {
    pool.EnqueueJob([i]() { work(i % 3 + 1, i); });
  }
}

• Thread Pool의 전체 코드는 이런 모습이 될 것이다.

Thread Pool 리팩토링

이전의 thread pool은 개념적으로 매우 좋으나 사실 받는 인자가 없고 무조건 void를 반환하는 다소 유연하지 않은 thread pool이다. 따라서 이번에는 템플릿을 활용해 제대로 된 thread pool을 만들어 보자.

설계 1: 작업 추가하는 부분 수정

// job 을 추가한다.
template <typename F, typename... Args>
std::future<typename std::result_of<F(Args...)>::type> 
EnqueueJob(F f, Args... args);

• 벌써 뭔가 이상하다...? 하지만 하나씩 차근차근 확인해보자.
• typename... 은 가변 타입 파라메터 팩으로 타입이 여러개 올 수 있는 것을 의미한다. 즉 함수에 인자가 몇개가 올지 모르기 때문에 가변으로 해둔 것이다.
• 그리고 당연히 EnqueueJob의 반환값은 원래 함수가 반환하는 타입으로 한다. 즉 typename std::result_of<F(Args...)>::type을 반환한다.
• future 타입으로 반환하는 이유는 워커 스레드가 나중에 계산해 돌려줄 결과를 호출한 쪽이 볼 수 있게 하기 위함이다.

jobs_.push([f, args...]() { f(args...); });

• push는 f와 arfs...를 캡쳐로 받아서 람다함수를 넘겨준다.
• 근데 이렇게 하면 f(args...)의 리턴값은 또 어떻게 알 수 있을까?
• 이는 이전에 배운 package_task를 이용한다.

using return_type = typename std::result_of<F(Args...)>::type;
std::packaged_task<return_type()> job(std::bind(f, args...));

• using은 타입에 별명을 붙이는 기능을 수행한다.
• package_task<타입 이름()>은 나중에 operator()가 호출이 되면 타입 값을 계산해서 promise/future 공유 상태에 저장하는 래퍼가 된다.
• bind는 f와 args...를 묶어서 job()을 호출하면 f(args...)를 호출하게 해준다.
• 즉, 원래 jobs_가 void를 리턴하는 함수들을 원하기 때문에 bind를 통해 맞춰준다.

std::future<return_type> job_result_future = job.get_future();
{
  std::lock_guard<std::mutex> lock(m_job_q_);
  jobs_.push([&job]() { job(); });
}

• packaged_task 내부에는 promise를 가지고 있기 때문에 get_future()를 통해 promise와 연결된 future<return_type>을 하나 꺼낸다.
• job_result_future은 나중에 job이 실행되면 return_type값을 받아볼 수 있는 티켓의 개념이다.

전체 코드

namespace ThreadPool {
class ThreadPool {
 public:
  ThreadPool(size_t num_threads);
  ~ThreadPool();

  // job 을 추가한다.
  template <class F, class... Args>
  std::future<typename std::result_of<F(Args...)>::type> EnqueueJob(
    F f, Args... args);

 private:
  // 총 Worker 쓰레드의 개수.
  size_t num_threads_;
  // Worker 쓰레드를 보관하는 벡터.
  std::vector<std::thread> worker_threads_;
  // 할일들을 보관하는 job 큐.
  std::queue<std::function<void()>> jobs_;
  // 위의 job 큐를 위한 cv 와 m.
  std::condition_variable cv_job_q_;
  std::mutex m_job_q_;

  // 모든 쓰레드 종료
  bool stop_all;

  // Worker 쓰레드
  void WorkerThread();
};

ThreadPool::ThreadPool(size_t num_threads)
    : num_threads_(num_threads), stop_all(false) {
  worker_threads_.reserve(num_threads_);
  for (size_t i = 0; i < num_threads_; ++i) {
    worker_threads_.emplace_back([this]() { this->WorkerThread(); });
  }
}

void ThreadPool::WorkerThread() {
  while (true) {
    std::unique_lock<std::mutex> lock(m_job_q_);
    cv_job_q_.wait(lock, [this]() { return !this->jobs_.empty() || stop_all; });
    if (stop_all && this->jobs_.empty()) {
      return;
    }

    // 맨 앞의 job 을 뺀다.
    std::function<void()> job = std::move(jobs_.front());
    jobs_.pop();
    lock.unlock();

    // 해당 job 을 수행한다 :)
    job();
  }
}

ThreadPool::~ThreadPool() {
  stop_all = true;
  cv_job_q_.notify_all();

  for (auto& t : worker_threads_) {
    t.join();
  }
}

template <class F, class... Args>
std::future<typename std::result_of<F(Args...)>::type> ThreadPool::EnqueueJob(
  F f, Args... args) {
  if (stop_all) {
    throw std::runtime_error("ThreadPool 사용 중지됨");
  }

  using return_type = typename std::result_of<F(Args...)>::type;
  std::packaged_task<return_type()> job(std::bind(f, args...));

  std::future<return_type> job_result_future = job.get_future();
  {
    std::lock_guard<std::mutex> lock(m_job_q_);
    jobs_.push([&job]() { job(); });
  }
  cv_job_q_.notify_one();

  return job_result_future;
}

}  // namespace ThreadPool

int work(int t, int id) {
  printf("%d start \n", id);
  std::this_thread::sleep_for(std::chrono::seconds(t));
  printf("%d end after %ds\n", id, t);
  return t + id;
}

int main() {
  ThreadPool::ThreadPool pool(3);

  std::vector<std::future<int>> futures;
  for (int i = 0; i < 10; i++) {
    futures.emplace_back(pool.EnqueueJob(work, i % 3 + 1, i));
  }
  for (auto& f : futures) {
    printf("result : %d \n", f.get());
  }
}

• 자 이제 이렇게 하면 무사히 동작할까?

0 start 
5 start 
5 start 
libc++abi: terminating due to uncaught exception of type 
std::__1::future_error: The associated promise has been 
destructed prior to the associated state becoming ready.

• fuck!! 왜 이런 문제가 발생했을까?
• 천천히 읽어보니 promise가 문제가 있는 것 같다.
• future객체가 파괴되어서 그런거 같은데 왜 파괴가 될까?

std::future<typename std::result_of<F(Args...)>::type> ThreadPool::EnqueueJob(
  F f, Args... args) {
  if (stop_all) {
    throw std::runtime_error("ThreadPool 사용 중지됨");
  }

  using return_type = typename std::result_of<F(Args...)>::type;
  std::packaged_task<return_type()> job(std::bind(f, args...));

  std::future<return_type> job_result_future = job.get_future();
  {
    std::lock_guard<std::mutex> lock(m_job_q_);
    jobs_.push([&job]() { job(); });
  }
  cv_job_q_.notify_one();

  return job_result_future;
}

• Enqueue에 정의된 job 객체는 지역 변수이다. 따라서 Enqueue 함수가 리턴하면 파괴되기 때문에 [&job](){job();}이 접근하려고 할때 이미 그 객체는 파괴된 상태가 되는 것이다.

shared_ptr 활용하기

• 이때 shared_ptr에 packaged_task를 보관하게 되면 해결할 수 있을 것이다.

auto job = 
std::make_shared<std::packaged_task<return_type()>>(std::bind(f, args...));

std::future<return_type> job_result_future = job->get_future();
{
  std::lock_guard<std::mutex> lock(m_job_q_);
  jobs_.push([job]() { (*job)(); });
}

auto job =
  std::make_shared<std::packaged_task<return_type()>>(std::bind(f, args...));

• make_shared를 통해 packaged_task를 shared_ptr로 감싼다.
• 이렇게 하면 객체가 사라질때까지 힙 영역에 저장을 하면서 shared_ptr이 살아있게 된다. 즉 job이 살아있게 된다.
• ✅ 이렇게 하면 드디어 성공적으로 thread pool을 만들었다. 동작은 매우 수월하게 하지만 마지막으로 한가지만 고치면 더욱 좋은 코드가 될 것이다.

인자 원본으로 받기

ThreadPool::EnqueueJob(F f, Args... args);

• 지금은 이렇게 인자들을 복사해서 받고있다.
• 나중에 엄청나게 큰 클래스나 STL을 넘기면 큰 복사 비용이 들게 될 것이다.
• 이는 레퍼런스 + std::forward()를 통해 해결할 수 있다.

template <class F, class... Args>
std::future<typename std::result_of<F(Args...)>::type> EnqueueJob(
  F&& f, Args&&... args);

• 일단 인자들을 이렇게 우측값 래퍼런스로 바꿔준다.
• 여기서 &&를 템플릿에서 사용하면 "복사하지 말고, 넘어온 그대로(IValue or RValue)받아서 다음으로 넘길 준비를 한다는 뜻이다.

auto job = std::make_shared<std::packaged_task<return_type()>>(
  std::bind(std::forward<F>(f), std::forward<Args>(args)...));

• 그리고 bind 함수에 forward로 인자를 전달해주면 된다.
• 이렇게 되면 불필요한 복사 없이 Enqueue함수에 인자들을 전달할 수 있게 된다.

참고자료

씹어먹는 C++ <15.스레드풀> - 모두의 코드