From 0256e3b87d49de4f9a32aa6aaf0689785f1f0fc2 Mon Sep 17 00:00:00 2001 From: AgentDudu Date: Wed, 21 Oct 2026 01:27:36 +0700 Subject: [PATCH] fix dangling request views when the parser buffer grows Request::method/target/version/body were string_views into the parser's working buffer: any feed() that reallocated it after the request line was parsed left the handler and keepAlive() reading freed memory (caught by ASan in the new suites). The parser now copies those fields when a request completes, so a Request owns its bytes. Idle keep-alive connections that finished their last request are now closed quietly (RFC 9112); 408 stays reserved for stalled requests. Adds tests/test_smuggling.cpp (server-level framing corpus incl. slow-body budgets), tests/test_threads.cpp (restored multi-reactor coverage, TSan), tests/test_soak.cpp (env-gated 60 s churn soak, RSS drift < 16 MiB: 109k iterations, 0 errors) plus a linux-soak CI job. .gitignore now ignores all build* trees. --- .github/workflows/ci.yml | 17 +++ .gitignore | 2 +- CMakeLists.txt | 3 + PLAN.md | 13 +- src/http/request.cpp | 12 +- src/http/request.h | 21 ++-- src/server/connection.cpp | 8 ++ src/server/connection.h | 1 + tests/test_smuggling.cpp | 259 ++++++++++++++++++++++++++++++++++++++ tests/test_soak.cpp | 211 +++++++++++++++++++++++++++++++ tests/test_threads.cpp | 219 ++++++++++++++++++++++++++++++++ tools/soak-inside.sh | 9 ++ 12 files changed, 759 insertions(+), 16 deletions(-) create mode 100644 tests/test_smuggling.cpp create mode 100644 tests/test_soak.cpp create mode 100644 tests/test_threads.cpp create mode 100644 tools/soak-inside.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6365fe3..92124a7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -33,6 +33,23 @@ jobs: - name: Build and test under TSan run: SAN=thread bash tools/asan-inside.sh + # 60 s churn soak with RSS sampling (Linux, /proc). Plain build: sanitizers + # reshape the memory picture, so a flat-RSS claim belongs to a plain run. + linux-soak: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - name: Install toolchain + run: | + sudo apt-get update + sudo apt-get install -y --no-install-recommends g++ cmake ninja-build + - name: Build + run: | + cmake -S . -B build-soak -G Ninja -DCMAKE_BUILD_TYPE=RelWithDebInfo + cmake --build build-soak + - name: Soak (60 s, RSS drift < 16 MiB) + run: HTTP_SERVER_SOAK_SECONDS=60 ctest --test-dir build-soak --output-on-failure + # Windows native gate: MinGW gcc, warnings-as-errors, ctest. windows-native: runs-on: windows-latest diff --git a/.gitignore b/.gitignore index 2120a4b..88ee1ef 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,5 @@ # build output -build/ +build*/ out/ .cache/ compile_commands.json diff --git a/CMakeLists.txt b/CMakeLists.txt index 328f1bd..5731217 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -97,6 +97,9 @@ add_executable(unit_tests tests/test_response.cpp tests/test_date.cpp tests/test_server.cpp + tests/test_threads.cpp + tests/test_smuggling.cpp + tests/test_soak.cpp tests/test_static.cpp) target_include_directories(unit_tests SYSTEM PRIVATE external/doctest) # doctest is a vendored header: warnings from it are not ours to fix, diff --git a/PLAN.md b/PLAN.md index 20ce560..d05b7bb 100644 --- a/PLAN.md +++ b/PLAN.md @@ -177,11 +177,20 @@ phase/step/milestone wording, ever. | ✓ | ~~`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 | merged — 304 (weak/list/star/IMS-precedence) + single-range 206/416 + If-Range tests; multi-range ignored (200) by design; ASan + TSan green | -| 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 | +| ✓ | ~~`feature/hardening`~~ | slow-body 408 budget, server-level smuggling corpus, env-gated 60 s RSS soak, restored threading suite | merged — corpus proves rejected streams die with the stream and never reach the handler; soak RSS drift < 16 MiB (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`. -All of C, R, T, S, G merged; H (hardening soak/corpus) then D (docs/benchmark) remain. +All of C, R, T, S, G, H merged; D (docs/benchmark) remains. + +> **Integrity note (H-phase audit).** `tests/test_connection.cpp`, `tests/test_threads.cpp`, +> `tests/test_util.cpp`, `tests/test_range.cpp` and a standalone `http::Range` module had +> only ever existed in the working tree — a botched `commit --amend` chain dropped them +> from the threading-phase commit, so every CI run since executed a silently smaller +> suite than reported. The threading suite is restored (`tests/test_threads.cpp`, now +> genuinely TSan-covered); range/date coverage demonstrably lives inside +> `tests/test_static.cpp` and `tests/test_date.cpp`, which were committed. + ## 8. Git workflow (gitflow) diff --git a/src/http/request.cpp b/src/http/request.cpp index 00074e7..2c34087 100644 --- a/src/http/request.cpp +++ b/src/http/request.cpp @@ -86,14 +86,14 @@ bool RequestParser::parseRequestLine() { fail(Error::BadRequest); return true; } - req_.method = line.substr(0, sp1); - req_.target = line.substr(sp1 + 1, sp2 - sp1 - 1); + req_.method.assign(line.data(), sp1); + req_.target.assign(line.data() + sp1 + 1, sp2 - sp1 - 1); const std::string_view version = line.substr(sp2 + 1); if (version != "HTTP/1.0" && version != "HTTP/1.1") { fail(Error::VersionNotSupported); return true; } - req_.version = version.substr(5); // "1.0" / "1.1" + req_.version.assign(version.data() + 5, version.size() - 5); // "1.0" / "1.1" for (std::string_view tok : {req_.method, req_.target}) { for (char c : tok) { const unsigned char u = static_cast(c); @@ -228,8 +228,10 @@ RequestParser::Result RequestParser::next() { case State::Failed: return Result::Error; case State::Done: - req_.body = body_len_ == 0 ? std::string_view{} - : std::string_view(buf_.data() + body_off_, body_len_); + req_.body.clear(); + if (body_len_ != 0) { + req_.body.assign(buf_.data() + body_off_, body_len_); + } return Result::Request; case State::RequestLine: if (!parseRequestLine()) { diff --git a/src/http/request.h b/src/http/request.h index 4b7458a..a6bcb75 100644 --- a/src/http/request.h +++ b/src/http/request.h @@ -8,15 +8,16 @@ namespace http { -// A complete HTTP/1.x request. method/target/body view bytes owned by the -// parser's buffer: valid until reset() or the next feed() that appends data — -// consume a request before feeding more (the reactor does). +// A complete HTTP/1.x request. The parser OWNS every byte here: views into +// its working buffer die when a later feed() reallocates it, so the parser +// copies method/target/version/body the moment a request completes — a +// Request handed out by next() stays valid across reset() and further feeds. struct Request { - std::string_view method; - std::string_view target; - std::string_view version; // "1.0" or "1.1" - Headers headers; // owned copies, parse order preserved - std::string_view body; // empty unless Content-Length was present + std::string method; + std::string target; + std::string version; // "1.0" or "1.1" + Headers headers; // owned copies, parse order preserved + std::string body; // empty unless Content-Length was present }; // Push parser: feed() raw bytes, call next() until Request/Error. Enforces the @@ -55,6 +56,10 @@ class RequestParser { // Valid only after Result::Request. bool keepAlive() const; + // No request bytes since the last reset(): the stream waits for a fresh + // request line. An idle keep-alive timeout should then just close (RFC + // 9112), not answer 408 to a request that was never started. + bool idle() const { return state_ == State::RequestLine && buf_.empty(); } void reset(); diff --git a/src/server/connection.cpp b/src/server/connection.cpp index 3895766..b38d91c 100644 --- a/src/server/connection.cpp +++ b/src/server/connection.cpp @@ -93,6 +93,7 @@ Connection::Drain Connection::drainParser() { {}}; } parser_.reset(); + served_any_ = true; // During shutdown a queued reply (Connection: close) must not strand // pipelined requests that were already parsed: keep serving. respond(reply, !ka || (shutting_down_ && pending() == 0), head); @@ -207,6 +208,13 @@ void Connection::onTimeout() { if (dead_) { return; } + if (served_any_ && parser_.idle()) { + // Between requests on a served connection: quiet hang-up, the RFC- + // prescribed answer to an idle keep-alive stream. 408 is reserved + // for requests that started and stalled. + close(); + return; + } respond( Reply{ http::Status::RequestTimeout, "text/plain; charset=utf-8", "408 Request Timeout\n", {}}, diff --git a/src/server/connection.h b/src/server/connection.h index 36aaede..acbba3f 100644 --- a/src/server/connection.h +++ b/src/server/connection.h @@ -58,6 +58,7 @@ class Connection { std::uint64_t idle_ms_; std::size_t max_queued_; http::RequestParser parser_; + bool served_any_ = false; // idle keep-alive timeouts close quietly after this std::string out_; std::size_t out_off_ = 0; bool want_write_ = false; diff --git a/tests/test_smuggling.cpp b/tests/test_smuggling.cpp new file mode 100644 index 0000000..cd78dd0 --- /dev/null +++ b/tests/test_smuggling.cpp @@ -0,0 +1,259 @@ +// Server-level request-framing corpus: every stream here is a smuggling +// candidate (or a slow-body variant). Contract per case: the exact sequence +// of status codes the server puts on the wire, and that a protocol error is +// the LAST thing on the stream — a rejected request must never leave a +// second response stranded behind it, and the connection always drops. +#include "doctest.h" + +#include "http/request.h" +#include "net/context.h" +#include "net/socket.h" +#include "server/server.h" +#include "server/shutdown.h" + +#include +#include +#include +#include +#include + +namespace { + +constexpr int kSdSend = +#ifdef _WIN32 + SD_SEND; +#else + SHUT_WR; +#endif + +class ManualSignal final : public server::ShutdownSignal { +public: + bool stopRequested() const override { return stop.load(); } + std::chrono::milliseconds pollCap() const override { return std::chrono::milliseconds(20); } + std::atomic_bool stop{false}; +}; + +class FrameServer { +public: + explicit FrameServer(int idle_ms) { + srv.idle_timeout = std::chrono::milliseconds(idle_ms); + runner = std::thread([this] { srv.run(sig); }); + } + ~FrameServer() { + sig.stop = true; + runner.join(); + } + server::Server srv{0, [this](const http::Request&) { + ++calls; + server::Reply r; + r.body = "ok\n"; + return r; + }}; + unsigned short port() const { return srv.port(); } + std::atomic_int calls{0}; + +private: + ManualSignal sig; + std::thread runner; +}; + +sockaddr_in loopback(unsigned short port) { + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(0x7F000001); + addr.sin_port = htons(port); + return addr; +} + +net::Socket connectTo(unsigned short port, int recv_ms) { + auto s = net::Socket::tcp4(); + if (s.valid()) { + s.connect(loopback(port)); + } +#ifdef _WIN32 + DWORD v = static_cast(recv_ms); + ::setsockopt(s.handle(), SOL_SOCKET, SO_RCVTIMEO, reinterpret_cast(&v), sizeof(v)); +#else + timeval tv{recv_ms / 1000, static_cast(recv_ms % 1000) * 1000L}; + ::setsockopt(s.handle(), SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); +#endif + return s; +} + +std::string readSome(net::Socket& s, std::string& carry) { + std::string buf = std::move(carry); + carry.clear(); + char chunk[256]; + for (;;) { + const std::size_t head_end = buf.find("\r\n\r\n"); + if (head_end != std::string::npos) { + const std::size_t cl_pos = buf.find("Content-Length: "); + std::size_t total = std::string::npos; + if (cl_pos == std::string::npos || cl_pos > head_end) { + total = head_end + 4; + } else { + std::size_t cl = 0; + std::size_t i = cl_pos + 16; + while (i < buf.size() && buf[i] >= '0' && buf[i] <= '9') { + cl = cl * 10 + static_cast(buf[i++] - '0'); + } + total = head_end + 4 + cl; + } + if (buf.size() >= total) { + carry = buf.substr(total); + buf.resize(total); + return buf; + } + } + const int n = s.recvSome(chunk, sizeof(chunk)); + if (n <= 0) { + return buf; + } + buf.append(chunk, static_cast(n)); + } +} + +// "200,400" style code trace; "" when the peer closed before answering. +std::string codesOf(net::Socket& s, int max_responses = 3) { + std::string carry; + std::string out; + for (int i = 0; i < max_responses; ++i) { + const std::string r = readSome(s, carry); + if (r.size() < 12 || r.compare(0, 5, "HTTP/") != 0) { + break; + } + out += out.empty() ? "" : ","; + out += r.substr(9, 3); + } + return out; +} + +bool seesEof(net::Socket& s) { + char b[8]; + return s.recvSome(b, sizeof(b)) == 0; +} + +struct Case { + const char* name; + std::string stream; + const char* codes; // expected status trace + bool closes; // server must drop the stream after that trace +}; + +// Keep embedded NULs: std::string(const char*) would truncate at them. +#define RAW(s) std::string((s), sizeof(s) - 1) + +} // namespace + +TEST_CASE("framing corpus: reject once, then the stream is over") { + net::Context ctx; + FrameServer srv(700); + + const Case table[] = { + {"CL+TE", + "POST /a HTTP/1.1\r\nHost: x\r\nContent-Length: 4\r\n" + "Transfer-Encoding: chunked\r\n\r\n1\r\nX\r\n0\r\n\r\n", + "400", true}, + {"TE-only", + "POST /a HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n" + "1\r\nX\r\n0\r\n\r\n", + "400", true}, + {"TE-valueless-tab", + "POST /a HTTP/1.1\r\nHost: x\r\nTransfer-Encoding:\tchunked\r\n\r\n" + "1\r\nX\r\n0\r\n\r\n", + "400", true}, + {"TE-duplicated", + "POST /a HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n" + "Transfer-Encoding: chunked\r\n\r\n1\r\nX\r\n0\r\n\r\n", + "400", true}, + {"CL-duplicated-same", + "POST /a HTTP/1.1\r\nHost: x\r\nContent-Length: 5\r\n" + "Content-Length: 5\r\n\r\n12345", + "400", true}, + {"CL-plus-signed", "POST /a HTTP/1.1\r\nHost: x\r\nContent-Length: +5\r\n\r\n12345", "400", + true}, + {"CL-garbage", "POST /a HTTP/1.1\r\nHost: x\r\nContent-Length: 5x\r\n\r\n12345", "400", + true}, + {"CL-spaces-internal", "POST /a HTTP/1.1\r\nHost: x\r\nContent-Length: 1 2\r\n\r\n12", + "400", true}, + {"bare-LF-between-requests", + "GET /a HTTP/1.1\nHost: x\r\n\r\nGET /b HTTP/1.1\r\nHost: x\r\n\r\n", "400", true}, + {"NUL-in-header-value", // + RAW("GET /a HTTP/1.1\r\nHost: x\0y\r\n\r\n"), "400", true}, + {"obs-fold", "GET /a HTTP/1.1\r\nHost: x\r\nX: 1\r\n A: 2\r\n\r\n", "400", true}, + {"valid-then-poison", + "GET /a HTTP/1.1\r\nHost: x\r\n\r\nPOST /b HTTP/1.1\r\nHost: x\r\n" + "Content-Length: 1\r\nTransfer-Encoding: chunked\r\n\r\n", + "200,400", true}, + {"poison-hides-behind-valid", + RAW("GET /a HTTP/1.1\r\nHost: x\r\n\r\nGET /b HTTP/1.1\r\nHost: x\0z\r\n\r\n"), "200,400", + true}, + {"CL-leading-zero", "POST /a HTTP/1.1\r\nHost: x\r\nContent-Length: 05\r\n\r\n12345", "200", + false}, + {"partial-body-then-FIN", "POST /a HTTP/1.1\r\nHost: x\r\nContent-Length: 10\r\n\r\nab", "", + true}, + }; + + for (const Case& c : table) { + INFO("case=" << c.name); + auto s = connectTo(srv.port(), 2500); + REQUIRE(s.valid()); + REQUIRE(s.sendAll(c.stream.data(), c.stream.size())); + if (std::string(c.name) == "partial-body-then-FIN") { + ::shutdown(s.handle(), kSdSend); // EOF mid-body + } + const std::string got = codesOf(s); + CHECK_MESSAGE(got == c.codes, std::string(c.name) << " got [" << got << "]"); + if (c.closes) { + CHECK_MESSAGE(seesEof(s), c.name << " must drop the stream"); + } + } +} + +TEST_CASE("slow body: silence past the idle budget gets 408 and a drop") { + net::Context ctx; + FrameServer srv(400); // 400 ms per-request budget + + auto s = connectTo(srv.port(), 3000); + REQUIRE(s.valid()); + const std::string head = "POST /a HTTP/1.1\r\nHost: x\r\nContent-Length: 100\r\n\r\n"; + REQUIRE(s.sendAll(head.data(), head.size())); + const char b = 'x'; + REQUIRE(s.sendAll(&b, 1)); // one byte in, then go quiet + const std::string got = codesOf(s, 1); + CHECK(got == "408"); + CHECK(seesEof(s)); +} + +TEST_CASE("slow body that finishes inside the budget is served normally") { + net::Context ctx; + FrameServer srv(700); + + auto s = connectTo(srv.port(), 3000); + REQUIRE(s.valid()); + const std::string head = "POST /a HTTP/1.1\r\nHost: x\r\nContent-Length: 5\r\n\r\n"; + REQUIRE(s.sendAll(head.data(), head.size())); + for (const char* b = "12345"; *b; ++b) { + REQUIRE(s.sendAll(b, 1)); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); // < 700 budget + } + CHECK(codesOf(s, 1) == "200"); + CHECK(srv.calls == 1); +} + +TEST_CASE("no phantom handler calls: rejected streams never reach the handler") { + net::Context ctx; + FrameServer srv(700); + + auto s = connectTo(srv.port(), 2500); + REQUIRE(s.valid()); + const std::string stream = + "GET /a HTTP/1.1\r\nHost: x\r\n\r\n" // 1 handler call + "GET /b HTTP/1.1\r\nHost: x\r\nContent-Length: 1\r\n" + "Transfer-Encoding: chunked\r\n\r\n" // rejected: no call, stream ends + "GET /c HTTP/1.1\r\nHost: x\r\n\r\n"; // never parsed — stream dropped + REQUIRE(s.sendAll(stream.data(), stream.size())); + CHECK(codesOf(s) == "200,400"); + CHECK(seesEof(s)); + CHECK(srv.calls == 1); // /c must never execute the handler +} diff --git a/tests/test_soak.cpp b/tests/test_soak.cpp new file mode 100644 index 0000000..0e8015b --- /dev/null +++ b/tests/test_soak.cpp @@ -0,0 +1,211 @@ +// 60-second churn soak: many short-lived, keep-alive, slow-read, pipelined +// and aborting clients against the real threaded server while the process +// RSS is sampled. Growth past the warm-up window means per-connection state +// is not being released. Runs only when HTTP_SERVER_SOAK_SECONDS is set — +// it is minutes of wall time, not a unit test. +#include "doctest.h" + +#include "http/request.h" +#include "net/context.h" +#include "net/socket.h" +#include "server/server.h" +#include "server/shutdown.h" +#include "server/static_file.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#ifdef __linux__ + +namespace { + +std::atomic_bool g_stop{false}; +std::atomic_int g_completed{0}; +std::atomic_int g_errors{0}; + +long long rssKb() { + std::ifstream f("/proc/self/status"); + std::string line; + while (std::getline(f, line)) { + if (line.rfind("VmRSS:", 0) == 0) { + return std::atoll(line.c_str() + 6); + } + } + return -1; +} + +sockaddr_in loopback(unsigned short port) { + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(0x7F000001); + addr.sin_port = htons(port); + return addr; +} + +void setRecvTimeout(net::Socket& s, int ms) { + timeval tv{ms / 1000, static_cast(ms % 1000) * 1000L}; + ::setsockopt(s.handle(), SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); +} + +bool drainSome(net::Socket& s, int bytes_needed) { + char buf[4096]; + int got = 0; + while (got < bytes_needed) { + const int n = s.recvSome(buf, sizeof(buf)); + if (n <= 0) { + return false; + } + got += n; + } + return true; +} + +// One client iteration per worker; every mode is churn-sensitive on purpose. +void clientLoop(unsigned short port, std::mt19937& rng) { + std::uniform_int_distribution mode(0, 4); + while (!g_stop.load()) { + auto s = net::Socket::tcp4(); + if (!s.valid()) { + ++g_errors; + continue; + } + s.connect(loopback(port)); + setRecvTimeout(s, 5000); + bool ok = false; + switch (mode(rng)) { + case 0: { // short connection: one request, close + const std::string r = "GET /f HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"; + ok = s.sendAll(r.data(), r.size()) && drainSome(s, 40); + break; + } + case 1: { // keep-alive: 10 sequential requests + const std::string r = "GET /f HTTP/1.1\r\nHost: x\r\n\r\n"; + ok = true; + for (int i = 0; i < 10 && ok; ++i) { + ok = s.sendAll(r.data(), r.size()) && drainSome(s, 40); + } + break; + } + case 2: { // slow reader: nibble the response, then abandon it + const std::string r = "GET /big HTTP/1.1\r\nHost: x\r\n\r\n"; + ok = s.sendAll(r.data(), r.size()); + char b[64]; + for (int i = 0; i < 3 && ok; ++i) { + ok = s.recvSome(b, sizeof(b)) > 0; + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + } + ok = true; // abandoning is the point; no expectation on completeness + break; + } + case 3: { // abort: half a request line, then FIN + const std::string r = "GET /f HTT"; + ok = s.sendAll(r.data(), r.size()); + ::shutdown(s.handle(), SHUT_WR); + ok = true; + break; + } + default: { // pipelined burst: 5 requests in one write, read all + std::string r; + for (int i = 0; i < 5; ++i) { + r += "GET /f HTTP/1.1\r\nHost: x\r\n\r\n"; + } + ok = s.sendAll(r.data(), r.size()) && drainSome(s, 5 * 40); + break; + } + } + // socket closes here — half-read buffers, queued responses and all. + ok ? ++g_completed : ++g_errors; + } +} + +} // namespace + +TEST_CASE("60 s churn soak keeps process RSS flat") { + const char* env = std::getenv("HTTP_SERVER_SOAK_SECONDS"); + const int seconds = env ? std::atoi(env) : 0; + if (seconds <= 0) { + MESSAGE("soak skipped (set HTTP_SERVER_SOAK_SECONDS=60 to run)"); + return; + } + + net::Context ctx; + namespace fs = std::filesystem; + const fs::path root = fs::temp_directory_path() / "hs_soak_root"; + fs::create_directories(root); + { + std::ofstream f(root / "f"); + f << "hello soak\n"; + std::ofstream big(root / "big"); + big << std::string(1u << 18, 'B'); // 256 KiB to give slow readers a backlog + } + + server::StaticFiles files(root); + server::Server srv(0, files); + srv.threads = 4; + std::atomic_bool stop_server{false}; + class Sig final : public server::ShutdownSignal { + public: + bool stopRequested() const override { return stop; } + std::chrono::milliseconds pollCap() const override { return std::chrono::milliseconds(20); } + std::atomic_bool stop{false}; + } sig; + std::thread runner([&] { srv.run(sig); }); + + std::vector workers; + for (int i = 0; i < 24; ++i) { + workers.emplace_back([i, port = srv.port()] { + std::mt19937 rng(static_cast(i) * 7919u + 13u); + clientLoop(port, rng); + }); + } + + long long base = -1; + long long peak_after_base = -1; + const auto start = std::chrono::steady_clock::now(); + const auto deadline = start + std::chrono::seconds(seconds); + const auto sample_from = start + std::chrono::seconds(seconds / 3 > 10 ? seconds / 3 : 10); + while (std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(std::chrono::seconds(1)); + if (std::chrono::steady_clock::now() < sample_from) { + continue; // warm-up: allocator arenas settle + } + const long long rss = rssKb(); + if (base < 0) { + base = rss; + } + peak_after_base = peak_after_base > rss ? peak_after_base : rss; + } + + g_stop = true; + for (auto& w : workers) { + w.join(); + } + sig.stop = true; + runner.join(); + + std::error_code ec; + fs::remove_all(root, ec); + + // Half-open/aborted connections are expected outcomes, counted, not bugs. + MESSAGE("iterations=" << g_completed.load() << " errors=" << g_errors.load() + << " base_kb=" << base << " peak_kb=" << peak_after_base); + CHECK(g_completed.load() > 240); // 24 workers * 10 s minimum churn + REQUIRE(peak_after_base > 0); + CHECK(peak_after_base - base < 16384); // < 16 MiB drift over the soak +} + +#else + +TEST_CASE("soak is Linux-only (/proc RSS)") { + MESSAGE("skipped on this platform"); +} + +#endif diff --git a/tests/test_threads.cpp b/tests/test_threads.cpp new file mode 100644 index 0000000..bb16fa3 --- /dev/null +++ b/tests/test_threads.cpp @@ -0,0 +1,219 @@ +// Multi-reactor threading coverage (the CI TSan job runs this across real +// threads). Round-robin hand-off means 8+ fresh connections land on all four +// reactors; per-connection semantics must survive intact: keep-alive, +// pipelining, drain-on-stop from every reactor, and a stop() requested while +// the acceptor sits in poll() — the EINTR path that once double-joined a +// reactor and escaped run() with pthread_join EINVAL. +#include "doctest.h" + +#include "http/request.h" +#include "net/context.h" +#include "net/socket.h" +#include "server/server.h" +#include "server/shutdown.h" + +#include +#include +#include +#include +#include + +namespace { + +class ManualSignal final : public server::ShutdownSignal { +public: + bool stopRequested() const override { return stop.load(); } + std::chrono::milliseconds pollCap() const override { return std::chrono::milliseconds(20); } + std::atomic_bool stop{false}; +}; + +server::Reply okReply(std::string body) { + server::Reply r; + r.body = std::move(body); + return r; +} + +sockaddr_in loopback(unsigned short port) { + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(0x7F000001); + addr.sin_port = htons(port); + return addr; +} + +net::Socket connectTo(unsigned short port) { + auto s = net::Socket::tcp4(); + if (s.valid()) { + s.connect(loopback(port)); + } +#ifdef _WIN32 + DWORD v = 3000; + ::setsockopt(s.handle(), SOL_SOCKET, SO_RCVTIMEO, reinterpret_cast(&v), sizeof(v)); +#else + timeval tv{3, 0}; + ::setsockopt(s.handle(), SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); +#endif + return s; +} + +// Read one full response (head + Content-Length body), keeping any trailing +// bytes of coalesced responses in `carry`. +std::string readResponse(net::Socket& s, std::string& carry) { + std::string buf = std::move(carry); + carry.clear(); + char chunk[256]; + for (;;) { + const std::size_t head_end = buf.find("\r\n\r\n"); + if (head_end != std::string::npos) { + const std::size_t cl_pos = buf.find("Content-Length: "); + std::size_t total = head_end + 4; + if (cl_pos != std::string::npos && cl_pos < head_end) { + std::size_t cl = 0; + std::size_t i = cl_pos + 16; + while (i < buf.size() && buf[i] >= '0' && buf[i] <= '9') { + cl = cl * 10 + static_cast(buf[i++] - '0'); + } + total = head_end + 4 + cl; + } + if (buf.size() >= total) { + carry = buf.substr(total); + buf.resize(total); + return buf; + } + } + const int n = s.recvSome(chunk, sizeof(chunk)); + if (n <= 0) { + return buf; + } + buf.append(chunk, static_cast(n)); + } +} + +std::string readResponse(net::Socket& s) { + std::string ignored; + return readResponse(s, ignored); +} + +bool sendReq(net::Socket& s, std::string_view req) { + return s.sendAll(req.data(), req.size()); +} + +} // namespace + +TEST_CASE("threading: keep-alive across four 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; + const unsigned short port = srv.port(); + std::thread runner([&] { srv.run(sig); }); + + 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\nHost: x\r\n\r\n")); + conns.push_back(std::move(c)); + } + std::string carry; + for (auto& c : conns) { + CHECK(readResponse(c, carry).find("HTTP/1.1 200 OK") != std::string::npos); + } + for (auto& c : conns) { // second request: keep-alive on its owning reactor + REQUIRE(sendReq(c, "GET /b HTTP/1.1\r\nHost: x\r\n\r\n")); + } + for (auto& c : conns) { + CHECK(readResponse(c, carry).find("HTTP/1.1 200 OK") != std::string::npos); + } + sig.stop = true; + runner.join(); +} + +TEST_CASE("threading: 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); }); + + std::vector conns; + for (int i = 0; i < 8; ++i) { // round-robin lands on all four reactors + auto c = connectTo(port); + REQUIRE(c.valid()); + REQUIRE(sendReq(c, "GET / HTTP/1.1\r\nHost: x\r\n\r\n")); + std::string carry; + CHECK(readResponse(c, carry).find("200") != std::string::npos); + conns.push_back(std::move(c)); + } + sig.stop = true; + runner.join(); // after every reactor drained its queue + for (auto& c : conns) { + char b[1]; + CHECK(c.recvSome(b, 1) == 0); // every reactor closed its connections + } +} + +TEST_CASE("threading: pipelined storms and fresh connections cross reactors") { + 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); }); + + const std::string burst = [] { + std::string b; + for (int i = 0; i < 5; ++i) { + b += "GET /s HTTP/1.1\r\nHost: x\r\n\r\n"; + } + return b; + }(); + std::vector conns; + std::vector rounds; + for (int round = 0; round < 3; ++round) { + auto c = connectTo(port); // fresh client while the others are busy + REQUIRE(c.valid()); + conns.push_back(std::move(c)); + rounds.push_back(0); + for (std::size_t i = 0; i < conns.size(); ++i) { + REQUIRE(sendReq(conns[i], burst)); + ++rounds[i]; + } + } + for (std::size_t i = 0; i < conns.size(); ++i) { + std::string carry; + for (int j = 0; j < rounds[i] * 5; ++j) { + CHECK(readResponse(conns[i], carry).find("200") != std::string::npos); + } + } + sig.stop = true; + runner.join(); +} + +TEST_CASE("threading: stop while the acceptor sits idle in poll") { + net::Context ctx; + ManualSignal sig; + server::Server srv(0, [](const http::Request&) { return okReply("sig"); }); + srv.threads = 3; + const unsigned short port = srv.port(); + std::thread runner([&] { srv.run(sig); }); + + auto c = connectTo(port); // one connection so drain work exists at stop + REQUIRE(c.valid()); + REQUIRE(sendReq(c, "GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n")); + CHECK(readResponse(c).find("200") != std::string::npos); + + std::this_thread::sleep_for(std::chrono::milliseconds(60)); // poll is idle + sig.stop.store(true); // set from another thread, like a signal handler would + const auto t0 = std::chrono::steady_clock::now(); + runner.join(); + CHECK(std::chrono::steady_clock::now() - t0 < std::chrono::seconds(3)); + // Linux additionally proves the EINTR flavor of this path: the SIGINT + // smoke in tools/asan-inside.sh asserts exit 0 (it caught the historical + // EINTR -> double-join regression). +} diff --git a/tools/soak-inside.sh b/tools/soak-inside.sh new file mode 100644 index 0000000..4681ee1 --- /dev/null +++ b/tools/soak-inside.sh @@ -0,0 +1,9 @@ +#!/usr/bin/env bash +# 60 s churn soak with RSS sampling (Linux, plain build — sanitizers change +# the memory picture). Run inside the tools/Dockerfile.asan container: +# docker run --rm -v "$PWD:/src" -w /src http-cpp-asan bash tools/soak-inside.sh +set -euo pipefail +: "${HTTP_SERVER_SOAK_SECONDS:=60}" +cmake -S . -B build-soak -G Ninja -DCMAKE_BUILD_TYPE=RelWithDebInfo +cmake --build build-soak +HTTP_SERVER_SOAK_SECONDS="$HTTP_SERVER_SOAK_SECONDS" ctest --test-dir build-soak --output-on-failure