Multi-reactor threading with socket hand-off inbox

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)

CMakePresets.json

@@ -1,56 +1,130 @@
 {
   "version": 3,
-  "cmakeMinimumRequired": { "major": 3, "minor": 25, "patch": 0 },
+  "cmakeMinimumRequired": {
+    "major": 3,
+    "minor": 25
+  },
   "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
+      }
     }
   ]
 }

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 |

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,

src/main.cpp

@@ -13,16 +13,25 @@ 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<unsigned>(std::atoi(argv[3]));
+        if (threads > 128) {
+            std::fprintf(stderr, "usage: threads must be 0-128\n");
+            return 2;
+        }
+    }
 
     try {
         net::Context ctx;
         app::ShutdownSignals signals;
         const std::string root = argc > 2 ? argv[2] : ".";
         server::StaticFiles files(root);
         server::Server server(static_cast<std::uint16_t>(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);

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

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);
 }

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<sockaddr*>(&a), sizeof(a)) != SOCKET_ERROR &&
+              ::listen(l, 1) != SOCKET_ERROR &&
+              ::getsockname(l, reinterpret_cast<sockaddr*>(&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<sockaddr*>(&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);
 }

src/net/wakeup.cpp

@@ -0,0 +1,34 @@
+#include "net/wakeup.h"
+
+#include "net/detail_raw.h"
+
+#include <cstdint>
+
+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<int>(sizeof(buf))) {
+    }
+}
+
+} // namespace net

src/net/wakeup.h

@@ -0,0 +1,29 @@
+#pragma once
+
+#include "net/socket.h"
+
+#include <utility>
+
+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