folly::CPUThreadPoolExecutor — CPU-bound 작업의 표준 thread pool

1
#include <folly/executors/CPUThreadPoolExecutor.h>
2

3
folly::CPUThreadPoolExecutor pool(8);   // 8 threads
4

5
pool.add([] {
6
  doExpensiveWork();
7
});
8

9
// Future와 통합
10
computeSemi()
11
  .via(&pool)
12
  .thenValue([](int x) { return process(x); });

1
// folly/executors/CPUThreadPoolExecutor.h (요약)
2
class CPUThreadPoolExecutor : public ThreadPoolExecutor {
3
 public:
4
  CPUThreadPoolExecutor(
5
      size_t numThreads,
6
      std::unique_ptr<BlockingQueue<CPUTask>> taskQueue =
7
          std::make_unique<UnboundedBlockingQueue<CPUTask>>(),
8
      std::shared_ptr<ThreadFactory> threadFactory =
9
          std::make_shared<NamedThreadFactory>("CPUThreadPool"));
10

11
  void add(Func) override;
12
  void add(Func, std::chrono::milliseconds timeout, Func expireCallback);
13
  void addWithPriority(Func, int8_t priority) override;
14

15
  size_t numActiveThreads() const;
16
  size_t numPendingTasks() const;
17
};

1
// 무제한 — 메모리 폭발 가능
2
auto q1 = std::make_unique<folly::UnboundedBlockingQueue<CPUTask>>();
3

4
// 제한 — 가득 차면 add()가 block
5
auto q2 = std::make_unique<folly::LifoSemMPMCQueue<CPUTask>>(1000);
6

7
// priority queue
8
auto q3 = std::make_unique<folly::PriorityLifoSemMPMCQueue<CPUTask>>(
9
    3,        // 3 priority levels
10
    1000);    // capacity per level
11

12
folly::CPUThreadPoolExecutor pool(8, std::move(q3));
13

14
pool.addWithPriority([] { ... }, folly::Executor::HI_PRI);
15
pool.addWithPriority([] { ... }, folly::Executor::LO_PRI);

1
auto factory = std::make_shared<folly::NamedThreadFactory>("Parser");
2
folly::CPUThreadPoolExecutor pool(4, factory);
3

4
// 생성된 thread는 "Parser-0", "Parser-1", ... 이름을 가짐

1
// 사용자 정의 factory — CPU affinity
2
class AffinityThreadFactory : public folly::ThreadFactory {
3
 public:
4
  std::thread newThread(folly::Func&& f) override {
5
    return std::thread([f = std::move(f)]() mutable {
6
      cpu_set_t set;
7
      CPU_ZERO(&set);
8
      CPU_SET(specificCore_, &set);
9
      pthread_setaffinity_np(pthread_self(), sizeof(set), &set);
10
      f();
11
    });
12
  }
13
};

1
pool.add(
2
    [] { doWork(); },                       // task
3
    std::chrono::milliseconds(100),         // 100ms 내에 시작 안 되면 expire
4
    [] { LOG(WARN) << "task expired"; });   // expireCallback

1
class ThreadPoolExecutor {
2
 public:
3
  void stop();                  // 즉시 신규 task 거절, in-flight task는 진행
4
  void join();                  // 모든 in-flight 끝날 때까지 wait
5
  void setNumThreads(size_t);   // dynamic resize
6
};
7

8
// destructor가 자동 join
9
{
10
  folly::CPUThreadPoolExecutor pool(8);
11
  pool.add(...);
12
}   // ← 여기서 join (in-flight task 끝날 때까지)

1
// std::thread — 직접 manage
2
std::vector<std::thread> threads;
3
for (int i = 0; i < N; ++i) {
4
  threads.emplace_back([] { work(); });
5
}
6
for (auto& t : threads) t.join();
7

8
// std::async — launch policy 모호
9
auto f = std::async(std::launch::async, [] { return work(); });
10
int v = f.get();   // thread 생성/파괴 비용
11

12
// folly::CPUThreadPoolExecutor — pool reuse
13
static folly::CPUThreadPoolExecutor pool(8);
14
pool.add([] { work(); });   // thread 재사용

1
// 1. 매 함수 호출마다 새 pool 생성
2
void handle() {
3
  folly::CPUThreadPoolExecutor pool(4);   // thread 생성/파괴 — 매우 비쌈
4
  pool.add([] { work(); });
5
}   // destructor가 join
6

7
// 2. unbounded queue + slow consumer
8
folly::CPUThreadPoolExecutor pool(2);   // 2 thread
9
for (int i = 0; i < 1'000'000; ++i) {
10
  pool.add([] { sleep(1); });   // 큐가 무한정 자람
11
}
12

13
// 3. pool 안에서 .get() 호출
14
pool.add([&] {
15
  auto v = anotherSemi().via(&pool).get();   // deadlock 위험
16
  // pool worker 가 자기 자신을 wait
17
});
18

19
// 4. CPU pool에서 blocking I/O
20
pool.add([] {
21
  read(fd, buf, n);   // CPU thread 가 I/O로 block — pool starve
22
});