1
2
3
4
5
6

baseLoop (thread-0):  co_await async_accept() → socket
                      ↓
                      co_spawn(workerIoCtx, handleSession(socket))
                      ↓ [eventfd_write → epoll_ctl → pthread_cond_signal]
                      
workerLoop (thread-N): handleSession() → read → route → write

1
2
3
4
5
6
7

// 替换前：单 acceptor
std::unique_ptr<tcp::acceptor> acceptor_;

// 替换后：多 acceptor + 运行时回退
std::vector<std::unique_ptr<tcp::acceptor>> acceptors_;
std::vector<std::unique_ptr<IdleFd>> idleFds_;  // 每个 acceptor 配独立 IdleFd
bool reusePortEnabled_ {false};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

// 收集所有 loop（baseLoop + workers）
std::vector<AsioEventLoop*> allLoops;
allLoops.push_back(&baseLoop_);
if (ioPool_)
    for (auto* loop : ioPool_->getAllLoops())
        allLoops.push_back(loop);

#if defined(__linux__) || defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__)
// 尝试 SO_REUSEPORT：每个 loop 创建独立 acceptor（Linux/macOS/BSD）
for (auto* loop : allLoops)
{
    auto acc = std::make_unique<tcp::acceptor>(loop->getIoContext());
    acc->open(endpoint.protocol());
    acc->set_option(reuse_address(true));
    
    boost::system::error_code ec;
    using reuse_port = boost::asio::detail::socket_option::boolean<SOL_SOCKET, SO_REUSEPORT>;
    acc->set_option(reuse_port(true), ec);
    if (ec) { /* 回退到单 acceptor */ break; }
    
    acc->bind(endpoint);
    acc->listen();
}
#endif

// Windows / 低版本内核自动回退到单 acceptor

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

if (reusePortEnabled_)
{
    // SO_REUSEPORT 路径（Linux/macOS/BSD）：socket 已在当前 loop 上
    auto socket = co_await acceptor.async_accept(use_awaitable);
    boost::asio::co_spawn(
        co_await boost::asio::this_coro::executor,  // 零跨线程
        handleSession(std::move(socket)),
        boost::asio::detached);
}
else
{
    // 回退路径（Windows 等）：socket 直接创建在目标 worker 上
    auto& targetIoCtx = ioPool_->getNextLoop()->getIoContext();
    tcp::socket socket(targetIoCtx.get_executor());  // IOCP 关联在正确线程
    co_await acceptor.async_accept(socket, use_awaitable);
    boost::asio::co_spawn(targetIoCtx.get_executor(),
                          handleSession(std::move(socket)),
                          boost::asio::detached);
}

1
2
3
4
5
6
7

for (;;)  // keep-alive 请求循环
{
    deadline->expires_after(60s);     // → timerfd_settime + epoll_ctl(MOD)
    deadline->async_wait(callback);
    co_await http::async_read(...);
    deadline->cancel();               // → timerfd_settime(0) + epoll_ctl(DEL)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

// 连接级 atomic 活跃时间戳
auto lastActiveMs = std::make_shared<std::atomic<int64_t>>(now_ms());

// 连接级 timer 自续期链（仅启动一次）
auto scheduleCheck = std::make_shared<std::function<void()>>();
*scheduleCheck = [&deadline, &socket, lastAct, timeoutMs, scheduleCheck]()
{
    deadline->expires_after(std::chrono::milliseconds(timeoutMs));
    deadline->async_wait([&socket, lastAct, timeoutMs, scheduleCheck](auto ec)
    {
        if (ec) return;
        auto elapsed = now_ms() - lastAct->load(std::memory_order_relaxed);
        if (elapsed >= timeoutMs)
            socket.close();     // 真正超时
        else
            (*scheduleCheck)(); // 有活动，续期
    });
};
(*scheduleCheck)();

// 请求循环中：零系统调用的活跃标记
for (;;)
{
    co_await http::async_read(socket, buffer, parser, use_awaitable);
    lastActiveMs->store(now_ms(), std::memory_order_relaxed);  // 替代 expires_after + cancel
    // ... 路由处理 ...
    co_await http::async_write(socket, response, use_awaitable);
    lastActiveMs->store(now_ms(), std::memory_order_relaxed);
}

1
2
3
4
5

% time     seconds  usecs/call     calls    errors syscall
------ ----------- ----------- --------- --------- ----------------
100.00    0.011936         119       100           epoll_ctl
------ ----------- ----------- --------- --------- ----------------
100.00    0.011936         119       100           total