folly::IOThreadPoolExecutor — libevent 기반 I/O pool

1
#include <folly/executors/IOThreadPoolExecutor.h>
2
#include <folly/io/async/EventBase.h>
3

4
folly::IOThreadPoolExecutor io(4);   // 4 EventBase, 각 thread당 1개
5

6
// EventBase 직접 가져오기
7
auto* evb = io.getEventBase();   // round-robin으로 선택
8

9
evb->runInEventBaseThread([] {
10
  // 이 lambda는 EventBase loop 안에서 실행됨
11
  registerSocket(...);
12
});
13

14
// 또는 add()로 단순 schedule
15
io.add([] { doIoWork(); });

1
// folly/executors/IOThreadPoolExecutor.h (요약)
2
class IOThreadPoolExecutor : public IOExecutor, public ThreadPoolExecutor {
3
 public:
4
  IOThreadPoolExecutor(
5
      size_t numThreads,
6
      std::shared_ptr<ThreadFactory> threadFactory =
7
          std::make_shared<NamedThreadFactory>("IOThreadPool"),
8
      folly::EventBaseManager* ebm = folly::EventBaseManager::get(),
9
      Options options = {});
10

11
  EventBase* getEventBase() override;   // round-robin
12
  std::vector<EventBase*> getAllEventBases();
13
  void add(Func) override;
14
};

1
Pool (4 threads)
2
├── Thread 0 ── EventBase 0 ── [event_base_loop()]
3
├── Thread 1 ── EventBase 1 ── [event_base_loop()]
4
├── Thread 2 ── EventBase 2 ── [event_base_loop()]
5
└── Thread 3 ── EventBase 3 ── [event_base_loop()]

1
// 안티패턴 — 같은 pool로 CPU + I/O
2
folly::CPUThreadPoolExecutor pool(8);
3
pool.add([] {
4
  auto data = read(fd, ...);   // blocking I/O → CPU thread starve
5
  process(data);
6
});
7

8
// 권장 — 분리
9
folly::IOThreadPoolExecutor io(2);
10
folly::CPUThreadPoolExecutor cpu(8);
11

12
io.getEventBase()->runInEventBaseThread([&] {
13
  auto data = readAsync(fd);    // non-blocking
14
  data.via(&cpu).thenValue(process);
15
});

1
// 단일 EventBase — single thread
2
folly::EventBase evb;
3
std::thread t([&] { evb.loopForever(); });
4

5
evb.runInEventBaseThread([] { ... });
6

7
evb.terminateLoopSoon();
8
t.join();

1
class MyClient : public folly::AsyncSocket::ConnectCallback {
2
  void connectSuccess() noexcept override { ... }
3
  void connectErr(folly::AsyncSocketException const&) noexcept override { ... }
4
};
5

6
io.getEventBase()->runInEventBaseThread([&] {
7
  auto sock = folly::AsyncSocket::newSocket(io.getEventBase());
8
  sock->connect(&client, addr);
9
});

1
io.getEventBase()->runAfterDelay(
2
    [] { LOG(INFO) << "5s elapsed"; },
3
    5000);

1
fetchAsync(url)
2
  .via(io.getEventBase())     // EventBase는 Executor 인터페이스 구현
3
  .thenValue([](Response r) { ... });

1
// 같은 connection은 같은 EventBase에 pin
2
class Connection {
3
  folly::EventBase* evb_;   // 생성 시 한 번 정함
4

5
 public:
6
  Connection(folly::IOThreadPoolExecutor& pool)
7
      : evb_(pool.getEventBase()) {}
8

9
  void send(Msg m) {
10
    evb_->runInEventBaseThread([this, m = std::move(m)]() mutable {
11
      writeImpl(std::move(m));
12
    });
13
  }
14
};

1
// 1. IOThreadPoolExecutor에서 blocking call
2
io.add([] {
3
  std::this_thread::sleep_for(std::chrono::seconds(1));   // EventBase loop 멈춤
4
});
5

6
// 2. 다른 EventBase의 socket을 직접 조작
7
auto* other = io.getAllEventBases()[1];
8
sock->setEventBase(other);   // race — other가 loop 중일 수 있음
9

10
// 3. EventBase loop이 안 도는데 schedule
11
folly::EventBase evb;
12
evb.runAfterDelay([] { ... }, 1000);
13
// evb.loopForever() 호출 안 함 — callback 실행 안 됨
14

15
// 4. IO pool에서 expensive computation
16
io.add([&] {
17
  for (auto& item : million_items) compute(item);   // 다른 connection이 처리 안 됨
18
});

1
folly::EventBase            ≈ asio::io_context
2
folly::IOThreadPoolExecutor ≈ multi-threaded io_context pool
3
runInEventBaseThread        ≈ post() / dispatch()