C++ 스레드 풀 직접 구현하기: 작업 큐, packaged_task로 결과 받기, 종료 시 데드락
이 글의 핵심
조건 변수로 작업을 기다리는 워커 스레드와 작업 큐로 스레드 풀을 직접 구현하고, future·packaged_task로 결과를 돌려받는 방법을 정리합니다. 우선순위 큐 확장과 종료 시 데드락, 작업 안 예외 처리 같은 자주 발생하는 문제도 다룹니다.
스레드 풀이란?
미리 생성된 스레드들로 작업을 효율적으로 처리
std::thread 하나를 생성하는 비용은 결코 작지 않습니다 — OS는 스레드마다 커널 자료구조를 할당하고, 스택 메모리를 예약하며, 스케줄러에 새 실행 단위를 등록해야 합니다. 짧은 작업 1000개를 처리하기 위해 매번 thread t(task)로 새 스레드를 만들고 버린다면, 실제 작업 시간보다 스레드 생성·소멸 오버헤드가 더 클 수 있습니다. 게다가 무제한으로 스레드를 만들면 CPU 코어 수를 훨씬 초과하는 스레드가 동시에 경쟁하게 되어, 컨텍스트 스위칭 비용만 늘어나고 실질적인 처리량은 오히려 떨어지는 “스레드 폭주”가 발생할 수 있습니다. 스레드 풀은 이 두 문제를 한 번에 해결합니다 — 미리 정해진 개수의 스레드를 한 번만 만들어 두고, 작업이 들어올 때마다 그 스레드들에게 나눠 재사용시켜 생성 비용을 상각하는 동시에 동시 실행 스레드 수를 코어 수에 맞게 제한합니다.
// 매번 스레드 생성 (비효율)
for (int i = 0; i < 1000; i++) {
thread t(task);
t.detach();
}
// 스레드 풀 (효율적)
ThreadPool pool(4);
for (int i = 0; i < 1000; i++) {
pool.enqueue(task);
}기본 구현
이 구현의 핵심은 각 워커 스레드가 실행하는 무한 루프입니다 — cv.wait(lock, predicate)는 predicate가 참이 될 때까지(여기서는 “큐에 작업이 있거나 종료 신호가 왔을 때까지”) 락을 풀고 잠들었다가, 조건이 충족되면 락을 다시 잡고 깨어나는 조건 변수의 표준 사용법입니다. 이 대기 방식이 중요한 이유는, 만약 조건 변수 없이 while (tasks.empty()) {}처럼 바쁜 대기(busy-wait)로 구현했다면 작업이 없는 동안에도 각 워커 스레드가 CPU 코어 하나씩을 100% 점유하며 아무 일도 하지 않는 낭비가 생기기 때문입니다. stop 플래그와 cv.notify_all()을 소멸자에서 함께 쓰는 것도 의도적입니다 — 대기 중인 모든 워커를 깨워 stop && tasks.empty() 조건으로 루프를 빠져나가게 하지 않으면, join()이 영원히 반환되지 않는 채로 소멸자가 멈춰버립니다.
#include <thread>
#include <queue>
#include <functional>
#include <condition_variable>
#include <future>
class ThreadPool {
private:
vector<thread> workers;
queue<function<void()>> tasks;
mutex mtx;
condition_variable cv;
bool stop = false;
public:
ThreadPool(size_t numThreads) {
for (size_t i = 0; i < numThreads; i++) {
workers.emplace_back([this] {
while (true) {
function<void()> task;
{
unique_lock<mutex> lock(mtx);
cv.wait(lock, [this] {
return stop || !tasks.empty();
});
if (stop && tasks.empty()) {
return;
}
task = move(tasks.front());
tasks.pop();
}
task();
}
});
}
}
template<class F>
void enqueue(F&& f) {
{
unique_lock<mutex> lock(mtx);
tasks.emplace(forward<F>(f));
}
cv.notify_one();
}
~ThreadPool() {
{
unique_lock<mutex> lock(mtx);
stop = true;
}
cv.notify_all();
for (auto& worker : workers) {
worker.join();
}
}
};future 지원
앞의 기본 구현은 void() 작업만 큐에 넣을 수 있고, 작업의 결과값을 돌려받거나 완료를 기다릴 방법이 없다는 한계가 있습니다. std::packaged_task가 이 문제를 해결합니다 — 임의의 반환 타입을 가진 호출 가능 객체를 감싸 실행 결과(또는 던져진 예외)를 std::future를 통해 나중에 꺼내 볼 수 있게 만들어 주는 래퍼입니다. make_shared<packaged_task<...>>로 감싸는 이유는 packaged_task가 이동 전용(move-only) 타입인데, 큐에 저장하는 function<void()>는 복사 가능해야 하기 때문입니다 — shared_ptr로 감싸면 람다 캡처([task]() { (*task)(); })가 포인터만 복사하므로 이 제약을 우회할 수 있습니다. enqueue가 res(task의 get_future() 결과)를 작업을 큐에 넣기 전에 미리 얻어 두는 순서도 중요합니다 — 작업이 다른 스레드에서 이미 실행을 시작한 뒤에 future를 요청하면 경쟁 조건이 생길 수 있으므로, 항상 작업을 큐에 넣기 전에 future를 확보해 호출자에게 반환합니다.
인터넷에 퍼져 있는 스레드 풀 예제 상당수가 std::result_of를 쓰는데, 이 타입 특성은 C++17에서 deprecated되고 C++20에서 제거되었습니다. -std=c++20으로 빌드하면 'result_of' is not a member of 'std' 같은 에러가 나므로 std::invoke_result_t<F, Args...>로 바꿔야 합니다. 또 bind 대신 C++20에서는 람다의 팩 초기화 캡처([f = forward<F>(f), ...args = forward<Args>(args)])로 인자를 옮길 수 있어 이동 전용 인자도 자연스럽게 다룰 수 있습니다. 소멸 중인 풀에 작업을 넣는 경우를 막기 위해 stop 확인을 추가한 것도 원본 예제들이 흔히 빠뜨리는 부분입니다.
class ThreadPool {
private:
vector<thread> workers;
queue<function<void()>> tasks;
mutex mtx;
condition_variable cv;
bool stop = false;
public:
ThreadPool(size_t numThreads) {
for (size_t i = 0; i < numThreads; i++) {
workers.emplace_back([this] {
while (true) {
function<void()> task;
{
unique_lock<mutex> lock(mtx);
cv.wait(lock, [this] {
return stop || !tasks.empty();
});
if (stop && tasks.empty()) {
return;
}
task = move(tasks.front());
tasks.pop();
}
task();
}
});
}
}
template<class F, class... Args>
auto enqueue(F&& f, Args&&... args)
-> future<invoke_result_t<F, Args...>>
{
// result_of는 C++17에서 deprecated, C++20에서 제거됨 → invoke_result 사용
using return_type = invoke_result_t<F, Args...>;
auto task = make_shared<packaged_task<return_type()>>(
bind(forward<F>(f), forward<Args>(args)...)
);
future<return_type> res = task->get_future();
{
unique_lock<mutex> lock(mtx);
if (stop) throw runtime_error("enqueue on stopped ThreadPool");
tasks.emplace([task]() { (*task)(); });
}
cv.notify_one();
return res;
}
~ThreadPool() {
{
unique_lock<mutex> lock(mtx);
stop = true;
}
cv.notify_all();
for (auto& worker : workers) {
worker.join();
}
}
};실전 예시
예시 1: 병렬 계산
작업을 먼저 전부 제출한 뒤 결과를 나중에 모으는 이 “제출-수집 분리” 패턴이 스레드 풀을 쓸 때 가장 흔한 형태입니다. 10개의 compute 호출이 4개의 워커 스레드에 나뉘어 병렬로 실행되므로, 순차 실행이라면 1000ms(각 100ms × 10) 걸릴 작업이 약 300ms에 끝납니다. 4개 스레드가 한 번에 4개씩 처리하므로 4 + 4 + 2개, 세 번의 라운드가 필요하기 때문이며, 작업 수가 스레드 수의 배수가 아니면 마지막 라운드에서 일부 스레드가 노는 이 “꼬리” 효과가 생깁니다.result.get()을 반복문에서 순서대로 호출하는 것은 결과가 도착한 순서가 아니라 제출한 순서대로 값을 받는다는 뜻이며, 앞쪽 future가 아직 완료되지 않았다면 그 get() 호출에서 블로킹되어 기다립니다.
int compute(int x) {
this_thread::sleep_for(chrono::milliseconds(100));
return x * x;
}
int main() {
ThreadPool pool(4);
vector<future<int>> results;
// 작업 제출
for (int i = 0; i < 10; i++) {
results.emplace_back(pool.enqueue(compute, i));
}
// 결과 수집
for (auto& result : results) {
cout << result.get() << endl;
}
}예시 2: 파일 처리
이 예시는 결과값이 필요 없는 “실행 후 잊기(fire-and-forget)” 유형의 작업에 enqueue를 쓰는 경우입니다. 주석에 적힌 “자동으로 완료 대기(소멸자)“가 정확히 무슨 뜻인지 짚어볼 필요가 있습니다 — main 함수가 끝나며 pool이 스코프를 벗어나면 ThreadPool의 소멸자가 호출되고, 그 소멸자는 stop = true를 설정한 뒤 모든 워커에 대해 join()을 호출합니다. join()은 해당 스레드가 큐에 남은 작업을 마저 처리하고 종료할 때까지 블로킹되므로, 명시적으로 “모든 파일 처리가 끝날 때까지 기다려라”라고 쓰지 않아도 pool 객체의 수명 자체가 그 보장을 제공합니다.
void processFile(const string& filename) {
cout << "처리 중: " << filename << endl;
this_thread::sleep_for(chrono::milliseconds(500));
}
int main() {
ThreadPool pool(4);
vector<string> files = {
"file1.txt", "file2.txt", "file3.txt",
"file4.txt", "file5.txt", "file6.txt"
};
for (const auto& file : files) {
pool.enqueue(processFile, file);
}
// 자동으로 완료 대기 (소멸자)
}예시 3: 웹 크롤러
crawl 함수가 자기 자신을 다시 pool.enqueue로 제출하는 이 재귀적 패턴은 편리해 보이지만 실무에서는 신중하게 다뤄야 합니다. 링크가 계속 발견되는 한 큐에 새 작업이 끊임없이 추가되므로, 실제 웹사이트를 대상으로 하면 이 코드는 종료 조건이 없는 채로 무한정 확장될 수 있습니다(visited 집합이 중복 방문은 막아 주지만, 사이트 규모나 최대 깊이를 제한하지 않으면 여전히 통제 불가능하게 커집니다). mtx로 visited 접근을 보호한 것은 여러 워커 스레드가 동시에 같은 집합을 읽고 쓸 수 있기 때문에 반드시 필요한 처리이며, 이 락이 없다면 set에 대한 동시 삽입이 데이터 경쟁(data race)이 되어 정의되지 않은 동작을 유발합니다. 마지막 줄의 sleep_for(seconds(2))도 실전 코드에서는 대체할 필요가 있는 자리표시자입니다 — 실제로는 “미해결 작업 수”를 세는 카운터나 배리어로 모든 크롤링이 끝났음을 확인해야지, 고정된 시간만큼 잠들어 있다가 아직 끝나지 않은 작업을 방치한 채 프로그램을 종료해서는 안 됩니다.
이 클래스에는 눈에 잘 띄지 않는 수명 버그도 있습니다. C++에서 멤버는 선언의 역순으로 소멸하므로, pool이 가장 먼저 선언된 WebCrawler에서는 mtx와 visited가 먼저 파괴되고 pool이 마지막에 소멸하면서 남은 작업을 처리합니다. 그 사이 워커가 실행하는 작업은 이미 파괴된 visited와 mtx를 건드리게 되어 정의되지 않은 동작이 됩니다. 스레드 풀을 멤버로 둘 때는 풀을 가장 마지막에 선언해 가장 먼저 소멸(= 작업을 모두 끝내고 스레드를 join)하게 만드는 것이 규칙입니다. 이런 버그는 대부분 프로그램 종료 시점에만 가끔 크래시로 나타나서, “종료할 때만 가끔 죽는다”는 증상을 보면 가장 먼저 멤버 선언 순서를 확인해 볼 만합니다.
#include <set>
class WebCrawler {
private:
ThreadPool pool;
set<string> visited;
mutex mtx;
public:
WebCrawler(size_t numThreads) : pool(numThreads) {}
void crawl(const string& url) {
{
lock_guard<mutex> lock(mtx);
if (visited.count(url)) {
return;
}
visited.insert(url);
}
pool.enqueue([this, url]() {
cout << "크롤링: " << url << endl;
// 페이지 다운로드
this_thread::sleep_for(chrono::milliseconds(100));
// 링크 추출 (시뮬레이션)
vector<string> links = {
url + "/page1",
url + "/page2"
};
for (const auto& link : links) {
crawl(link);
}
});
}
};
int main() {
WebCrawler crawler(4);
crawler.crawl("http://example.com");
this_thread::sleep_for(chrono::seconds(2));
}예시 4: 이미지 처리
예시 1과 구조는 같지만, 여기서는 int 대신 Image라는 구조체 전체를 future로 주고받는다는 점이 다릅니다 — result_of 기반의 enqueue 시그니처 덕분에 반환 타입이 무엇이든(원시 타입, 구조체, 심지어 void) 동일한 enqueue 함수 하나로 대응할 수 있다는 것이 이 스레드 풀 설계의 실질적인 장점입니다. 이미지 6장을 4개 스레드로 처리하면 스레드 수보다 작업 수가 많으므로, 먼저 끝난 워커가 큐에서 다음 이미지를 자동으로 집어가는 방식으로 자연스럽게 부하가 분산됩니다 — 작업을 스레드에 미리 고정 배분하지 않고 공유 큐에서 각자 꺼내가게 하는 이 방식이 워크 스틸링과 유사한 효과를 내며, 작업마다 처리 시간이 다를 때도 특정 스레드만 놀고 다른 스레드만 바쁜 불균형을 줄여줍니다.
struct Image {
string filename;
int width, height;
};
Image processImage(const string& filename) {
cout << "처리 중: " << filename << endl;
this_thread::sleep_for(chrono::milliseconds(200));
return {filename, 800, 600};
}
int main() {
ThreadPool pool(4);
vector<string> images = {
"img1.jpg", "img2.jpg", "img3.jpg",
"img4.jpg", "img5.jpg", "img6.jpg"
};
vector<future<Image>> results;
for (const auto& img : images) {
results.emplace_back(pool.enqueue(processImage, img));
}
for (auto& result : results) {
Image img = result.get();
cout << img.filename << ": "
<< img.width << "x" << img.height << endl;
}
}우선순위 큐
기본 구현의 queue<function<void()>>를 priority_queue<Task>로 바꾸면 “먼저 들어온 순서”가 아니라 “우선순위가 높은 순서”로 작업을 처리할 수 있습니다. 다만 이 구현에는 실제로 컴파일은 되지만 미묘하게 잘못된 부분이 있습니다 — std::priority_queue::top()은 const 참조를 반환하는데, move(tasks.top().func)처럼 const 객체에 std::move를 적용하면 실제로는 이동이 일어나지 않고 function의 복사 생성자가 호출됩니다(rvalue-reference-to-const는 이동 생성자가 아니라 복사 생성자 오버로드에 바인딩되기 때문입니다). function<void()>가 캡처한 람다가 무겁지 않다면 눈에 띄는 문제는 아니지만, 큰 상태를 캡처하는 작업이 많다면 이 지점에서 기대한 이동 최적화가 조용히 무효화되고 있다는 점을 알아둘 가치가 있습니다 — 진짜 이동이 필요하다면 std::priority_queue 대신 vector<Task>와 std::push_heap/std::pop_heap을 직접 쓰는 방법이 있습니다. pop_heap은 최우선 원소를 벡터의 맨 뒤로 옮겨 주므로, tasks.back()에서 비const로 이동한 뒤 pop_back()하면 됩니다.
우선순위 큐에는 공정성 문제도 있습니다. priority_queue는 같은 우선순위끼리의 순서를 보장하지 않아 먼저 넣은 작업이 나중에 실행될 수 있고, 높은 우선순위 작업이 계속 들어오면 낮은 우선순위 작업은 영원히 실행되지 않는 기아(starvation)가 생깁니다. 같은 우선순위 안에서 FIFO를 지키려면 Task에 증가하는 순번을 넣어 비교에 함께 쓰고, 기아가 문제라면 대기 시간이 길어질수록 우선순위를 올려 주는 에이징(aging)을 적용합니다.
class PriorityThreadPool {
private:
struct Task {
int priority;
function<void()> func;
bool operator<(const Task& other) const {
return priority < other.priority; // 높은 우선순위 먼저
}
};
vector<thread> workers;
priority_queue<Task> tasks;
mutex mtx;
condition_variable cv;
bool stop = false;
public:
PriorityThreadPool(size_t numThreads) {
for (size_t i = 0; i < numThreads; i++) {
workers.emplace_back([this] {
while (true) {
function<void()> task;
{
unique_lock<mutex> lock(mtx);
cv.wait(lock, [this] {
return stop || !tasks.empty();
});
if (stop && tasks.empty()) {
return;
}
task = move(tasks.top().func);
tasks.pop();
}
task();
}
});
}
}
template<class F>
void enqueue(int priority, F&& f) {
{
unique_lock<mutex> lock(mtx);
tasks.push({priority, forward<F>(f)});
}
cv.notify_one();
}
~PriorityThreadPool() {
{
unique_lock<mutex> lock(mtx);
stop = true;
}
cv.notify_all();
for (auto& worker : workers) {
worker.join();
}
}
};자주 발생하는 문제
문제 1: 데드락
이 문제는 워커 스레드 수가 유한하다는 사실을 간과할 때 발생합니다. 스레드 풀에 워커가 4개뿐인데, 그 4개가 모두 “다른 작업의 결과를 기다리는” 작업을 실행 중이고 그 다른 작업들이 아직 큐에서 실행될 차례를 못 얻었다면, 아무도 그 작업을 실행해 줄 워커가 남아있지 않아 영원히 대기하게 됩니다 — 이는 스레드 풀 특유의 데드락으로, 스레드 개수가 충분히 크면(작업 깊이보다 많으면) 우연히 피해질 수도 있지만 재현 조건이 부하나 스레드 수에 따라 달라지므로 발견하기 어렵습니다. 안전한 규칙은 풀 안에서 실행 중인 작업이 같은 풀에 새 작업을 넣고 그 결과를 동기적으로 기다리지 않는 것이며, 작업 간 의존이 꼭 필요하다면 별도의 전용 풀을 두거나 콜백/컨티뉴에이션 기반으로 재구성해 동기 대기 자체를 없애는 것이 근본적인 해법입니다.
// ❌ 작업이 다른 작업 대기
pool.enqueue([&pool]() {
auto f = pool.enqueue([] { return 42; });
f.get(); // 데드락 가능
});
// ✅ 중첩 작업 피하기문제 2: 예외 처리
기본 구현(void() 작업)에서 작업 안의 예외가 잡히지 않으면, 그 예외는 워커 스레드의 최상위까지 전파되어 std::terminate를 호출하고 프로그램 전체가 종료됩니다 — 예외가 어느 작업에서 발생했는지, 무슨 예외였는지에 대한 정보도 없이 프로세스가 죽어버리는 것이 최악의 실패 모드입니다. packaged_task 기반 enqueue는 이 문제를 근본적으로 해결합니다 — packaged_task는 감싼 함수가 예외를 던지면 그것을 잡아 future의 공유 상태에 저장해 두고, 호출자가 f.get()을 호출하는 시점에 그 예외를 그대로 다시 던져줍니다. 즉 예외가 발생한 워커 스레드는 안전하게 다음 작업으로 넘어가고, 호출자는 자신이 원하는 시점에 try/catch로 그 예외를 정상적으로 처리할 수 있습니다 — 이것이 프로덕션 스레드 풀이 거의 예외 없이 future 기반 인터페이스를 제공하는 이유입니다.
// ❌ 예외 무시
pool.enqueue([] {
throw runtime_error("에러");
});
// ✅ future로 예외 전달
auto f = pool.enqueue([] {
throw runtime_error("에러");
return 42;
});
try {
f.get();
} catch (const exception& e) {
cout << e.what() << endl;
}문제 3: 스레드 수
CPU 바운드 작업(순수 연산 위주)에서 물리 코어 수보다 훨씬 많은 스레드를 만들면, 실제로 동시에 실행될 수 있는 스레드는 코어 수만큼인데 나머지는 컨텍스트 스위칭만 유발하며 서로의 실행 시간을 빼앗습니다 — ThreadPool pool(1000)이 4코어 머신에서 오히려 4개짜리 풀보다 느릴 수 있는 이유입니다. hardware_concurrency()가 CPU 바운드 작업의 합리적인 기본값인 이유는 정확히 이 지점을 겨냥합니다 — 물리적으로 동시에 실행 가능한 만큼만 스레드를 만들어 컨텍스트 스위칭 오버헤드를 최소화합니다. 다만 이 기준은 작업이 순수 연산일 때만 유효하며, 작업이 파일 I/O나 네트워크 대기를 포함해 자주 블로킹된다면 코어 수보다 많은 스레드를 두어야 블로킹된 스레드가 쉬는 동안 다른 스레드가 CPU를 활용할 수 있습니다 — 이 글 상단 FAQ에서 “I/O 바운드는 코어 수의 2~4배”라고 언급한 것이 이 이유 때문입니다.
// ❌ 너무 많은 스레드
ThreadPool pool(1000); // 오버헤드
// ✅ CPU 코어 수 기반
ThreadPool pool(thread::hardware_concurrency());FAQ
Q1: 스레드 풀은 언제 사용하나요?
A:
- 많은 작은 작업
- 스레드 생성 비용 절감
- 리소스 제한
Q2: 스레드 수는?
A:
- CPU 바운드:
hardware_concurrency() - I/O 바운드: 더 많이 (2-4배)
Q3: 성능 향상은?
A: 스레드 생성/소멸 비용 제거. 작은 작업에서 큰 효과.
Q4: 우선순위는?
A: priority_queue 사용.
Q5: 작업 취소는?
A: atomic<bool> 플래그로 구현.
Q6: 스레드 풀 학습 리소스는?
A:
- “C++ Concurrency in Action”
- Boost.Asio
- “Effective Modern C++“