feat: single-threaded poll reactor with graceful shutdown

.github/workflows/ci.yml

@@ -17,6 +17,22 @@ jobs:
           docker run --rm -v "$GITHUB_WORKSPACE:/src" -w /src http-cpp-asan \
             bash tools/asan-inside.sh
 
+  # Concurrency gate: reactor tests under ThreadSanitizer (Linux only; no TSan
+  # runtime on the Windows MinGW toolchain). Native runner, not a container:
+  # TSan needs vm.mmap_rnd_bits=28, which kernel 6.x containers cannot set.
+  linux-tsan:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Widen TSan address space
+        run: sudo sysctl -w vm.mmap_rnd_bits=28
+      - name: Install toolchain
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y --no-install-recommends g++ cmake ninja-build netcat-openbsd
+      - name: Build and test under TSan
+        run: SAN=thread bash tools/asan-inside.sh
+
   # Windows native gate: MinGW gcc, warnings-as-errors, ctest.
   windows-native:
     runs-on: windows-latest

CMakeLists.txt

@@ -7,6 +7,7 @@ set(CMAKE_CXX_EXTENSIONS OFF)
 set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
 
 option(HTTP_SERVER_ENABLE_SANITIZERS "Build with address and undefined sanitizers" OFF)
+option(HTTP_SERVER_ENABLE_TSAN "Build with ThreadSanitizer (Linux; reactor tests)" OFF)
 
 include(CheckCXXSourceCompiles)
 set(CMAKE_REQUIRED_FLAGS "-fsanitize=address,undefined")
@@ -26,6 +27,14 @@ if(HTTP_SERVER_ENABLE_SANITIZERS)
   add_link_options(-fsanitize=address,undefined)
 endif()
 
+if(HTTP_SERVER_ENABLE_TSAN)
+  if(HTTP_SERVER_ENABLE_SANITIZERS)
+    message(FATAL_ERROR "TSAN and ASAN/UBSAN sanitizer options are mutually exclusive.")
+  endif()
+  add_compile_options(-fsanitize=thread -fno-omit-frame-pointer)
+  add_link_options(-fsanitize=thread)
+endif()
+
 set(HTTP_WARNINGS -Wall -Wextra -Wpedantic -Wshadow -Wconversion -Werror)
 
 # http layer is platform-neutral.
@@ -59,8 +68,21 @@ if(WIN32)
 endif()
 target_compile_options(net PRIVATE ${HTTP_WARNINGS})
 
-add_executable(http_server src/main.cpp)
-target_link_libraries(http_server PRIVATE net http)
+# server: reactor + connection, platform-neutral; signals implementation is per OS.
+add_library(server STATIC
+  src/server/server.cpp
+  src/server/connection.cpp)
+target_link_libraries(server PUBLIC net http)
+target_compile_options(server PRIVATE ${HTTP_WARNINGS})
+
+if(WIN32)
+  set(app_signals_src src/app/signals_win.cpp)
+else()
+  set(app_signals_src src/app/signals_posix.cpp)
+endif()
+
+add_executable(http_server src/main.cpp ${app_signals_src})
+target_link_libraries(http_server PRIVATE net http server)
 target_compile_options(http_server PRIVATE ${HTTP_WARNINGS})
 
 enable_testing()
@@ -69,11 +91,12 @@ add_executable(unit_tests
   tests/test_main.cpp
   tests/test_socket.cpp
   tests/test_request.cpp
-  tests/test_response.cpp)
+  tests/test_response.cpp
+  tests/test_server.cpp)
 target_include_directories(unit_tests SYSTEM PRIVATE external/doctest)
 # doctest is a vendored header: warnings from it are not ours to fix,
 # hence SYSTEM above; our own test code still builds warning-clean.
 target_compile_options(unit_tests PRIVATE -Wall -Wextra -Werror)
-target_link_libraries(unit_tests PRIVATE net http)
+target_link_libraries(unit_tests PRIVATE net http server)
 
 add_test(NAME unit_tests COMMAND unit_tests)

PLAN.md

@@ -173,14 +173,15 @@ phase/step/milestone wording, ever.
 | ✓ | ~~`feature/response-writer`~~ | Response, headers, status, chunk encoder | merged |
 | ✓ | ~~`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` | poll loop, connection state machine, hand-off inbox, timeouts, buffering, signals (win + posix), graceful shutdown — all single-source except `app/signals_*` | integration: in-memory 200, keep-alive, 408 idle, Ctrl-C/SIGTERM exit 0; **TSan job added and green** |
+| 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 |
 | S | `feature/static-files` | root mapping, traversal guard, MIME, index, 404/405, error pages (`std::filesystem`) | status-by-status integration tests; traversal rejected on both OSes |
 | G | `feature/caching-and-range` | ETag, Last-Modified, If-None-Match, Range/206/416 | 304 + 206 tests incl. multi-range rejection |
 | H | `feature/hardening` | pipelining, slow-body limit, max-body/max-heads, request-smuggling corpus, soak | pipelined + slow-client tests; ASan clean; 60 s soak flat RSS on Linux |
 | D | `feature/docs-and-benchmark` | README (build/run per OS, limits), load script, numbers | documented; req/s + RSS recorded on both platforms |
 
 Then `release/0.1.0` → `develop` + PR to `main` (approval required) → tag `v0.1.0`.
-C is next; R needs C.
+R merged single-threaded; S (static files) stacks on it — T (threading) can land after S.
 
 ## 8. Git workflow (gitflow)
 

docs/ARCHITECTURE.md

@@ -21,8 +21,20 @@ until then.
   push/incremental, enforces head/body/count limits, and rejects bare-LF endings,
   obsolete folding, duplicate `Content-Length`, `Content-Length`+`Transfer-Encoding`,
   non-token names/methods, unknown versions. Pending: URL decode/normalise, MIME.
-- `app/` — signal handling (pending, win + posix)
-- `server/` — listener, connection, static file handler (pending)
+- `server/` — single-threaded poll reactor (`Server::run`): accept batch, per-connection
+  deadlines (408), backpressure via POLLOUT, graceful stop (drain queued responses,
+  grace cap, then drop). `Connection` = socket + `RequestParser` + output buffer;
+  protocol errors are answered once (best effort) then the connection drops —
+  smuggling candidates never share a stream. `ShutdownSignal` is the pluggable stop
+  trigger; tests drive it directly.
+- `app/` — shutdown signals, one paragraph precisely because behaviour differs:
+  Linux uses a self-pipe (`pipe2`, handler writes one byte → poll wakes instantly,
+  SIGINT drains and exits 0 — smoke-tested in the container gate); Windows cannot
+  union a waitable event with WSAPoll, so the console-control handler sets a flag
+  and blocks until the loop ends (`notifyStopped()`), with the reactor capped at a
+  100 ms poll while idle. A returning handler would mean "exit now" — hence block.
 
-Gates: `tools/ck.ps1` (Windows) and `tools/ck-asan-docker.sh` (whole tree, ASan+UBSan,
-Ubuntu 24.04 container). The socket test file is the same source on both platforms.
+Gates: `tools/ck.ps1` (Windows), `tools/ck-asan-docker.sh` (whole tree, ASan+UBSan,
+Ubuntu 24.04 container, plus SIGINT graceful-stop smoke), and TSan (`SAN=thread`,
+Linux runner — kernel containers can't set the `vm.mmap_rnd_bits` TSan needs).
+The socket and server test files are the same source on both platforms.

src/app/signals.h

@@ -0,0 +1,32 @@
+#pragma once
+
+#include "server/shutdown.h"
+
+#include <memory>
+
+namespace app {
+
+// Process-level SIGINT/SIGTERM (POSIX) or console-control (Windows) shutdown.
+// Installs OS handlers in the constructor — one instance per process.
+//
+// POSIX: self-pipe — the handler writes a byte, the reactor's poll wakes at once.
+// Windows: the handler sets a flag and blocks (a returning handler means "handled,
+// exit now" would kill the graceful drain); the reactor notices within pollCap().
+class ShutdownSignals final : public server::ShutdownSignal {
+  public:
+    struct Impl; // OS state, defined in the per-OS .cpp
+
+    ShutdownSignals();
+    ~ShutdownSignals() override;
+
+    bool stopRequested() const override;
+    void preparePoll(net::PollSet& set) override;
+    void afterPoll(const net::PollSet& set) override;
+    std::chrono::milliseconds pollCap() const override;
+    void notifyStopped() override;
+
+  private:
+    std::unique_ptr<Impl> impl_;
+};
+
+} // namespace app

src/app/signals_posix.cpp

@@ -0,0 +1,82 @@
+#include "app/signals.h"
+
+#include "net/poll.h"
+
+#include <csignal>
+#include <fcntl.h>
+#include <poll.h> // for the pollfd revents bits (net/poll.h pulls it too)
+#include <unistd.h>
+
+#include <atomic>
+
+struct app::ShutdownSignals::Impl {
+    int pipe[2] = {-1, -1};
+};
+
+namespace {
+app::ShutdownSignals::Impl* g_impl = nullptr;
+std::atomic_bool g_stop{false};
+
+// Async-signal-safe: write() only. The atomic store is a fallback for a failed
+// self-pipe; atomic_bool is lock-free on every platform we target.
+void onSignal(int) {
+    g_stop.store(true);
+    if (g_impl != nullptr && g_impl->pipe[1] >= 0) {
+        const char byte = 'x';
+        const ssize_t ignored = ::write(g_impl->pipe[1], &byte, 1);
+        (void)ignored;
+    }
+}
+} // namespace
+
+namespace app {
+
+ShutdownSignals::ShutdownSignals() : impl_(std::make_unique<Impl>()) {
+    if (::pipe2(impl_->pipe, O_CLOEXEC | O_NONBLOCK) != 0) {
+        impl_->pipe[0] = impl_->pipe[1] = -1;
+        return; // degraded: reactor still re-checks after every poll
+    }
+    g_impl = impl_.get();
+    struct sigaction sa {};
+    sa.sa_handler = onSignal;
+    sigemptyset(&sa.sa_mask);
+    sa.sa_flags = SA_RESTART; // send/poll EINTR handling exists, but avoid churn
+    ::sigaction(SIGINT, &sa, nullptr);
+    ::sigaction(SIGTERM, &sa, nullptr);
+    ::signal(SIGPIPE, SIG_IGN); // belt over MSG_NOSIGNAL's trousers
+}
+
+ShutdownSignals::~ShutdownSignals() {
+    if (impl_->pipe[0] >= 0) {
+        ::close(impl_->pipe[0]);
+        ::close(impl_->pipe[1]);
+    }
+}
+
+bool ShutdownSignals::stopRequested() const { return g_stop.load(); }
+
+void ShutdownSignals::preparePoll(net::PollSet& set) {
+    if (impl_->pipe[0] >= 0) {
+        set.add(impl_->pipe[0], net::PollIn);
+    }
+}
+
+void ShutdownSignals::afterPoll(const net::PollSet& set) {
+    for (std::size_t i = 0; i < set.size(); ++i) {
+        if (set.socket(i) == impl_->pipe[0] && (set.revents(i) & net::PollIn)) {
+            char drain[64];
+            while (::read(impl_->pipe[0], drain, sizeof(drain)) > 0) {
+            }
+            g_stop.store(true);
+            return;
+        }
+    }
+}
+
+std::chrono::milliseconds ShutdownSignals::pollCap() const {
+    return std::chrono::milliseconds(0); // self-pipe wakes poll immediately
+}
+
+void ShutdownSignals::notifyStopped() {}
+
+} // namespace app

src/app/signals_win.cpp

@@ -0,0 +1,66 @@
+#include "app/signals.h"
+
+#include "net/poll.h"
+
+#ifndef WIN32_LEAN_AND_MEAN
+#define WIN32_LEAN_AND_MEAN
+#endif
+#include <windows.h>
+
+#include <atomic>
+
+struct app::ShutdownSignals::Impl {};
+
+namespace {
+std::atomic_bool g_stop{false};
+HANDLE g_done = nullptr; // set once the reactor loop has ended
+
+BOOL WINAPI onConsoleEvent(DWORD type) {
+    if (type != CTRL_C_EVENT && type != CTRL_BREAK_EVENT && type != CTRL_CLOSE_EVENT &&
+        type != CTRL_LOGOFF_EVENT && type != CTRL_SHUTDOWN_EVENT) {
+        return FALSE; // not ours; next handler / default
+    }
+    g_stop.store(true);
+    // Returning TRUE means "handled, continue" — but returning at all races the
+    // 5s close grace, so block until the reactor is done and let process exit
+    // take us out.
+    if (g_done != nullptr) {
+        ::WaitForSingleObject(g_done, 4000);
+    } else {
+        ::Sleep(4000);
+    }
+    return TRUE;
+}
+} // namespace
+
+namespace app {
+
+ShutdownSignals::ShutdownSignals() : impl_(std::make_unique<Impl>()) {
+    g_done = ::CreateEventW(nullptr, TRUE, FALSE, nullptr);
+    ::SetConsoleCtrlHandler(&onConsoleEvent, TRUE);
+}
+
+ShutdownSignals::~ShutdownSignals() {
+    ::SetConsoleCtrlHandler(&onConsoleEvent, FALSE);
+    if (g_done != nullptr) {
+        ::CloseHandle(g_done);
+        g_done = nullptr;
+    }
+}
+
+bool ShutdownSignals::stopRequested() const { return g_stop.load(); }
+
+void ShutdownSignals::preparePoll(net::PollSet& set) { (void)set; }
+void ShutdownSignals::afterPoll(const net::PollSet& set) { (void)set; }
+
+std::chrono::milliseconds ShutdownSignals::pollCap() const {
+    return std::chrono::milliseconds(100); // no waitable socket+event union on WSAPoll
+}
+
+void ShutdownSignals::notifyStopped() {
+    if (g_done != nullptr) {
+        ::SetEvent(g_done);
+    }
+}
+
+} // namespace app

src/main.cpp

@@ -1,2 +1,36 @@
-// Placeholder entry point. Replaced as the socket and server layers land.
-int main() { return 0; }
+#include "app/signals.h"
+#include "http/request.h"
+#include "net/context.h"
+#include "server/server.h"
+
+#include <cstdio>
+#include <cstdlib>
+#include <exception>
+
+int main(int argc, char** argv) {
+    int port = 8080;
+    if (argc > 1) {
+        port = std::atoi(argv[1]);
+    }
+    if (port < 1 || port > 65535) {
+        std::fprintf(stderr, "usage: http_server [port]\n");
+        return 2;
+    }
+
+    try {
+        net::Context ctx;
+        app::ShutdownSignals signals;
+        server::Server server(static_cast<std::uint16_t>(port),
+                              [](const http::Request&) {
+                                  return std::string("Hello from http-server-cpp\n");
+                              });
+        std::printf("listening on http://127.0.0.1:%d/ (stop with Ctrl-C)\n", port);
+        std::fflush(stdout);
+        server.run(signals);
+        std::puts("stopped");
+    } catch (const std::exception& e) {
+        std::fprintf(stderr, "error: %s\n", e.what());
+        return 1;
+    }
+    return 0;
+}

src/net/poll.h

@@ -8,6 +8,13 @@
 
 namespace net {
 
+// Portable poll event bits (POLL* spellings differ across OS headers).
+inline constexpr short PollIn = POLLIN;
+inline constexpr short PollOut = POLLOUT;
+inline constexpr short PollErr = POLLERR;
+inline constexpr short PollHup = POLLHUP;
+inline constexpr short PollNval = POLLNVAL;
+
 // Thin poll()/WSAPoll fd set. Only the first pollSockets() return value of entries
 // carry meaningful revents (Windows leaves the rest untouched).
 class PollSet {

src/net/socket.h

@@ -46,6 +46,9 @@ class Socket {
     // WouldBlock failure returns false; check lastError().kind. The reactor never
     // uses it; it buffers per-connection instead.
     bool sendAll(const char* data, std::size_t len);
+    // What the kernel accepts right now: >0 bytes, -1 error (WouldBlock possible on
+    // a non-blocking socket — check lastError().kind). The reactor's write path.
+    int sendSome(const char* data, std::size_t len);
     // >0 bytes read, 0 peer closed, -1 error (WouldBlock possible on non-blocking).
     int recvSome(void* buffer, std::size_t capacity);
 

src/net/socket_common.cpp

@@ -96,6 +96,13 @@ bool Socket::sendAll(const char* data, std::size_t len) {
     return true;
 }
 
+int Socket::sendSome(const char* data, std::size_t len) {
+    if (!valid() || len == 0) {
+        return -1;
+    }
+    return detail::rawSend(handle_, data, std::min(len, static_cast<std::size_t>(INT_MAX)));
+}
+
 int Socket::recvSome(void* buffer, std::size_t capacity) {
     if (!valid()) {
         return -1;