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