diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4497633..6365fe3 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/CMakeLists.txt b/CMakeLists.txt index 20f9ad2..00687d7 100644 --- a/CMakeLists.txt +++ b/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) diff --git a/PLAN.md b/PLAN.md index 7914e02..cdf2d91 100644 --- a/PLAN.md +++ b/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) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 9e96feb..74fb520 100644 --- a/docs/ARCHITECTURE.md +++ b/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. diff --git a/src/app/signals.h b/src/app/signals.h new file mode 100644 index 0000000..e5382bb --- /dev/null +++ b/src/app/signals.h @@ -0,0 +1,32 @@ +#pragma once + +#include "server/shutdown.h" + +#include + +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_; +}; + +} // namespace app diff --git a/src/app/signals_posix.cpp b/src/app/signals_posix.cpp new file mode 100644 index 0000000..b6b15d4 --- /dev/null +++ b/src/app/signals_posix.cpp @@ -0,0 +1,82 @@ +#include "app/signals.h" + +#include "net/poll.h" + +#include +#include +#include // for the pollfd revents bits (net/poll.h pulls it too) +#include + +#include + +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()) { + 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 diff --git a/src/app/signals_win.cpp b/src/app/signals_win.cpp new file mode 100644 index 0000000..1449889 --- /dev/null +++ b/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 + +#include + +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()) { + 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 diff --git a/src/main.cpp b/src/main.cpp index 46ea6c7..f0e7cf0 100644 --- a/src/main.cpp +++ b/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 +#include +#include + +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(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; +} diff --git a/src/net/poll.h b/src/net/poll.h index 59aba5e..7ba7d6d 100644 --- a/src/net/poll.h +++ b/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 { diff --git a/src/net/socket.h b/src/net/socket.h index a92259a..56cc062 100644 --- a/src/net/socket.h +++ b/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); diff --git a/src/net/socket_common.cpp b/src/net/socket_common.cpp index e3fd4ae..098ce57 100644 --- a/src/net/socket_common.cpp +++ b/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(INT_MAX))); +} + int Socket::recvSome(void* buffer, std::size_t capacity) { if (!valid()) { return -1; diff --git a/src/server/connection.cpp b/src/server/connection.cpp new file mode 100644 index 0000000..01f486b --- /dev/null +++ b/src/server/connection.cpp @@ -0,0 +1,156 @@ +#include "server/connection.h" + +#include "http/response.h" +#include "http/status.h" +#include "net/error.h" + +namespace server { + +Connection::Connection(net::Socket socket, const Handler& handler, + std::chrono::milliseconds idle) + : sock_(std::move(socket)), handler_(&handler), + idle_ms_(static_cast(idle.count() < 0 ? 0 : idle.count())) { + deadline_ms_ = util::monotonicMs() + idle_ms_; +} + +void Connection::respond(http::Status status, std::string_view body, bool closeAfter) { + http::Response resp(status); + resp.setHeader("Content-Type", "text/plain; charset=utf-8"); + if (closeAfter) { + resp.setHeader("Connection", "close"); + } + resp.setBody(std::string(body)); + out_ += resp.serialize(); + want_write_ = out_off_ < out_.size(); + close_after_write_ = close_after_write_ || closeAfter; +} + +bool Connection::drainParser() { + for (;;) { + const auto res = parser_.next(); + if (res == http::RequestParser::Result::NeedMore) { + return true; + } + if (res == http::RequestParser::Result::Error) { + using Err = http::RequestParser::Error; + http::Status st; + switch (parser_.error()) { + case Err::HeadTooLarge: + case Err::TooManyHeaders: + st = http::Status::HeaderFieldsTooLarge; + break; + case Err::BodyTooLarge: + st = http::Status::PayloadTooLarge; + break; + case Err::VersionNotSupported: + st = http::Status::HTTPVersionNotSupported; + break; + default: + st = http::Status::BadRequest; + break; + } + // Protocol violation: answer once (best effort), then drop the + // connection — smuggling candidates never share a stream. + respond(st, std::string(http::reasonPhrase(st)) + "\n", true); + flush(); + close(); + return false; + } + const bool ka = parser_.keepAlive(); + const std::string body = (*handler_)(parser_.request()); + parser_.reset(); + respond(http::Status::OK, body, !ka || shutting_down_); + deadline_ms_ = util::monotonicMs() + idle_ms_; + if (close_after_write_) { + return false; + } + } +} + +void Connection::onReadable() { + if (dead_) { + return; + } + bool more = drainParser(); + while (more) { + if (close_after_write_ || dead_) { + more = false; + break; + } + const int n = sock_.recvSome(rbuf_, sizeof(rbuf_)); + if (n > 0) { + parser_.feed(std::string_view(rbuf_, static_cast(n))); + more = drainParser(); + continue; + } + if (n < 0 && net::lastError().kind == net::ErrorKind::WouldBlock) { + more = false; + break; + } + close(); // EOF or error + return; + } + if (shutting_down_ && !want_write_) { + close(); + return; + } + if (want_write_) { + flush(); // optimistic: often fits the kernel without a POLLOUT round-trip + } + if (!dead_ && close_after_write_ && out_off_ == out_.size()) { + close(); + } +} + +void Connection::flush() { + while (!dead_ && out_off_ < out_.size()) { + const int n = sock_.sendSome(out_.data() + out_off_, out_.size() - out_off_); + if (n > 0) { + out_off_ += static_cast(n); + continue; + } + if (net::lastError().kind == net::ErrorKind::WouldBlock) { + want_write_ = true; + return; + } + close(); + return; + } + if (!dead_ && out_off_ == out_.size()) { + out_.clear(); + out_off_ = 0; + want_write_ = false; + } +} + +void Connection::onWritable() { + if (dead_) { + return; + } + flush(); + if (!dead_ && close_after_write_ && out_off_ == out_.size()) { + close(); + } +} + +void Connection::onTimeout() { + if (dead_) { + return; + } + respond(http::Status::RequestTimeout, "408 Request Timeout\n", true); + flush(); + close(); +} + +void Connection::beginShutdown() { + shutting_down_ = true; + if (!want_write_ && out_off_ == out_.size()) { + close(); + } else { + close_after_write_ = true; + } +} + +void Connection::close() { dead_ = true; } + +} // namespace server diff --git a/src/server/connection.h b/src/server/connection.h new file mode 100644 index 0000000..fff9bbf --- /dev/null +++ b/src/server/connection.h @@ -0,0 +1,62 @@ +#pragma once + +#include "http/request.h" +#include "http/status.h" +#include "net/socket.h" +#include "util/time.h" + +#include +#include +#include +#include + +namespace server { + +// One keep-alive HTTP connection driven by reactor callbacks. Reads requests +// through http::RequestParser, renders responses through the shared handler, +// buffers outbound bytes, and closes when told (idle timeout, parse error, +// Connection: close, or server shutdown). +class Connection { + public: + // Renders the body for one request. Must not retain request views. + using Handler = std::function; + + Connection(net::Socket socket, const Handler& handler, std::chrono::milliseconds idle); + + Connection(const Connection&) = delete; + Connection& operator=(const Connection&) = delete; + + const net::Socket& socket() const { return sock_; } + bool dead() const { return dead_; } + bool wantsWrite() const { return want_write_ && !dead_; } + std::uint64_t deadlineMs() const { return deadline_ms_; } + + void onReadable(); + void onWritable(); + void onTimeout(); // 408 best effort, then close + void onPeerError() { close(); } + // Finish what is queued, then close; no further reads. + void beginShutdown(); + + private: + void respond(http::Status status, std::string_view body, bool closeAfter); + // Maps http::RequestParser::Error; returns false once the connection is dying. + bool drainParser(); + void flush(); + void close(); + + net::Socket sock_; + const Handler* handler_; + std::uint64_t idle_ms_; + http::RequestParser parser_; + std::string out_; + std::size_t out_off_ = 0; + bool want_write_ = false; + bool close_after_write_ = false; + bool shutting_down_ = false; + bool dead_ = false; + std::uint64_t deadline_ms_ = 0; + char rbuf_[16384]; +}; + +} // namespace server diff --git a/src/server/server.cpp b/src/server/server.cpp new file mode 100644 index 0000000..aa25362 --- /dev/null +++ b/src/server/server.cpp @@ -0,0 +1,146 @@ +#include "server/server.h" + +#include "net/error.h" +#include "net/poll.h" +#include "util/time.h" + +#include +#include +#include + +namespace server { + +Server::Server(std::uint16_t port, Connection::Handler handler) + : handler_(std::move(handler)) { + listener_ = net::Socket::tcp4(); + bool ok = listener_.valid(); + if (ok) { + ok = listener_.setReuseAddress(); + } + if (ok) { + ok = listener_.bindAny(port); + } + if (ok) { + ok = listener_.setNonBlocking(true); // accepted sockets inherit (both OSes) + } + if (ok) { + ok = listener_.listen(128); + } + if (!ok) { + const auto e = net::lastError(); + throw std::runtime_error("listener setup failed (os error " + std::to_string(e.code) + + ")"); + } +} + +void Server::run(ShutdownSignal& signal) { + 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; + } + + 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 + for (auto& c : conns_) { + set.add(c->socket().handle(), + net::PollIn | (c->wantsWrite() ? static_cast(net::PollOut) + : static_cast(0))); + } + + const std::uint64_t now = util::monotonicMs(); + int timeout = -1; // -1: block until event (Linux self-pipe wakes on signal) + for (auto& c : conns_) { + if (c->dead()) { + continue; + } + const long long left = + c->deadlineMs() >= now ? static_cast(c->deadlineMs() - now) : 0; + 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 = timeout < 0 ? static_cast(left) + : std::min(timeout, static_cast(left)); + } + + 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) + } + 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)); + } + } + + for (std::size_t i = 0; i < existing; ++i) { + Connection& c = *conns_[i]; + if (c.dead()) { + continue; + } + const short re = set.revents(base + i); + if (re & net::PollIn) { + c.onReadable(); + } + if (!c.dead() && (re & net::PollOut) && c.wantsWrite()) { + c.onWritable(); + } + if (!c.dead() && (re & (net::PollErr | net::PollHup | net::PollNval))) { + c.onPeerError(); + } + } + + const std::uint64_t after = util::monotonicMs(); + for (auto& c : conns_) { + if (!c->dead() && after >= c->deadlineMs()) { + c->onTimeout(); + } + } + reap(); + } + conns_.clear(); // grace expired with bytes still queued — drop + signal.notifyStopped(); +} + +void Server::reap() { + conns_.erase(std::remove_if(conns_.begin(), conns_.end(), + [](const std::unique_ptr& c) { + return c->dead(); + }), + conns_.end()); +} + +} // namespace server diff --git a/src/server/server.h b/src/server/server.h new file mode 100644 index 0000000..856a41e --- /dev/null +++ b/src/server/server.h @@ -0,0 +1,40 @@ +#pragma once + +#include "net/socket.h" +#include "server/connection.h" +#include "server/shutdown.h" + +#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. +class Server { + public: + // Throws std::runtime_error when bind/listen fails. + Server(std::uint16_t port, Connection::Handler handler); + Server(const Server&) = delete; + Server& operator=(const Server&) = delete; + + std::uint16_t port() const { return listener_.localPort(); } + + void run(ShutdownSignal& signal); + + std::chrono::milliseconds idle_timeout{5000}; // per-connection request timeout + std::chrono::milliseconds shutdown_grace{5000}; // drain budget after stop + int accept_batch = 64; // accepts per poll wakeup + + private: + void reap(); + + net::Socket listener_; + Connection::Handler handler_; + std::vector> conns_; +}; + +} // namespace server diff --git a/src/server/shutdown.h b/src/server/shutdown.h new file mode 100644 index 0000000..12e95a8 --- /dev/null +++ b/src/server/shutdown.h @@ -0,0 +1,35 @@ +#pragma once + +#include + +namespace net { +class PollSet; +} + +namespace server { + +// Pluggable stop trigger for the reactor. app/ implements it over OS signals; +// tests drive it directly. The reactor re-checks stopRequested() after every +// poll, so an implementation without a wake handle (Windows console events) +// only needs a bounded pollCap(). +class ShutdownSignal { + public: + virtual ~ShutdownSignal() = default; + + virtual bool stopRequested() const = 0; + + // Integrate the wake handle into the poll set (Linux self-pipe). Default: none. + virtual void preparePoll(net::PollSet& set) { (void)set; } + + // Inspect revents after poll; consume a wake and record it as a stop request. + virtual void afterPoll(const net::PollSet& set) { (void)set; } + + // Maximum poll wait while waiting for the stop signal; 0 = no cap. + virtual std::chrono::milliseconds pollCap() const { return std::chrono::milliseconds(0); } + + // Called once after the loop ends, before run() returns (lets a blocking + // Windows console-control handler return). + virtual void notifyStopped() {} +}; + +} // namespace server diff --git a/src/util/time.h b/src/util/time.h new file mode 100644 index 0000000..a726baf --- /dev/null +++ b/src/util/time.h @@ -0,0 +1,16 @@ +#pragma once + +#include +#include + +namespace util { + +// Wall-clock-independent milliseconds for deadlines (immune to clock changes). +inline std::uint64_t monotonicMs() { + return static_cast( + std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count()); +} + +} // namespace util diff --git a/tests/test_server.cpp b/tests/test_server.cpp new file mode 100644 index 0000000..77b1ded --- /dev/null +++ b/tests/test_server.cpp @@ -0,0 +1,181 @@ +#include "doctest.h" + +#include "http/request.h" +#include "net/context.h" +#include "net/error.h" +#include "net/socket.h" +#include "server/server.h" +#include "server/shutdown.h" + +#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}; +}; + +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) { +#ifdef _WIN32 + DWORD v = static_cast(ms); + ::setsockopt(s.handle(), SOL_SOCKET, SO_RCVTIMEO, reinterpret_cast(&v), + sizeof(v)); +#else + timeval tv{ms / 1000, static_cast(ms % 1000) * 1000L}; + ::setsockopt(s.handle(), SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); +#endif +} + +net::Socket connectTo(unsigned short port) { + auto s = net::Socket::tcp4(); + if (s.valid()) { + s.connect(loopback(port)); + } + setRecvTimeout(s, 3000); + return s; +} + +// One full response: head + Content-Length bytes, or whatever arrived before +// timeout/EOF. +std::string readResponse(net::Socket& s) { + std::string buf; + char chunk[256]; + for (;;) { + const int n = s.recvSome(chunk, sizeof(chunk)); + if (n <= 0) { + return buf; + } + buf.append(chunk, static_cast(n)); + const std::size_t head_end = buf.find("\r\n\r\n"); + if (head_end == std::string::npos) { + continue; + } + const std::size_t cl_pos = buf.find("Content-Length: "); + if (cl_pos == std::string::npos || cl_pos > head_end) { + return buf; // our server always sets Content-Length; nothing to wait for + } + 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'); + } + if (buf.size() >= head_end + 4 + cl) { + return buf; + } + } +} + +bool seesEof(net::Socket& s) { + char b[8]; + return s.recvSome(b, sizeof(b)) == 0; // 0 = clean EOF; -1 = timeout/error +} + +bool sendReq(net::Socket& s, std::string_view req) { + return s.sendAll(req.data(), req.size()); +} + +} // namespace + +TEST_CASE("200 over keep-alive, then Connection: close, then clean stop") { + net::Context ctx; + ManualSignal sig; + server::Server srv(0, [](const http::Request& req) { + return req.target == "/close" ? std::string("bye\n") : std::string("ok\n"); + }); + const unsigned short port = srv.port(); + REQUIRE(port > 0); + std::thread runner([&] { srv.run(sig); }); + + auto c = connectTo(port); + REQUIRE(c.valid()); + REQUIRE(sendReq(c, "GET /a HTTP/1.1\r\n\r\n")); + const std::string r1 = readResponse(c); + CHECK(r1.find("HTTP/1.1 200 OK") != std::string::npos); + CHECK(r1.find("Content-Length: 3") != std::string::npos); + CHECK(r1.find("\r\n\r\nok\n") != std::string::npos); + + // same socket: keep-alive worked; this one asks to close + REQUIRE(sendReq(c, "GET /close HTTP/1.1\r\nConnection: close\r\n\r\n")); + const std::string r2 = readResponse(c); + CHECK(r2.find("HTTP/1.1 200 OK") != std::string::npos); + CHECK(r2.find("Connection: close") != std::string::npos); + CHECK(seesEof(c)); // server closed after flush + + sig.stop.store(true); + runner.join(); +} + +TEST_CASE("idle connection receives 408 then gets closed") { + net::Context ctx; + ManualSignal sig; + server::Server srv(0, [](const http::Request&) { return std::string("ok\n"); }); + srv.idle_timeout = std::chrono::milliseconds(150); + const unsigned short port = srv.port(); + std::thread runner([&] { srv.run(sig); }); + + auto c = connectTo(port); + REQUIRE(c.valid()); + std::this_thread::sleep_for(std::chrono::milliseconds(400)); + const std::string r = readResponse(c); + CHECK(r.find("408") != std::string::npos); + CHECK(seesEof(c)); + + // server still serving: fresh connection works + auto c2 = connectTo(port); + REQUIRE(c2.valid()); + REQUIRE(sendReq(c2, "GET / HTTP/1.1\r\n\r\n")); + CHECK(readResponse(c2).find("200") != std::string::npos); + + sig.stop.store(true); + runner.join(); +} + +TEST_CASE("malformed request gets 400 and the connection drops") { + net::Context ctx; + ManualSignal sig; + server::Server srv(0, [](const http::Request&) { return std::string("ok\n"); }); + const unsigned short port = srv.port(); + std::thread runner([&] { srv.run(sig); }); + + auto c = connectTo(port); + REQUIRE(c.valid()); + REQUIRE(sendReq(c, "BOGUS\r\n\r\n")); + const std::string r = readResponse(c); + CHECK(r.find("400 Bad Request") != std::string::npos); + CHECK(seesEof(c)); + + sig.stop.store(true); + runner.join(); +} + +TEST_CASE("stop while connections are idle returns promptly") { + net::Context ctx; + ManualSignal sig; + server::Server srv(0, [](const http::Request&) { return std::string("ok\n"); }); + const unsigned short port = srv.port(); + std::thread runner([&] { srv.run(sig); }); + + auto c = connectTo(port); + REQUIRE(c.valid()); + // One served request guarantees the reactor accepted this connection; it is + // then idle when the stop lands. + REQUIRE(sendReq(c, "GET / HTTP/1.1\r\n\r\n")); + CHECK(readResponse(c).find("200") != std::string::npos); + sig.stop.store(true); + runner.join(); // idle connection is closed by beginShutdown; no grace wait + CHECK(seesEof(c)); +} diff --git a/tools/Dockerfile.asan b/tools/Dockerfile.asan index e2bd661..a59ae45 100644 --- a/tools/Dockerfile.asan +++ b/tools/Dockerfile.asan @@ -1,4 +1,4 @@ FROM ubuntu:24.04 RUN apt-get update \ - && apt-get install -y --no-install-recommends g++ cmake ninja-build ca-certificates \ + && apt-get install -y --no-install-recommends g++ cmake ninja-build ca-certificates netcat-openbsd \ && rm -rf /var/lib/apt/lists/* diff --git a/tools/asan-inside.sh b/tools/asan-inside.sh index 7ff9ff0..0fee41e 100644 --- a/tools/asan-inside.sh +++ b/tools/asan-inside.sh @@ -1,10 +1,36 @@ #!/usr/bin/env bash -# Runs inside the sanitizer container: full-tree ASan/UBSan build + ctest. -# Used by tools/ck-asan-docker.sh locally and by the ci linux job. +# Runs inside the sanitizer container: full-tree sanitizer build + ctest. +# SAN selects the sanitizer set: default address,undefined; thread also valid. +# Used by tools/ck-asan-docker.sh locally and by the ci linux jobs. set -euo pipefail +SAN="${SAN:-address,undefined}" +if [ "$SAN" = "thread" ]; then + OPT=HTTP_SERVER_ENABLE_TSAN +else + OPT=HTTP_SERVER_ENABLE_SANITIZERS +fi cmake -S . -B /tmp/asan -G Ninja \ -DCMAKE_BUILD_TYPE=Debug \ - -DHTTP_SERVER_ENABLE_SANITIZERS=ON + -D${OPT}=ON cmake --build /tmp/asan -j export ASAN_OPTIONS=detect_leaks=1:detect_stack_use_after_return=1 +export TSAN_OPTIONS=halt_on_error=1 + +# Graceful-shutdown smoke: SIGINT must drain, exit 0 (leak check rides on exit). +echo "== SIGINT graceful stop smoke ==" +/tmp/asan/http_server 18099 & +srv=$! +sleep 0.5 +line=$(printf 'GET / HTTP/1.1\r\n\r\n' | timeout 5 nc 127.0.0.1 18099 | head -1) +echo "$line" | grep -q "200" || { echo "smoke: expected 200, got: $line"; kill -9 $srv; exit 1; } +kill -INT $srv +for _ in $(seq 40); do kill -0 $srv 2>/dev/null || break; sleep 0.1; done +if kill -0 $srv 2>/dev/null; then + echo "smoke: server still running 4s after SIGINT" + kill -9 $srv + exit 1 +fi +wait $srv +echo "smoke ok" + ctest --test-dir /tmp/asan --output-on-failure