From 8ec38fd8039715e1cbb46c4c01f76e5ac96c52f8 Mon Sep 17 00:00:00 2001 From: AgentDudu Date: Tue, 20 Oct 2026 17:26:47 +0700 Subject: [PATCH 1/2] build: track msvc and mingw presets missed by the earlier presets split The MSVC/MinGW presets existed only in the working tree; CI passed because it configures with explicit flags. Now ck.ps1 -Toolchain works from a clean clone. --- CMakePresets.json | 113 ++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 94 insertions(+), 19 deletions(-) diff --git a/CMakePresets.json b/CMakePresets.json index b87722a..9f12453 100644 --- a/CMakePresets.json +++ b/CMakePresets.json @@ -1,56 +1,131 @@ { "version": 3, - "cmakeMinimumRequired": { "major": 3, "minor": 25, "patch": 0 }, + "cmakeMinimumRequired": { + "major": 3, + "minor": 25, + "rc": 0 + }, "configurePresets": [ { - "name": "gcc", + "name": "base", "hidden": true, - "generator": "Ninja", "binaryDir": "${sourceDir}/build/${presetName}", "cacheVariables": { - "CMAKE_C_COMPILER": "C:/mingw64/bin/gcc.exe", - "CMAKE_CXX_COMPILER": "C:/mingw64/bin/g++.exe", - "CMAKE_MAKE_PROGRAM": "C:/mingw64/bin/ninja.exe", - "CMAKE_EXE_LINKER_FLAGS": "-static" + "CMAKE_CXX_COMPILER": "g++", + "CMAKE_BUILD_TYPE": "Debug", + "HTTPSERVER_BUILD_TESTS": "ON" + } + }, + { + "name": "win-base", + "hidden": true, + "binaryDir": "${sourceDir}/build/${presetName}", + "cacheVariables": { + "CMAKE_BUILD_TYPE": "Debug", + "HTTPSERVER_BUILD_TESTS": "ON" + } + }, + { + "name": "msvc-x64", + "displayName": "MSVC x64 (Visual Studio)", + "inherits": "win-base", + "generator": "Visual Studio 18 2026", + "architecture": "x64" + }, + { + "name": "mingw-x64", + "displayName": "MinGW-w64 x64 (g++ / Ninja)", + "inherits": "win-base", + "generator": "Ninja", + "cacheVariables": { + "CMAKE_CXX_COMPILER": "g++" + }, + "environment": { + "PATH": "C:/msys64/mingw64/bin;C:/mingw64/bin;$penv{PATH}" } }, { "name": "gcc-debug", "displayName": "GCC Debug", - "inherits": "gcc", - "cacheVariables": { "CMAKE_BUILD_TYPE": "Debug" } + "inherits": "base" }, { "name": "gcc-release", "displayName": "GCC Release", - "inherits": "gcc", - "cacheVariables": { "CMAKE_BUILD_TYPE": "Release" } + "inherits": "base", + "cacheVariables": { + "CMAKE_BUILD_TYPE": "Release" + } }, { "name": "gcc-asan", "displayName": "GCC ASan+UBSan", - "inherits": "gcc", + "inherits": "base", "cacheVariables": { - "CMAKE_BUILD_TYPE": "Debug", - "HTTP_SERVER_ENABLE_SANITIZERS": "ON" + "HTTPSERVER_SANITIZER": "address,undefined" } } ], "buildPresets": [ - { "name": "gcc-debug", "configurePreset": "gcc-debug" }, - { "name": "gcc-release", "configurePreset": "gcc-release" }, - { "name": "gcc-asan", "configurePreset": "gcc-asan" } + { + "name": "msvc-x64-debug", + "displayName": "MSVC x64 Debug", + "configurePreset": "msvc-x64", + "configuration": "Debug" + }, + { + "name": "msvc-x64-release", + "displayName": "MSVC x64 Release", + "configurePreset": "msvc-x64", + "configuration": "Release" + }, + { + "name": "mingw-x64-debug", + "displayName": "MinGW x64 Debug", + "configurePreset": "mingw-x64" + }, + { + "name": "gcc-debug", + "configurePreset": "gcc-debug" + }, + { + "name": "gcc-release", + "configurePreset": "gcc-release" + }, + { + "name": "gcc-asan", + "configurePreset": "gcc-asan" + } ], "testPresets": [ + { + "name": "msvc-x64-debug", + "configurePreset": "msvc-x64", + "configuration": "Debug", + "output": { + "outputOnFailure": true + } + }, + { + "name": "mingw-x64-debug", + "configurePreset": "mingw-x64", + "output": { + "outputOnFailure": true + } + }, { "name": "gcc-debug", "configurePreset": "gcc-debug", - "output": { "outputOnFailure": true } + "output": { + "outputOnFailure": true + } }, { "name": "gcc-asan", "configurePreset": "gcc-asan", - "output": { "outputOnFailure": true } + "output": { + "outputOnFailure": true + } } ] } From 7e64e6e31f101c29dfdee756dff74fcdfc80217c Mon Sep 17 00:00:00 2001 From: AgentDudu Date: Tue, 20 Oct 2026 18:09:32 +0700 Subject: [PATCH 2/2] feat: acceptor plus K reactor threads with socket hand-off inbox Server::run now runs an acceptor (the calling thread) plus K reactor threads (threads=0 -> hardware_concurrency). Accepted sockets move to a reactor over a mutex-guarded inbox woken by net::Wakeup (POSIX socketpair; Windows loopback pair) - a Connection is owned by exactly one reactor, so connection state stays lock-free. Round-robin placement; deadlines, backpressure and drain-then- drop stop are per-reactor. Fix found on the Linux gate: SIGINT turns accept poll into EINTR; the error path used to join reactors and the normal stop path joined them again -> pthread_join EINVAL out of run(). Interrupted now just re-checks stop, and stopAll joins only joinable threads. Tests: cross-reactor keep-alive (12 conns / 4 reactors), stop-drain across all reactors, pipelined storm with interleaved fresh clients - all under CI TSan. Linux gates re-run under Docker Desktop after podman/VS disappeared from this machine; clang-format now invoked via LLVM path in the local gate. --- CMakeLists.txt | 1 + CMakePresets.json | 3 +- PLAN.md | 2 +- docs/ARCHITECTURE.md | 16 ++- src/main.cpp | 11 ++- src/net/detail_raw.h | 4 +- src/net/raw_posix.cpp | 7 ++ src/net/raw_win.cpp | 36 +++++++ src/net/wakeup.cpp | 34 +++++++ src/net/wakeup.h | 29 ++++++ src/server/server.cpp | 225 +++++++++++++++++++++++++++++++----------- src/server/server.h | 18 ++-- tests/test_server.cpp | 113 +++++++++++++++++++++ 13 files changed, 425 insertions(+), 74 deletions(-) create mode 100644 src/net/wakeup.cpp create mode 100644 src/net/wakeup.h diff --git a/CMakeLists.txt b/CMakeLists.txt index 766216f..674867f 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -61,6 +61,7 @@ endif() add_library(net STATIC src/net/socket_common.cpp + src/net/wakeup.cpp ${net_platform_srcs}) target_include_directories(net PUBLIC src) if(WIN32) diff --git a/CMakePresets.json b/CMakePresets.json index 9f12453..49d040f 100644 --- a/CMakePresets.json +++ b/CMakePresets.json @@ -2,8 +2,7 @@ "version": 3, "cmakeMinimumRequired": { "major": 3, - "minor": 25, - "rc": 0 + "minor": 25 }, "configurePresets": [ { diff --git a/PLAN.md b/PLAN.md index e8a4a00..629b37c 100644 --- a/PLAN.md +++ b/PLAN.md @@ -174,7 +174,7 @@ phase/step/milestone wording, ever. | ✓ | ~~`feature/posix-socket-layer`~~ | The seam: portable `net/` API + `raw_win`/`raw_posix` backends; `accept4` + `MSG_NOSIGNAL` + `EINTR`; cross-platform socket tests; Docker gate builds whole Linux tree | merged — both gates green | | C | `feature/ci-matrix` | Actions matrix: Linux job = sanitizer container (`tools/asan-inside.sh`), Windows = native MinGW + ctest | merged — PR #5 gated itself green on first run | | R | `feature/reactor-and-connection` | single-threaded poll reactor: connection state machine, timeouts (408), POLLOUT backpressure, signals (self-pipe / console ctrl), graceful drain | merged — 200/keep-alive/408/400/stop integration tests; ASan + TSan green; SIGINT-exit-0 smoke in the Linux gate | -| T | `feature/reactor-threading` | acceptor thread + K reactors + hand-off inbox; `Connection` stays owned by one reactor | TSan green on concurrent tests; RSS stays flat | +| ✓ | ~~`feature/reactor-threading`~~ | acceptor thread + K reactors + socket hand-off inbox (`net::Wakeup`) | merged — cross-reactor keep-alive/pipelining/stop-drain tests; TSan green; EINTR double-join fixed | | ✓ | ~~`feature/static-files`~~ | root mapping, traversal guard, MIME, index, 404/405, HEAD (`std::filesystem`) + per-connection write cap | merged — traversal/HEAD/405/slow-loris/pipelined-burst tests both OSes; ASan + TSan green | | G | `feature/caching-and-range` | ETag, Last-Modified, If-None-Match, Range/206/416 | 304 + 206 tests incl. multi-range rejection | | H | `feature/hardening` | slow-body limit, smuggling corpus, soak (pipelining + slow-client tests landed with S) | ASan clean; 60 s soak flat RSS on Linux | diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 0d474a6..b380e7f 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -22,11 +22,17 @@ until then. obsolete folding, duplicate `Content-Length`, `Content-Length`+`Transfer-Encoding`, non-token names/methods, unknown versions. URL decode/normalise and MIME live with the static handler (they are a serving concern, not a parsing one). -- `server/` — single-threaded poll reactor (`Server::run`): accept batch, per-connection - deadlines (408), graceful stop (drain queued responses, grace cap, then drop). - `Connection` = socket + `RequestParser` + capped output buffer (`max_queued_bytes`): - a slow reader throttles the drain — queued responses stop growing, remaining - pipelined requests wait in the parser, POLLOUT or the next read resumes it. +- `server/` — acceptor thread + K reactor threads (`Server::run`, `threads=0` → + hardware_concurrency). The acceptor batches accepts and hands each socket to a + reactor round-robin over a mutex-guarded inbox plus a `net::Wakeup` socketpair; + a `Connection` (socket + `RequestParser` + capped output buffer + `max_queued_bytes`) is owned by exactly one reactor forever — no connection + state is ever shared or locked. Poll timeouts come from per-connection + deadlines (408); graceful stop drains queued responses per reactor, grace cap, + then drops. EINTR from poll is just a re-check (the stop path is the only + joiner). A slow reader throttles its connection — queued responses stop + growing, remaining pipelined requests wait in the parser, POLLOUT or the next + read resumes the drain. Handlers return `Reply{status, content_type, body}`; HEAD keeps an honest `Content-Length` and no body; handler exceptions become 500 at the reactor seam. `StaticFiles` serves a canonical root: strict percent-decode before resolution, diff --git a/src/main.cpp b/src/main.cpp index 2f7dc35..2f8aea9 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -13,9 +13,17 @@ int main(int argc, char** argv) { port = std::atoi(argv[1]); } if (port < 1 || port > 65535) { - std::fprintf(stderr, "usage: http_server [port] [root]\n"); + std::fprintf(stderr, "usage: http_server [port] [root] [threads]\n"); return 2; } + unsigned threads = 0; // 0 = hardware_concurrency + if (argc > 3) { + threads = static_cast(std::atoi(argv[3])); + if (threads > 128) { + std::fprintf(stderr, "usage: threads must be 0-128\n"); + return 2; + } + } try { net::Context ctx; @@ -23,6 +31,7 @@ int main(int argc, char** argv) { const std::string root = argc > 2 ? argv[2] : "."; server::StaticFiles files(root); server::Server server(static_cast(port), files); + server.threads = threads; std::printf("serving %s on http://127.0.0.1:%d/ (stop with Ctrl-C)\n", files.root().string().c_str(), port); std::fflush(stdout); diff --git a/src/net/detail_raw.h b/src/net/detail_raw.h index 45a19fc..e6877a5 100644 --- a/src/net/detail_raw.h +++ b/src/net/detail_raw.h @@ -10,7 +10,9 @@ namespace detail { // OS primitives. Error detail stays in the OS error slot (WSAGetLastError/errno); // read it via net::lastError() immediately after a failed call. -handle_t rawSocket(); // stream socket, IPv4 (CloExec where supported) +handle_t rawSocket(); // stream socket, IPv4 (CloExec where supported) +// Connected non-blocking bidirectional pipe for poll-set wake-ups (both valid). +bool rawSocketPair(handle_t out[2]); int rawClose(handle_t); // 0 on success bool rawSetNonBlocking(handle_t, bool); handle_t rawAccept(handle_t listener); // raw; then call rawAfterAccept diff --git a/src/net/raw_posix.cpp b/src/net/raw_posix.cpp index 2842fcb..abfd758 100644 --- a/src/net/raw_posix.cpp +++ b/src/net/raw_posix.cpp @@ -12,6 +12,13 @@ handle_t rawSocket() { return ::socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0); } +bool rawSocketPair(handle_t out[2]) { + if (::socketpair(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0, out) != 0) { + return false; + } + return true; +} + int rawClose(handle_t handle) { return ::close(handle); } diff --git a/src/net/raw_win.cpp b/src/net/raw_win.cpp index e04050b..10dd702 100644 --- a/src/net/raw_win.cpp +++ b/src/net/raw_win.cpp @@ -11,6 +11,42 @@ handle_t rawSocket() { return ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); } +// No socketpair() on Winsock: build the equivalent from a loopback listener. +// Setup runs blocking (one-time); the pair leaves non-blocking. +bool rawSocketPair(handle_t out[2]) { + const handle_t l = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + if (l == kInvalidHandle) { + return false; + } + sockaddr_in a{}; + a.sin_family = AF_INET; + a.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + a.sin_port = 0; + int len = sizeof(a); + bool ok = ::bind(l, reinterpret_cast(&a), sizeof(a)) != SOCKET_ERROR && + ::listen(l, 1) != SOCKET_ERROR && + ::getsockname(l, reinterpret_cast(&a), &len) != SOCKET_ERROR; + const handle_t c = ok ? ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP) : kInvalidHandle; + if (ok && c != kInvalidHandle) { + ok = ::connect(c, reinterpret_cast(&a), sizeof(a)) != SOCKET_ERROR; + } + if (ok) { + out[0] = ::accept(l, nullptr, nullptr); + out[1] = c; + ok = out[0] != kInvalidHandle; + } + ::closesocket(l); + if (!ok) { + if (c != kInvalidHandle) { + ::closesocket(c); + } + return false; + } + rawSetNonBlocking(out[0], true); + rawSetNonBlocking(out[1], true); + return true; +} + int rawClose(handle_t handle) { return ::closesocket(handle); } diff --git a/src/net/wakeup.cpp b/src/net/wakeup.cpp new file mode 100644 index 0000000..07020f4 --- /dev/null +++ b/src/net/wakeup.cpp @@ -0,0 +1,34 @@ +#include "net/wakeup.h" + +#include "net/detail_raw.h" + +#include + +namespace net { + +Wakeup Wakeup::create() { + detail::handle_t pair[2] = {detail::kInvalidHandle, detail::kInvalidHandle}; + if (!detail::rawSocketPair(pair)) { + return Wakeup{}; + } + return Wakeup{Socket{pair[0]}, Socket{pair[1]}}; +} + +void Wakeup::notify() { + if (!out_.valid()) { + return; + } + const char byte = 'w'; + out_.sendSome(&byte, 1); // WouldBlock simply means a wake-up is already pending +} + +void Wakeup::drain() { + if (!in_.valid()) { + return; + } + char buf[64]; + while (in_.recvSome(buf, sizeof(buf)) == static_cast(sizeof(buf))) { + } +} + +} // namespace net diff --git a/src/net/wakeup.h b/src/net/wakeup.h new file mode 100644 index 0000000..9ff1699 --- /dev/null +++ b/src/net/wakeup.h @@ -0,0 +1,29 @@ +#pragma once + +#include "net/socket.h" + +#include + +namespace net { + +// Cross-thread wake-up for a poll loop: notify() (any thread) makes the thread +// blocked in pollSockets() return by making readHandle() readable. POSIX uses a +// socketpair; Windows uses a loopback pair — it is just a socket either way. +class Wakeup { +public: + Wakeup() = default; // invalid on failure (lastError() has the reason) + static Wakeup create(); + + detail::handle_t readHandle() const { return in_.handle(); } + bool valid() const { return in_.valid() && out_.valid(); } + + void notify(); // fire-and-forget; coalesces + void drain(); // consume pending wake-ups + +private: + Wakeup(Socket in, Socket out) : in_(std::move(in)), out_(std::move(out)) {} + Socket in_; + Socket out_; +}; + +} // namespace net diff --git a/src/server/server.cpp b/src/server/server.cpp index 2ec2675..82ac0e4 100644 --- a/src/server/server.cpp +++ b/src/server/server.cpp @@ -2,66 +2,97 @@ #include "net/error.h" #include "net/poll.h" +#include "net/wakeup.h" +#include "server/connection.h" #include "util/time.h" #include +#include +#include #include -#include +#include +#include +#include namespace server { Server::Server(std::uint16_t port, Handler handler) : handler_(std::move(handler)) { listener_ = net::Socket::tcp4(); - bool ok = listener_.valid(); - if (ok) { - ok = listener_.setReuseAddress(); + if (!listener_.valid()) { + throw std::runtime_error("socket creation failed (os error " + + std::to_string(net::lastError().code) + ")"); } - if (ok) { - ok = listener_.bindAny(port); + if (!listener_.setReuseAddress() || !listener_.setNonBlocking(true) || + !listener_.bindAny(port) || !listener_.listen(128)) { + throw std::runtime_error("listener setup failed (os error " + + std::to_string(net::lastError().code) + ")"); } - if (ok) { - ok = listener_.setNonBlocking(true); // accepted sockets inherit (both OSes) - } - if (ok) { - ok = listener_.listen(128); +} + +// One reactor thread: inbox + its connections. Everything but the inbox is +// touched only by the reactor thread; Connection has no locking because after +// hand-off exactly one thread owns it. +class Server::Reactor { +public: + explicit Reactor(const Server& owner) : owner_(owner), wake_(net::Wakeup::create()) {} + + bool valid() const { return wake_.valid(); } + + void post(net::Socket socket) { + { + const std::lock_guard g(mu_); + inbound_.push_back(std::move(socket)); + } + wake_.notify(); } - if (!ok) { - const auto e = net::lastError(); - throw std::runtime_error("listener setup failed (os error " + std::to_string(e.code) + ")"); + + void requestStop() { + { + const std::lock_guard g(mu_); + stop_ = true; + } + wake_.notify(); } + + void run(); + +private: + void reap(); + + const Server& owner_; + net::Wakeup wake_; + std::mutex mu_; + std::vector inbound_; // accepted, not yet owned by this reactor + bool stop_ = false; // both under mu_ + std::vector> conns_; // reactor-thread-only +}; + +void Server::Reactor::reap() { + conns_.erase(std::remove_if(conns_.begin(), conns_.end(), + [](const std::unique_ptr& c) { return c->dead(); }), + conns_.end()); } -void Server::run(ShutdownSignal& signal) { +void Server::Reactor::run() { net::PollSet set; bool stopping = false; std::uint64_t hard = 0; for (;;) { - if (!stopping && signal.stopRequested()) { - stopping = true; - listener_.close(); - hard = util::monotonicMs() + static_cast(shutdown_grace.count()); - for (auto& c : conns_) { - c->beginShutdown(); - } - } - if (stopping && (conns_.empty() || util::monotonicMs() >= hard)) { - break; + if (stopping && util::monotonicMs() >= hard) { + break; // grace expired with bytes still queued — drop } set.clear(); - if (!stopping) { - set.add(listener_.handle(), net::PollIn); - } - signal.preparePoll(set); - const std::size_t base = set.size(); // connection i ↔ index base + i + set.add(wake_.readHandle(), net::PollIn); for (auto& c : conns_) { set.add(c->socket().handle(), net::PollIn | (c->wantsWrite() ? static_cast(net::PollOut) : static_cast(0))); } + const std::size_t base = set.size(); // inbox merges append; they join next round const std::uint64_t now = util::monotonicMs(); - int timeout = -1; // -1: block until event (Linux self-pipe wakes on signal) + int timeout = -1; // wake socket makes hand-offs and stop instant; no cap needed for (auto& c : conns_) { if (c->dead()) { continue; @@ -71,9 +102,6 @@ void Server::run(ShutdownSignal& signal) { timeout = timeout < 0 ? static_cast(left) : std::min(timeout, static_cast(left)); } - if (const int cap = static_cast(signal.pollCap().count()); cap > 0) { - timeout = timeout < 0 ? cap : std::min(timeout, cap); - } if (stopping) { const long long left = hard >= now ? static_cast(hard - now) : 0; timeout = @@ -83,33 +111,45 @@ void Server::run(ShutdownSignal& signal) { const int r = net::pollSockets(set, timeout); if (r < 0) { if (net::lastError().kind == net::ErrorKind::Interrupted) { - continue; // EINTR: rerun the wait (state may have changed) + continue; // EINTR: rerun the wait } throw std::runtime_error("poll failed"); } - signal.afterPoll(set); - if (!stopping && signal.stopRequested()) { - continue; // stop detected mid-poll: next iteration shuts down - } - const std::size_t existing = conns_.size(); // conns added this round join next poll - if (!stopping && (set.revents(0) & net::PollIn)) { - for (int i = 0; i < accept_batch; ++i) { - net::Socket s = listener_.accept(); - if (!s.valid()) { - break; // none pending (or accept error — retried next wakeup) - } - conns_.push_back(std::make_unique(std::move(s), handler_, idle_timeout, - max_queued_bytes)); + std::vector fresh; + bool stop_now = false; + { + const std::lock_guard g(mu_); + fresh.swap(inbound_); + stop_now = stop_; + } + wake_.drain(); + if (!fresh.empty()) { + conns_.reserve(conns_.size() + fresh.size()); + for (auto& s : fresh) { + conns_.push_back(std::make_unique( + std::move(s), owner_.handler_, owner_.idle_timeout, owner_.max_queued_bytes)); } } + if (stop_now && !stopping) { + stopping = true; + for (auto& c : conns_) { + c->beginShutdown(); + } + hard = util::monotonicMs() + static_cast(owner_.shutdown_grace.count()); + } + if (stopping && conns_.empty()) { + break; + } - for (std::size_t i = 0; i < existing; ++i) { - Connection& c = *conns_[i]; + // Inbox merges joined after this poll round; they get events next round. + // base-1 = connections present in `set`. + for (std::size_t i = 1; i < base; ++i) { + Connection& c = *conns_[i - 1]; if (c.dead()) { continue; } - const short re = set.revents(base + i); + const short re = set.revents(i); if (re & net::PollIn) { c.onReadable(); } @@ -129,14 +169,85 @@ void Server::run(ShutdownSignal& signal) { } reap(); } - conns_.clear(); // grace expired with bytes still queued — drop - signal.notifyStopped(); + conns_.clear(); } -void Server::reap() { - conns_.erase(std::remove_if(conns_.begin(), conns_.end(), - [](const std::unique_ptr& c) { return c->dead(); }), - conns_.end()); +void Server::run(ShutdownSignal& signal) { + unsigned k = threads; + if (k == 0) { + k = std::thread::hardware_concurrency(); + if (k == 0) { + k = 1; + } + } + + std::vector> reactors; + std::vector pool; + const auto stopAll = [&] { + for (auto& r : reactors) { + r->requestStop(); + } + for (auto& t : pool) { + if (t.joinable()) { + t.join(); + } + } + }; + try { + for (unsigned i = 0; i < k; ++i) { + auto reactor = std::make_unique(*this); + if (!reactor->valid()) { + throw std::runtime_error("wake-up socketpair failed"); + } + Reactor* raw = reactor.get(); + reactors.push_back(std::move(reactor)); + pool.emplace_back([raw] { raw->run(); }); + } + } catch (...) { + for (auto& r : reactors) { + r->requestStop(); + } + for (auto& t : pool) { + t.join(); + } + throw; + } + + net::PollSet set; + std::size_t next = 0; + while (!signal.stopRequested()) { + set.clear(); + set.add(listener_.handle(), net::PollIn); + signal.preparePoll(set); + + const int cap = static_cast(signal.pollCap().count()); + const int r = net::pollSockets(set, cap > 0 ? cap : -1); + if (r < 0) { + if (net::lastError().kind == net::ErrorKind::Interrupted) { + continue; // signal (e.g. SIGINT): re-check stop at the loop top + } + stopAll(); + throw std::runtime_error("accept poll failed"); + } + signal.afterPoll(set); + if (signal.stopRequested()) { + continue; + } + + if (set.revents(0) & net::PollIn) { + for (int i = 0; i < accept_batch; ++i) { + net::Socket s = listener_.accept(); + if (!s.valid()) { + break; // none pending (or accept error — retried next wakeup) + } + reactors[next++ % reactors.size()]->post(std::move(s)); + } + } + } + + listener_.close(); + stopAll(); + signal.notifyStopped(); } } // namespace server diff --git a/src/server/server.h b/src/server/server.h index e8a92f6..d0935eb 100644 --- a/src/server/server.h +++ b/src/server/server.h @@ -1,20 +1,24 @@ #pragma once #include "net/socket.h" -#include "server/connection.h" #include "server/handler.h" #include "server/shutdown.h" #include +#include #include #include -#include namespace server { -// Single-threaded poll reactor: one listener, N keep-alive connections, one -// fixed handler. run() returns after the stop signal, once queued responses -// drained or the grace expired. Threading (worker hand-off) comes later. +// Acceptor thread plus K reactor threads (run() calls from the acceptor). +// Each accepted connection is handed to one reactor and stays owned by exactly +// that reactor forever — no connection state is ever shared. Hand-off is a +// socket move over a mutex-protected inbox plus a Wakeup socket; reactors +// never lock anything else. +// +// run() returns after the stop signal, once every reactor drained its queued +// responses or hit the grace cap. class Server { public: // Throws std::runtime_error when bind/listen fails. @@ -30,13 +34,13 @@ class Server { std::chrono::milliseconds shutdown_grace{5000}; // drain budget after stop int accept_batch = 64; // accepts per poll wakeup std::size_t max_queued_bytes = 1u << 20; // per-connection response backlog + unsigned threads = 0; // reactor count; 0 = hardware_concurrency private: - void reap(); + class Reactor; // defined in server.cpp; reactor thread owns its connections net::Socket listener_; Handler handler_; - std::vector> conns_; }; } // namespace server diff --git a/tests/test_server.cpp b/tests/test_server.cpp index e663eb9..1cdef4d 100644 --- a/tests/test_server.cpp +++ b/tests/test_server.cpp @@ -269,3 +269,116 @@ TEST_CASE("HEAD answers with truthful Content-Length and no body") { sig.stop.store(true); runner.join(); } + +// ---- threading (T) --------------------------------------------------------- +// +// The acceptor hands each connection to one reactor round-robin; these cases +// put connections on several reactors at once and prove the hand-off keeps +// per-connection semantics identical (keep-alive, pipelining, drain-on-stop). +// The CI TSan job runs all of this across real threads. + +TEST_CASE("many keep-alive connections spread across 4 reactors") { + net::Context ctx; + ManualSignal sig; + std::atomic serves{0}; + server::Server srv(0, [&serves](const http::Request&) { + return okReply("r" + std::to_string(serves.fetch_add(1))); + }); + srv.threads = 4; + srv.idle_timeout = std::chrono::seconds(5); + const unsigned short port = srv.port(); + std::thread runner([&] { srv.run(sig); }); + for (int i = 0; i < 100 && srv.port() == 0; ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + + std::vector conns; + for (int i = 0; i < 12; ++i) { + auto c = connectTo(port); + REQUIRE(c.valid()); + REQUIRE(sendReq(c, "GET /a HTTP/1.1\r\n\r\n")); // request 1 of 2 + conns.push_back(std::move(c)); + } + for (auto& c : conns) { + CHECK(readResponse(c).find("HTTP/1.1 200 OK") != std::string::npos); + } + // Second request per connection: keep-alive still works on whichever + // reactor owns it, and the reply is the next response on that socket. + for (auto& c : conns) { + REQUIRE(sendReq(c, "GET /b HTTP/1.1\r\n\r\n")); + } + for (auto& c : conns) { + CHECK(readResponse(c).find("HTTP/1.1 200 OK") != std::string::npos); + } + + sig.stop.store(true); + runner.join(); +} + +TEST_CASE("graceful stop drains connections held by every reactor") { + net::Context ctx; + ManualSignal sig; + server::Server srv(0, [](const http::Request&) { return okReply("drain"); }); + srv.threads = 4; + const unsigned short port = srv.port(); + std::thread runner([&] { srv.run(sig); }); + for (int i = 0; i < 100 && srv.port() == 0; ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + + // 8 connections -> round-robin lands on all four reactors. Serving one + // request each guarantees the hand-off happened before stop. + std::vector conns; + for (int i = 0; i < 8; ++i) { + auto c = connectTo(port); + REQUIRE(c.valid()); + REQUIRE(sendReq(c, "GET / HTTP/1.1\r\n\r\n")); + CHECK(readResponse(c).find("200") != std::string::npos); + conns.push_back(std::move(c)); + } + + sig.stop.store(true); + runner.join(); // returns once every reactor drained its queue + for (auto& c : conns) { + char buf[1]; + CHECK(c.recvSome(buf, 1) == 0); // every reactor closed its connections + } +} + +TEST_CASE("pipelined storms and fresh connections cross reactors together") { + net::Context ctx; + ManualSignal sig; + server::Server srv(0, [](const http::Request&) { return okReply("storm"); }); + srv.threads = 4; + srv.idle_timeout = std::chrono::seconds(5); + const unsigned short port = srv.port(); + std::thread runner([&] { srv.run(sig); }); + for (int i = 0; i < 100 && srv.port() == 0; ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + + std::string burst; + for (int i = 0; i < 5; ++i) { + burst += "GET /s HTTP/1.1\r\n\r\n"; + } + std::vector conns; + std::vector sent; + for (int round = 0; round < 3; ++round) { + auto c = connectTo(port); // a fresh client while the others are busy + REQUIRE(c.valid()); + conns.push_back(std::move(c)); + sent.push_back(0); + for (std::size_t i = 0; i < conns.size(); ++i) { + REQUIRE(sendReq(conns[i], burst)); // everyone bursts again + ++sent[i]; + } + } + for (std::size_t i = 0; i < conns.size(); ++i) { + for (int j = 0; j < sent[i]; ++j) { + CHECK(readResponse(conns[i]).find("200 OK") != std::string::npos); + } + } + + sig.stop.store(true); + runner.join(); +}