C++17 스레드 풀 만들기: 작업 큐, future 반환, 우선순위·취소, 요청마다 스레드를 만들 때의 문제
들어가며: 작업마다 스레드를 만들면 안 되는 이유
요청을 병렬로 처리하려고 루프 안에서 작업마다 std::thread를 만드는 코드는 흔한 출발점입니다.
// 작업마다 스레드 생성
void process_requests(const std::vector<Request>& requests) {
std::vector<std::thread> threads;
for (const auto& req : requests) {
threads.emplace_back([&req] { handle_request(req); }); // 요청 1000개 → 스레드 1000개
}
for (auto& t : threads) t.join();
}요청이 적을 때는 문제가 없지만 수가 늘면 비용이 드러납니다. 스레드마다 스택 메모리(Linux 기본 8MB 가상 주소 공간, 실제로는 쓴 만큼만 물리 메모리)를 예약하고, 생성과 종료에 시스템 콜이 들며, 코어 수보다 훨씬 많은 스레드가 동시에 실행 가능 상태가 되면 컨텍스트 스위칭과 캐시 경쟁으로 처리량이 오히려 떨어집니다. 프로세스당 스레드 수 한도에 걸리면 std::system_error로 생성 자체가 실패합니다. std::async(std::launch::async, ...)도 libstdc++와 libc++에서는 호출마다 새 스레드를 만들므로 같은 문제가 있습니다(MSVC의 구현은 내부 스레드 풀을 씁니다).
스레드 풀은 미리 만든 고정 개수의 워커 스레드가 공유 작업 큐에서 작업을 꺼내 실행하는 구조입니다. 스레드 생성 비용을 한 번만 치르고, 동시에 실행되는 작업 수를 워커 수로 제한합니다.
ThreadPool pool(4);
std::vector<std::future<void>> done;
for (const auto& req : requests) {
done.push_back(pool.submit([&req] { handle_request(req); }));
}
for (auto& f : done) f.get(); // 모든 요청 처리 완료까지 대기 (req 참조가 유효한 동안)이 글은 C++17로 기본 스레드 풀을 만들고, std::future로 결과를 돌려받는 submit, 종료 방식, 우선순위, 백프레셔를 붙인 뒤, 스레드 풀에서 자주 나는 버그를 정리합니다.
스레드 풀의 구조
flowchart TB
subgraph main[제출하는 스레드]
S1[submit 작업1]
S2[submit 작업2]
S3[submit 작업3]
end
subgraph queue[작업 큐: mutex + condition_variable]
Q[FIFO 큐]
end
subgraph workers[워커 스레드]
W1[워커 1]
W2[워커 2]
W3[워커 3]
end
S1 --> Q
S2 --> Q
S3 --> Q
Q --> W1
Q --> W2
Q --> W3
submit()이 작업(호출 가능한 객체)을 큐에 넣고condition_variable로 대기 중인 워커 하나를 깨웁니다.- 워커는 락을 잡고 큐에서 작업 하나를 꺼낸 뒤, 락을 풀고 작업을 실행합니다.
- 큐가 비어 있으면 워커는
condition_variable에서 잠들어 CPU를 쓰지 않습니다. - 종료 요청을 받으면 워커는 큐가 빌 때까지 처리한 뒤 루프를 빠져나옵니다.
stateDiagram-v2
[*] --> 대기: 풀 시작
대기 --> 실행: 큐에서 작업 획득
실행 --> 실행: 작업 완료, 큐에 작업 남음
실행 --> 대기: 작업 완료, 큐 비어 있음
대기 --> 종료: 종료 요청 && 큐 비어 있음
작업을 실행하는 동안 락을 잡고 있지 않는 것이 핵심입니다. 락 안에서 작업을 실행하면 다른 워커가 큐에 접근하지 못해 풀이 사실상 단일 스레드가 됩니다.
future를 반환하는 스레드 풀
#include <condition_variable>
#include <functional>
#include <future>
#include <memory>
#include <mutex>
#include <queue>
#include <stdexcept>
#include <thread>
#include <type_traits>
#include <vector>
class ThreadPool {
public:
explicit ThreadPool(size_t num_threads) {
if (num_threads == 0) num_threads = 1; // hardware_concurrency()가 0을 돌려줄 수 있음
for (size_t i = 0; i < num_threads; ++i) {
workers_.emplace_back([this] { worker_loop(); });
}
}
ThreadPool(const ThreadPool&) = delete;
ThreadPool& operator=(const ThreadPool&) = delete;
template <typename F, typename... Args>
auto submit(F&& f, Args&&... args) -> std::future<std::invoke_result_t<F, Args...>> {
using R = std::invoke_result_t<F, Args...>;
auto task = std::make_shared<std::packaged_task<R()>>(
[f = std::forward<F>(f), tup = std::make_tuple(std::forward<Args>(args)...)]() mutable {
return std::apply(std::move(f), std::move(tup));
});
std::future<R> result = task->get_future();
{
std::lock_guard<std::mutex> lock(mutex_);
if (stop_) throw std::runtime_error("submit on stopped pool");
tasks_.emplace([task] { (*task)(); });
}
cv_.notify_one();
return result;
}
// 큐가 비고 실행 중인 작업도 없을 때까지 대기 (풀은 계속 사용 가능)
void wait_idle() {
std::unique_lock<std::mutex> lock(mutex_);
idle_cv_.wait(lock, [this] { return tasks_.empty() && busy_ == 0; });
}
// 큐에 남은 작업을 모두 처리한 뒤 워커 종료
~ThreadPool() {
{
std::lock_guard<std::mutex> lock(mutex_);
stop_ = true;
}
cv_.notify_all();
for (auto& w : workers_) w.join();
}
private:
void worker_loop() {
for (;;) {
std::function<void()> task;
{
std::unique_lock<std::mutex> lock(mutex_);
cv_.wait(lock, [this] { return stop_ || !tasks_.empty(); });
if (stop_ && tasks_.empty()) return;
task = std::move(tasks_.front());
tasks_.pop();
++busy_;
}
task(); // packaged_task는 예외를 future에 담으므로 여기서 던지지 않음
{
std::lock_guard<std::mutex> lock(mutex_);
--busy_;
if (tasks_.empty() && busy_ == 0) idle_cv_.notify_all();
}
}
}
std::vector<std::thread> workers_;
std::queue<std::function<void()>> tasks_;
std::mutex mutex_;
std::condition_variable cv_;
std::condition_variable idle_cv_;
size_t busy_ = 0;
bool stop_ = false;
};몇 가지 설계 결정을 짚어 보겠습니다.
std::function은 복사 가능한 호출 객체만 담을 수 있는데 std::packaged_task는 이동만 가능합니다. 그래서 packaged_task를 shared_ptr로 감싸 람다가 포인터를 복사해 들고 있게 했습니다. C++23의 std::move_only_function을 쓸 수 있다면 이 우회가 필요 없습니다.
반환 타입은 C++17의 std::invoke_result_t로 구합니다. 예전 코드에서 자주 보이는 std::result_of는 C++17에서 폐기 예정이 되었고 C++20에서 제거되었습니다. 인자는 std::bind 대신 튜플에 담아 std::apply로 전달했습니다. std::bind도 동작하지만, 인자를 항상 복사본으로 저장하고 자리표시자·중첩 bind 표현식 같은 특수 규칙이 있어 람다 쪽이 의도가 분명합니다. 어느 방식이든 인자는 복사·이동되어 저장되므로, 참조로 넘기고 싶다면 std::ref를 명시해야 합니다.
작업이 던진 예외는 packaged_task가 잡아 future에 저장하고, 호출자가 get()할 때 다시 던집니다. 따라서 이 구조에서는 워커 루프에 try-catch가 없어도 워커가 죽지 않습니다. 반면 결과를 받지 않는 작업을 std::function<void()>로 직접 큐에 넣는 “fire-and-forget” 방식이라면, 예외가 워커 스레드 밖으로 빠져나가 std::terminate로 프로세스 전체가 종료되므로 워커 루프에서 반드시 잡아야 합니다.
busy_는 락 안에서만 읽고 쓰므로 일반 변수로 충분합니다. wait_idle()이 “큐가 비었다”와 “실행 중인 작업이 없다”를 함께 확인하는 이유는, 큐만 보면 마지막 작업을 꺼내 실행하고 있는 순간을 유휴 상태로 착각하기 때문입니다.
#include <iostream>
int main() {
ThreadPool pool(std::max(1u, std::thread::hardware_concurrency()));
std::vector<long long> data(1'000'000, 1);
auto sum_range = [&data](size_t begin, size_t end) {
long long sum = 0;
for (size_t i = begin; i < end; ++i) sum += data[i];
return sum;
};
const size_t chunks = 8;
const size_t chunk = data.size() / chunks;
std::vector<std::future<long long>> futures;
for (size_t i = 0; i < chunks; ++i) {
size_t b = i * chunk;
size_t e = (i + 1 == chunks) ? data.size() : b + chunk;
futures.push_back(pool.submit(sum_range, b, e));
}
long long total = 0;
for (auto& f : futures) total += f.get();
std::cout << "sum = " << total << "\n"; // 1000000
}sum_range가 data를 참조로 캡처해도 안전한 것은, main이 모든 future의 get()을 마칠 때까지 data가 살아 있기 때문입니다. 마지막 청크는 나눗셈의 나머지를 포함하도록 끝을 data.size()로 잡았습니다.
종료 방식: 남은 작업을 처리할 것인가 버릴 것인가
위 소멸자는 stop_을 세운 뒤에도 워커가 stop_ && tasks_.empty()일 때만 종료하므로, 큐에 남은 작업을 모두 처리한 뒤 끝납니다. 종료 요청 이후에 들어오는 submit만 거부합니다. 서버를 내릴 때 이미 받은 요청은 처리하고 끝내는 정상 종료에는 이 동작이 맞습니다.
반대로 즉시 멈춰야 하는 경우에는 큐에 남은 작업을 버려야 합니다.
// ThreadPool 멤버로 추가
void shutdown_now() {
std::queue<std::function<void()>> dropped;
{
std::lock_guard<std::mutex> lock(mutex_);
stop_ = true;
std::swap(dropped, tasks_); // 큐를 비움
}
cv_.notify_all();
for (auto& w : workers_) if (w.joinable()) w.join();
// dropped가 소멸하면서 packaged_task들도 소멸 → 대기 중인 future.get()은
// std::future_error(broken_promise)를 받음
}버려진 작업의 future를 기다리던 쪽은 broken_promise 예외를 받으므로, 호출자가 이를 “취소됨”으로 처리하도록 정해 두어야 합니다. 이미 실행 중인 작업은 어느 방식으로도 강제로 멈출 수 없습니다. C++에는 스레드를 밖에서 안전하게 중단하는 방법이 없기 때문에, 긴 작업은 스스로 취소 신호를 확인해야 합니다.
void long_running_task(const std::atomic<bool>& cancel) {
for (int i = 0; i < 1'000'000; ++i) {
if (cancel.load(std::memory_order_relaxed)) return; // 주기적으로 확인
do_work(i);
}
}C++20에서는 std::stop_token과 std::stop_source가 같은 역할을 표준화했고, std::jthread는 소멸할 때 자동으로 정지를 요청하고 join합니다.
우선순위 작업 큐
#include <queue>
struct PriorityTask {
int priority;
uint64_t seq; // 같은 우선순위 안에서 제출 순서 유지
mutable std::function<void()> task; // top()이 const 참조라 꺼낼 때 이동하려고 mutable
bool operator<(const PriorityTask& o) const {
if (priority != o.priority) return priority < o.priority; // 숫자가 클수록 먼저
return seq > o.seq; // 먼저 들어온 것이 먼저
}
};
// 워커에서 꺼낼 때 (std::priority_queue<PriorityTask> queue_)
std::function<void()> next = std::move(queue_.top().task); // mutable 멤버라 이동 가능
queue_.pop();std::priority_queue를 쓸 때 놓치기 쉬운 점이 두 가지입니다. 첫째, top()이 const 참조를 돌려주므로 작업을 이동해 꺼내려면 멤버를 mutable로 두거나 별도 처리가 필요합니다. 복사하면 큰 캡처를 가진 작업마다 복사 비용이 듭니다. 둘째, 힙은 같은 우선순위끼리의 순서를 보장하지 않으므로, 제출 순서를 유지하려면 순번(seq)을 비교에 넣어야 합니다.
우선순위 큐에는 기아(starvation) 문제도 있습니다. 높은 우선순위 작업이 계속 들어오면 낮은 우선순위 작업은 영원히 실행되지 않습니다. 대기 시간이 길어진 작업의 우선순위를 올려 주거나(aging), 우선순위별로 큐를 나눠 가중치를 두고 번갈아 꺼내는 방식으로 완화합니다.
백프레셔: 큐 크기 제한
// ThreadPool에 max_queue_와 not_full_cv_ 추가
template <typename F>
auto submit_blocking(F&& f) -> std::future<std::invoke_result_t<F>> {
using R = std::invoke_result_t<F>;
auto task = std::make_shared<std::packaged_task<R()>>(std::forward<F>(f));
auto result = task->get_future();
{
std::unique_lock<std::mutex> lock(mutex_);
not_full_cv_.wait(lock, [this] { return stop_ || tasks_.size() < max_queue_; });
if (stop_) throw std::runtime_error("submit on stopped pool");
tasks_.emplace([task] { (*task)(); });
}
cv_.notify_one();
return result;
}
// 워커는 tasks_.pop() 직후 not_full_cv_.notify_one()을 호출제출 속도가 처리 속도보다 빠르면 무제한 큐는 결국 메모리를 다 씁니다. 큐에 상한을 두고 가득 차면 제출하는 쪽을 기다리게 하면, 생산자가 자연스럽게 소비자 속도에 맞춰집니다. 기다리게 할 수 없는 상황(이벤트 루프 스레드에서 제출하는 경우 등)이라면, 블로킹 대신 즉시 실패를 돌려주는 try_submit을 두고 호출자가 요청을 거절하게 합니다. 대기 중인 submit_blocking이 종료 요청 때 깨어날 수 있도록 stop_을 조건에 넣고, 종료 시 not_full_cv_.notify_all()도 호출해야 합니다.
자주 나는 버그
참조 캡처로 인한 use-after-free
void process(const std::string& path) {
pool.submit([&path] { load_file(path); }); // 위험
} // process가 반환하면 path가 가리키던 객체가 사라질 수 있음submit은 작업을 큐에 넣고 바로 반환하므로, 작업이 실행될 때는 지역 변수나 인자 참조가 이미 무효일 수 있습니다. 대부분은 우연히 동작하다가 부하가 걸려 작업이 늦게 실행될 때만 크래시가 나서 재현이 어렵습니다. 작업이 들고 있어야 하는 데이터는 값으로 캡처하거나([path], 큰 데이터면 [p = std::move(path)]), 여러 작업이 공유한다면 std::shared_ptr로 캡처합니다. 앞의 합산 예제처럼 참조 캡처를 쓰려면, 그 데이터가 모든 작업이 끝날 때까지 산다는 것을 같은 함수 안에서 future를 기다려 보장해야 합니다.
작업 안에서 같은 풀의 결과를 기다리는 데드락
ThreadPool pool(4);
auto f = pool.submit([&pool] {
auto inner = pool.submit([] { return 1; });
return inner.get(); // 워커가 하위 작업을 기다리며 막힘
});작업 안에서 같은 풀에 하위 작업을 제출하는 것 자체는 문제가 없습니다. 위험한 것은 그 결과를 get()으로 기다리는 것입니다. 워커 하나가 기다리는 동안 하위 작업은 다른 워커가 실행해야 하는데, 이런 작업이 워커 수만큼 동시에 실행되면 모든 워커가 기다리기만 하고 하위 작업을 실행할 워커가 남지 않아 풀 전체가 멈춥니다. 워커 수를 늘리면 발생 확률이 줄어들 뿐 사라지지 않습니다.
해결책은 워커 안에서 같은 풀의 결과를 동기적으로 기다리지 않는 것입니다. 하위 작업은 현재 스레드에서 직접 실행하거나, 작업을 단계별로 나눠 다음 단계를 콜백으로 제출하거나, 용도가 다른 작업(예: I/O 대기와 CPU 계산)은 별도 풀로 분리합니다. 기다리는 동안 큐의 다른 작업을 대신 실행하는(work stealing 풀의 “help while waiting”) 방식도 있습니다.
워커 수 정하기
CPU를 쓰는 계산 작업이라면 하드웨어 스레드 수(std::thread::hardware_concurrency()) 정도가 출발점입니다. 그보다 많으면 컨텍스트 스위칭만 늘어납니다. 작업이 I/O를 기다리는 시간이 길다면 기다리는 동안 다른 작업을 돌릴 수 있도록 더 많은 워커가 도움이 되는데, 적정 수는 대기 시간과 계산 시간의 비율에 따라 달라지므로 실제 부하로 측정하며 정합니다. 블로킹 I/O를 많이 기다리는 작업이라면 스레드를 늘리기보다 비동기 I/O(Asio 등)로 바꾸는 편이 근본적입니다.
hardware_concurrency()는 값을 알 수 없으면 0을 돌려주고, 컨테이너의 CPU 제한(cgroup)을 반영하지 않는 구현이 많습니다. 쿠버네티스에서 CPU 2개로 제한한 파드가 64코어 노드에 올라가면 64개 워커를 만드는 일이 생기므로, 운영 환경에서는 설정값으로 워커 수를 받을 수 있게 해 둡니다.
스레드별 데이터의 거짓 공유
워커마다 카운터를 두고 각자 증가시키는데도 느리다면 거짓 공유(false sharing)를 의심합니다. 서로 다른 워커의 카운터가 같은 64바이트 캐시 라인에 있으면, 한 워커가 쓸 때마다 다른 코어의 캐시 라인이 무효화됩니다. alignas(64)(또는 C++17의 std::hardware_destructive_interference_size)로 워커별 데이터를 캐시 라인 경계에 맞춥니다.
활용 예제
병렬 정렬: 기다리는 쪽을 워커 밖에 두기
재귀적으로 반을 나눠 한쪽을 풀에 제출하고 get()으로 기다리는 병렬 퀵소트는 앞에서 본 데드락 패턴 그 자체입니다. 재귀 호출이 워커 안에서 일어나면 워커가 하위 작업을 기다리게 되기 때문입니다. 분할은 호출하는 스레드에서 하고, 잘게 나눈 조각들의 정렬만 풀에 맡긴 뒤 호출하는 스레드에서 모두 기다리면 이 문제를 피할 수 있습니다.
#include <algorithm>
template <typename It>
void collect_sort_tasks(It first, It last, ThreadPool& pool,
std::vector<std::future<void>>& out, int depth) {
if (depth == 0 || last - first < 10'000) {
out.push_back(pool.submit([first, last] { std::sort(first, last); }));
return;
}
auto pivot = *(first + (last - first) / 2);
// 3분할: [pivot보다 작음][pivot과 같음][pivot보다 큼]
It mid1 = std::partition(first, last, [&](const auto& x) { return x < pivot; });
It mid2 = std::partition(mid1, last, [&](const auto& x) { return !(pivot < x); });
collect_sort_tasks(first, mid1, pool, out, depth - 1);
collect_sort_tasks(mid2, last, pool, out, depth - 1);
}
template <typename T>
void parallel_sort(std::vector<T>& v, ThreadPool& pool) {
std::vector<std::future<void>> futures;
collect_sort_tasks(v.begin(), v.end(), pool, futures, 4); // 최대 2^4개 조각
for (auto& f : futures) f.get();
}분할 단계는 호출 스레드에서 순차적으로 돌지만 조각들의 정렬은 병렬로 진행됩니다. 3분할을 쓴 이유는 같은 값이 많은 배열에서 한쪽 구간이 줄지 않아 재귀가 끝나지 않는 문제를 막기 위해서입니다. 실무에서는 C++17 병렬 알고리즘(std::sort(std::execution::par, ...))이 지원되는 환경이라면 그쪽이 더 간단합니다(GCC는 TBB가 필요).
파일 단위 병렬 처리
void make_thumbnails(const std::vector<std::string>& paths, ThreadPool& pool) {
std::vector<std::future<void>> futures;
for (const auto& path : paths) {
futures.push_back(pool.submit([path] { // 경로는 값으로 캡처
auto thumb = resize(load_image(path), 128, 128);
save_image(thumbnail_path(path), thumb);
}));
}
for (auto& f : futures) f.get(); // 첫 번째 예외가 여기서 다시 던져짐
}get()을 순서대로 호출하면 앞의 작업이 예외를 던졌을 때 뒤의 future들을 기다리지 않고 빠져나옵니다. 뒤의 작업들은 계속 실행되므로, 그 작업들이 참조하는 데이터가 사라지지 않도록 모든 future를 기다린 뒤 첫 예외를 다시 던지는 식으로 처리하는 것이 안전합니다.
데이터를 구간으로 나눌 때 나머지 처리
void update_entities(std::vector<Entity>& entities, ThreadPool& pool, size_t chunks) {
std::vector<std::future<void>> futures;
const size_t n = entities.size();
for (size_t i = 0; i < chunks; ++i) {
size_t begin = n * i / chunks;
size_t end = n * (i + 1) / chunks; // 마지막 구간이 나머지를 포함
futures.push_back(pool.submit([&entities, begin, end] {
for (size_t j = begin; j < end; ++j) entities[j].update();
}));
}
for (auto& f : futures) f.get();
}n / chunks로 구간 크기를 고정하면 나눗셈의 나머지 원소들이 어느 구간에도 들어가지 않습니다. n * i / chunks 형태로 경계를 계산하면 모든 원소가 정확히 한 번씩 포함됩니다. 각 작업이 서로 다른 원소만 수정하므로 락이 필요 없지만, update()가 다른 엔티티나 전역 상태를 건드린다면 그 부분은 따로 동기화해야 합니다. 그래픽 API의 그리기 호출처럼 특정 스레드에서만 호출해야 하는 작업은 이런 식으로 나눌 수 없습니다.
직접 만들까, 라이브러리를 쓸까
| 라이브러리 | 특징 | 라이선스 |
|---|---|---|
| BS::thread_pool | 헤더 하나, C++17 이상, 단일 공유 큐, 일시 정지·우선순위 등 기능 | MIT |
| CTPL | 작은 헤더 전용 구현 | Apache 2.0 |
| ThreadPool (progschj) | C++11 최소 구현, 이 글의 기본 구조와 같음 | zlib |
| oneTBB | work stealing 스케줄러, 병렬 알고리즘·컨테이너 포함 | Apache 2.0 |
| HPX | 분산·대규모 병렬 런타임 | BSL-1.0 |
이 글의 구현은 원리를 이해하고 요구사항에 맞게 고치기 위한 출발점입니다. 작업 부하가 불균형해서 한 워커만 바쁘고 나머지가 노는 상황이 잦거나, 작업 안에서 하위 작업을 많이 만드는 구조라면, 워커마다 로컬 큐를 두고 놀고 있는 워커가 다른 워커의 큐에서 작업을 가져오는 work stealing 스케줄러(oneTBB 등)를 검토합니다. 자세한 튜닝은 멀티스레드 성능 튜닝 글에서 다룹니다.
참고 자료
- cppreference - std::thread
- cppreference - std::condition_variable
- cppreference - std::packaged_task
- Anthony Williams, C++ Concurrency in Action, 2nd Edition (Manning)