From 558f2679c4dcbb75948bcb2dd7ae3d658ccc474b Mon Sep 17 00:00:00 2001 From: orinnn5 Date: Tue, 16 Jun 2026 18:15:51 +0700 Subject: [PATCH] feat(cpp_server): centralize errno classification + add TCP relay server MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit os_error.h: single source of truth mapping raw errno to app-level IoOutcome (Retry / WouldBlock / PeerGone / FdExhausted / NoBufferSpace / Fatal), plus an io_should_wait() helper for the immediate-I/O fast paths. Wire classify_errno() into all three event-loop backends, replacing the scattered `errno == EAGAIN || EWOULDBLOCK` / `== EINTR` checks: - event_loop_kqueue.cpp (5 sites) - event_loop_epoll.cpp (5 sites) - event_loop_uring.cpp (4 sites; io_uring uses negated errno, classifies -ret/-res) Behavior is unchanged — pure de-duplication of error handling. tcp_replay_server.cpp: standalone TCP relay (machine A -> relay -> machine B) with bidirectional pumping and a --selftest mode that needs no external hosts or capture files. Added a `make replay` target. MACOS_USERSPACE.md: notes on the macOS/kqueue userspace path — syscalls, the kqueue event model, the errno->decision table (now implemented in os_error.h), macOS-vs-Linux gotchas, and running on port 443 against ISP interference. Verified on macOS: make cpp_server + make test (test_http_parser, test_tcp_ring) clean under -Wall -Wextra; make replay + --selftest round-trips 1 MiB OK. epoll/uring are Linux-only and not built here — changes are mechanical token replacements through the shared, compile-checked header. Co-Authored-By: Claude Opus 4.8 (1M context) --- base/cpp_server/MACOS_USERSPACE.md | 247 +++++++++++++++++++++++ base/cpp_server/Makefile | 14 +- base/cpp_server/event_loop_epoll.cpp | 12 +- base/cpp_server/event_loop_kqueue.cpp | 22 +-- base/cpp_server/event_loop_uring.cpp | 9 +- base/cpp_server/os_error.h | 41 ++++ base/cpp_server/tcp_replay_server.cpp | 274 ++++++++++++++++++++++++++ 7 files changed, 596 insertions(+), 23 deletions(-) create mode 100644 base/cpp_server/MACOS_USERSPACE.md create mode 100644 base/cpp_server/os_error.h create mode 100644 base/cpp_server/tcp_replay_server.cpp diff --git a/base/cpp_server/MACOS_USERSPACE.md b/base/cpp_server/MACOS_USERSPACE.md new file mode 100644 index 0000000..7b07a45 --- /dev/null +++ b/base/cpp_server/MACOS_USERSPACE.md @@ -0,0 +1,247 @@ +# macOS Userspace Networking — Notes for `cpp_server` + +How this server talks to the macOS kernel: the syscalls it crosses on, the +kqueue event model it is built around, the OS error codes (`errno`) it must +translate into application decisions, and the macOS-specific traps that differ +from the Linux (io_uring/epoll) backends. + +Scope: this is the **macOS / kqueue** path (`event_loop_kqueue.cpp`, +`tcp_layer.cpp`, `platform.h`). The Linux path uses `io_uring` +(`event_loop_uring.cpp`) or `epoll` (`event_loop_epoll.cpp`) and is only +referenced here for contrast. + +--- + +## 1. The userspace ⇄ kernel boundary + +Userspace code never touches the network card, socket buffers, or the TCP state +machine directly. Every interaction is a **syscall** — a controlled trap into +the kernel. The cost model that drives the whole design: + +- A syscall is a mode switch (user → kernel → user), ~hundreds of ns to low µs. +- Data crossing the boundary is **copied** (kernel socket buffer ⇄ userspace + buffer) on every `read`/`recv`/`write`/`send`. +- The kernel owns readiness; userspace must *ask* (`kevent`) instead of + busy-looping on sockets. + +That is why the hot path here is: register interest once → block in one +`kevent()` → process a batch → copy data with as few `read`/`write` calls as +possible. The ring buffer in `TCPConnection` exists to keep those copies +contiguous and avoid re-`memmove`-ing between syscalls. + +### Syscalls this server crosses on (macOS) + +| Syscall | Where | Purpose | +|---|---|---| +| `socket()` | `tcp_layer.cpp::create_listener` | allocate socket + TCP state | +| `bind()`, `listen()` | `create_listener` | bind to port, open accept queue | +| `accept()` | `kqueue::process_event`, `accept_connection` | pull a new connection off the queue | +| `fcntl(F_SETFL, O_NONBLOCK)` | `set_nonblocking` | make I/O non-blocking (required for kqueue) | +| `setsockopt(TCP_NODELAY)` | `configure_accepted_socket` | disable Nagle for low latency | +| `setsockopt(SO_REUSEADDR/SO_REUSEPORT)` | `create_listener` | fast rebind / multi-worker listeners | +| `kqueue()` | `kqueue::setup` | create the kernel event queue | +| `kevent()` | everywhere | register interest **and** wait for events | +| `read()` / `write()` | `process_event` | copy bytes across the boundary | +| `pipe()` | `kqueue::setup` | self-pipe to wake the loop on shutdown | +| `close()` | connection teardown | release the fd | + +> **Note:** on macOS the server uses `read()`/`write()` on the socket fd. The +> Linux backend prefers `recv()`/`send()` (and io_uring `prep_recv`/`prep_send`) +> because those map more directly onto socket semantics. Functionally +> equivalent here since the fds are sockets. + +--- + +## 2. The kqueue model (how `kevent` actually behaves) + +`kqueue` is macOS/BSD's readiness notifier — the analogue of Linux `epoll`. One +`kevent()` call does double duty: you pass it a list of **changes** (filters to +add/enable/delete) and it returns a list of **events** that are ready. + +```c +struct kevent kev; +EV_SET(&kev, fd, EVFILT_READ, EV_ADD | EV_ENABLE | EV_ONESHOT, 0, 0, udata); +kevent(kq, &kev, 1, events, BATCH, &timeout); // submit change + wait +``` + +Key fields and how this codebase uses them: + +- **`ident`** — the fd. On return, `events[i].ident` tells you which fd fired. +- **`filter`** — `EVFILT_READ` (readable) or `EVFILT_WRITE` (writable). The loop + branches on this in `process_event`. +- **`flags`**: + - `EV_ADD | EV_ENABLE` — start watching. + - `EV_ONESHOT` — **fire once, then auto-remove.** This codebase arms reads and + writes one-shot, so after every event it must re-arm if it wants more. This + is deliberate: it gives explicit backpressure (no event storms) at the cost + of one re-register per I/O. + - `EV_DISABLE` — used to pause the listener under accept backpressure + (`set_accepting(false)` when at `max_connections`). + - `EV_DELETE` — remove on connection teardown (`unregister_connection`). + - `EV_EOF` — **set by the kernel when the peer closed its end.** The loop + checks this first in `process_event` and reports `ECONNRESET`. +- **`udata`** — opaque pointer. The server stuffs the `TCPConnection*` here, so + events carry their connection without a map lookup (it also keeps an + `fd_to_conn_` map as the source of truth). + +### The self-pipe shutdown trick + +`kevent()` blocks. To unblock it on shutdown without a race, `setup()` creates a +non-blocking `pipe()` and registers the read end with `EVFILT_READ`. `shutdown()` +writes one byte; the loop wakes, sees the wakeup fd in `ident`, and breaks. This +is the portable POSIX way to interrupt a blocking event wait (Linux uses +`eventfd`). + +### Adaptive timeout + +The loop passes a `timespec` timeout to `kevent()`. When idle it ramps the +timeout up (to `max_adaptive_timeout_ms`); on the first event it drops back to +1 ms. This trades a little latency under no load for far less CPU spin — relevant +because macOS has no `SO_BUSY_POLL`. + +--- + +## 3. OS error codes → application decisions (the errno table) + +This is the core mapping: when a syscall returns `-1`, `errno` tells you *why*, +and the app must turn that into one of a few decisions: **retry**, **re-arm and +wait**, **drop this connection**, **pause accepting**, or **log-and-continue**. + +`errno` is **thread-local** — safe to read per worker thread. Always read it +*immediately* after the failing call (any intervening libc call may clobber it). + +### Reference table (errnos actually handled in this codebase) + +| `errno` | Meaning (macOS) | Where handled | App decision | +|---|---|---|---| +| `EAGAIN` / `EWOULDBLOCK` | No data / would block on a non-blocking fd. **Not an error** — the normal "drained" signal. On macOS these two are the same value. | `tcp_layer.cpp`, all event loops (13× each) | Re-arm the one-shot filter and wait for the next event. On `accept()`, stop the accept burst. | +| `ECONNABORTED` | Client dropped the half-open connection before `accept()` completed. | `kqueue::process_event`, `accept_connection` | Skip this fd, keep accepting (`continue`). | +| `ECONNRESET` | Peer sent RST / closed abnormally. Also synthesized from `EV_EOF` and from `read() == 0` (clean EOF). | `kqueue::process_event` (2×) | Close the connection, free it, update backpressure. | +| `ENOBUFS` | No buffer space — here it's the *app's* ring buffer being full (`reserve_read_span().len == 0`), surfaced as `ENOBUFS` to the handler. | `kqueue::process_event` | Stop reading this connection; let the handler decide (usually drain or close). | +| `EMFILE` | Per-process fd limit hit. | `tcp_layer.cpp::accept_connection` | Log **rate-limited** (1 in 100), return null, keep serving existing conns. Raise `ulimit -n`. | +| `ENFILE` | System-wide fd limit hit. | `tcp_layer.cpp::accept_connection` | Same as `EMFILE` — rate-limited log, don't spam. | +| `EINTR` | Syscall interrupted by a signal before completing. | all event loops (`kevent` wait) | Re-check shutdown flag; if still running, retry the wait. Never treat as fatal. | + +### EOF vs error — a subtlety + +`read()`/`recv()` returning **`0`** means an orderly shutdown (peer closed), which +is *not* an `errno` condition. This codebase maps that `0` onto `ECONNRESET` in +the event payload so the upper layer has a single "connection is gone" path. +That's a deliberate simplification, not a literal RST. + +### A drop-in classifier for the app + +If you want the table *in code* rather than scattered `if (errno == …)` checks, +this is the shape — group raw `errno` into a small set of app-level outcomes: + +```cpp +// os_error.h — classify a failed syscall's errno into an app decision. +#include + +enum class IoOutcome { + Retry, // EINTR — redo the syscall + WouldBlock, // EAGAIN/EWOULDBLOCK — re-arm the filter, wait + PeerGone, // ECONNRESET/EPIPE/ECONNABORTED/EOF — drop this connection + FdExhausted, // EMFILE/ENFILE — shed load, log rate-limited + NoBufferSpace,// ENOBUFS — app/kernel buffer full, back off + Fatal, // anything unexpected — log errno and close +}; + +inline IoOutcome classify_errno(int e) { + switch (e) { + case EINTR: return IoOutcome::Retry; + case EAGAIN: return IoOutcome::WouldBlock; +#if EWOULDBLOCK != EAGAIN + case EWOULDBLOCK: return IoOutcome::WouldBlock; +#endif + case ECONNRESET: + case EPIPE: + case ECONNABORTED: return IoOutcome::PeerGone; + case EMFILE: + case ENFILE: return IoOutcome::FdExhausted; + case ENOBUFS: return IoOutcome::NoBufferSpace; + default: return IoOutcome::Fatal; + } +} +``` + +> On macOS `EWOULDBLOCK == EAGAIN`, so the `#if` guard avoids a duplicate-`case` +> compile error while staying correct on platforms where they differ. + +This is implemented in **`os_error.h`** and wired into all three event-loop +backends (`event_loop_kqueue.cpp`, `event_loop_epoll.cpp`, `event_loop_uring.cpp`). +The scattered `if (errno == EAGAIN || errno == EWOULDBLOCK)` / `== EINTR` checks +now go through `classify_errno()` (or the `io_should_wait()` convenience for the +immediate-I/O fast paths). io_uring uses negated errno, so it classifies `-ret` / +`-res`. Behavior is unchanged — this is purely a single source of truth. + +--- + +## 4. macOS-specific gotchas vs Linux + +Things that bite when porting the Linux fast paths to the kqueue backend: + +| Topic | macOS reality | Consequence in this codebase | +|---|---|---| +| `SIGPIPE` | Writing to a socket whose peer closed raises `SIGPIPE`, which **kills the process** by default. | `tcp_replay_server.cpp` does `signal(SIGPIPE, SIG_IGN)`. The main server relies on this too — alternatively set `SO_NOSIGPIPE` per-socket (macOS) or pass `MSG_NOSIGNAL` to `send()` (Linux only; macOS lacks it). | +| `SO_BUSY_POLL` | **Does not exist** on macOS. | Guarded out (`#if PLATFORM_LINUX`). The adaptive `kevent` timeout is the macOS substitute for latency tuning. | +| `io_uring` | Linux-only. | macOS uses kqueue; there is no shared-ring zero-syscall path. | +| `accept4()` | **No `accept4`** on macOS; accepted fds **do not inherit** `O_NONBLOCK` or `TCP_NODELAY`. | Every accepted fd must be reconfigured explicitly — `configure_accepted_socket()` does `fcntl(O_NONBLOCK)` + `TCP_NODELAY`. Forgetting this is a classic macOS hang. | +| `SO_REUSEPORT` | Exists but load-balances differently than Linux's kernel hashing. | Multi-worker listeners share the port; don't assume even distribution. | +| CPU affinity | macOS has no `sched_setaffinity`; only `THREAD_AFFINITY_POLICY` *hints*, usually ignored without privileges. | `main.cpp::set_cpu_affinity` treats failure as non-fatal and silent on macOS. | +| `EV_EOF` timing | kqueue may set `EV_EOF` **together with** still-readable bytes. | Current loop returns on `EV_EOF` before draining — last bytes can be dropped on a peer that closes immediately after writing. Worth revisiting if you see truncated final reads. | +| Privileged ports | Binding `< 1024` (e.g. **443**) requires root. | See §5. | + +--- + +## 5. Running on port 443 / surviving ISP interference + +This is a legitimate concern for a self-hosted relay (`tcp_replay_server`): ISPs +and middleboxes often throttle or block "unknown" high ports while leaving +**443/tcp (HTTPS)** alone, because blocking 443 breaks the web. Putting your own +traffic on 443 and making it look like ordinary TLS is the standard way to keep a +personal relay reachable. Practical, on-topic techniques: + +1. **Bind to 443.** It's a privileged port — needs root (`sudo`), a + `setcap`-style entitlement, or a front like `launchd`/reverse proxy that owns + 443 and forwards to your unprivileged port. On macOS, `pfctl` port redirects + or running the relay as root are the usual options. + +2. **Make it actually look like HTTPS, not just sit on 443.** Deep-packet + inspection (DPI) classifies by *content*, not port. A raw TCP stream on 443 is + easy to flag. Wrap the relay in **TLS** so the handshake and records are real + HTTPS: + - Terminate TLS at the relay (e.g. front with `stunnel`, `nginx stream`, or + add an `SSL_*`/`SecureTransport` layer to the relay's accept path). + - Present a valid certificate and a plausible **SNI**. Traffic then looks like + a normal HTTPS session to any on-path observer. + +3. **Keepalive + multiplexing.** Long-lived 443 connections that carry framed + data resemble HTTP/2 or WebSocket. Tools that survive hostile networks + (Shadowsocks, V2Ray/VLESS, Tor's `obfs4`/`meek`) all converge on "look like + TLS-to-a-CDN." If you need robustness beyond a hobby relay, those are the + reference designs rather than reinventing the obfuscation. + +4. **Fallbacks.** Try 443 first, fall back to 8443/993/some other commonly-open + port; detect throttling (sudden RSTs, latency cliffs → `ECONNRESET`/timeouts in + the table above) and rotate. + +> Caveat: putting traffic on 443 only defeats *port-based* filtering. Against DPI +> you need real TLS framing — plain bytes on 443 get fingerprinted quickly. And +> note this is for routing **your own** traffic across an interfering network; +> respect the acceptable-use terms of networks you don't own. + +--- + +## 6. Quick reference — verifying the macOS path + +```bash +make cpp_server # builds the kqueue backend on macOS +make test # test_http_parser + test_tcp_ring +./tcp_replay_server --selftest # A -> relay -> B -> relay -> A round-trip + +# inspect fds / limits while running +ulimit -n # current per-process fd cap (raise for EMFILE) +lsof -p | wc -l # how many fds the server holds +sudo dtruss -p # trace syscalls (macOS strace equivalent; needs SIP off) +``` diff --git a/base/cpp_server/Makefile b/base/cpp_server/Makefile index 07ea55f..fb0e18e 100644 --- a/base/cpp_server/Makefile +++ b/base/cpp_server/Makefile @@ -9,12 +9,14 @@ HFT_TARGET = hft_sim HFT_ENGINE_TARGET = hft_engine TEST_HTTP_PARSER_TARGET = test_http_parser TEST_TCP_RING_TARGET = test_tcp_ring +REPLAY_TARGET = tcp_replay_server SRC = main.cpp tcp_layer.cpp http_layer.cpp cpu_layer.cpp memory_layer.cpp config.cpp cache_layer.cpp event_loop.cpp HFT_SRC = hft_sim.cpp HFT_ENGINE_SRC = hft_engine.cpp TEST_HTTP_PARSER_SRC = tests/test_http_parser.cpp TEST_TCP_RING_SRC = tests/test_tcp_ring.cpp MINI_REDIS_SRC = mini_redis_server.cpp +REPLAY_SRC = tcp_replay_server.cpp # Platform-specific event loop sources ifeq ($(UNAME_S),Linux) @@ -29,6 +31,7 @@ HFT_ENGINE_OBJ = $(HFT_ENGINE_SRC:.cpp=.o) TEST_HTTP_PARSER_OBJ = $(TEST_HTTP_PARSER_SRC:.cpp=.o) TEST_TCP_RING_OBJ = $(TEST_TCP_RING_SRC:.cpp=.o) MINI_REDIS_OBJ = $(MINI_REDIS_SRC:.cpp=.o) +REPLAY_OBJ = $(REPLAY_SRC:.cpp=.o) # Core objects for non-server binaries (exclude main.cpp to avoid duplicate main) CORE_SRC = tcp_layer.cpp http_layer.cpp cpu_layer.cpp memory_layer.cpp config.cpp cache_layer.cpp event_loop.cpp @@ -60,7 +63,7 @@ else $(error Unsupported platform: $(UNAME_S)) endif -.PHONY: all clean check install-deps pgo pgo-clean hft hft-engine test mini-redis +.PHONY: all clean check install-deps pgo pgo-clean hft hft-engine test mini-redis replay all: check $(TARGET) @@ -74,6 +77,8 @@ test: check $(TEST_HTTP_PARSER_TARGET) $(TEST_TCP_RING_TARGET) mini-redis: check $(MINI_REDIS_TARGET) +replay: check $(REPLAY_TARGET) + check: @if [ "$(UNAME_S)" = "Linux" ]; then \ if [ "$(CHECK_LIBURING)" != "yes" ]; then \ @@ -110,11 +115,16 @@ $(MINI_REDIS_TARGET): $(MINI_REDIS_OBJ) $(CORE_OBJ) $(CXX) $(CXXFLAGS) -o $(MINI_REDIS_TARGET) $(MINI_REDIS_OBJ) $(CORE_OBJ) $(LDFLAGS) @echo "Build complete: $(MINI_REDIS_TARGET)" +# Standalone relay/replay server (no dependency on the server core objects) +$(REPLAY_TARGET): $(REPLAY_OBJ) + $(CXX) $(CXXFLAGS) -o $(REPLAY_TARGET) $(REPLAY_OBJ) $(LDFLAGS) + @echo "Build complete: $(REPLAY_TARGET)" + %.o: %.cpp $(CXX) $(CXXFLAGS) -c $< -o $@ clean: - rm -f $(TARGET) $(OBJ) $(HFT_TARGET) $(HFT_OBJ) $(HFT_ENGINE_TARGET) $(HFT_ENGINE_OBJ) $(TEST_HTTP_PARSER_TARGET) $(TEST_HTTP_PARSER_OBJ) $(TEST_TCP_RING_TARGET) $(TEST_TCP_RING_OBJ) $(MINI_REDIS_TARGET) $(MINI_REDIS_OBJ) + rm -f $(TARGET) $(OBJ) $(HFT_TARGET) $(HFT_OBJ) $(HFT_ENGINE_TARGET) $(HFT_ENGINE_OBJ) $(TEST_HTTP_PARSER_TARGET) $(TEST_HTTP_PARSER_OBJ) $(TEST_TCP_RING_TARGET) $(TEST_TCP_RING_OBJ) $(MINI_REDIS_TARGET) $(MINI_REDIS_OBJ) $(REPLAY_TARGET) $(REPLAY_OBJ) install-deps: @echo "Installing liburing-dev..." diff --git a/base/cpp_server/event_loop_epoll.cpp b/base/cpp_server/event_loop_epoll.cpp index e7404e0..5d0ec2f 100644 --- a/base/cpp_server/event_loop_epoll.cpp +++ b/base/cpp_server/event_loop_epoll.cpp @@ -3,6 +3,7 @@ #include "event_loop_epoll.h" #include "tcp_layer.h" #include "config.h" +#include "os_error.h" #include #include @@ -159,7 +160,7 @@ bool EpollEventLoop::register_write(TCPConnection* conn) { conn->consume_write((size_t)n); continue; } - if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) { + if (n < 0 && classify_errno(errno) == IoOutcome::WouldBlock) { break; // need EPOLLOUT } // fatal error @@ -210,7 +211,8 @@ void EpollEventLoop::process_listener_events() { while (true) { int fd = accept4(listener_fd_, nullptr, nullptr, SOCK_NONBLOCK | SOCK_CLOEXEC); if (fd < 0) { - if (errno == EAGAIN || errno == EWOULDBLOCK) break; + // WouldBlock = accept queue drained; any other error: stop this burst. + if (classify_errno(errno) == IoOutcome::WouldBlock) break; // transient accept errors: ignore break; } @@ -265,7 +267,7 @@ void EpollEventLoop::process_conn_event(TCPConnection* conn, uint32_t events) { conn->consume_write((size_t)n); continue; } - if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) { + if (n < 0 && classify_errno(errno) == IoOutcome::WouldBlock) { break; // wait for next EPOLLOUT } // fatal error @@ -329,7 +331,7 @@ void EpollEventLoop::process_conn_event(TCPConnection* conn, uint32_t events) { // EOF break; } - if (errno == EAGAIN || errno == EWOULDBLOCK) { + if (classify_errno(errno) == IoOutcome::WouldBlock) { break; } // fatal error @@ -361,7 +363,7 @@ void EpollEventLoop::run() { int timeout_ms = 10; // keep responsive; can make adaptive later int n = epoll_wait(epfd_, events_, MAX_EVENTS, timeout_ms); if (n < 0) { - if (errno == EINTR) { + if (classify_errno(errno) == IoOutcome::Retry) { // EINTR if (!running_ || !TCPServer::global_running_.load()) break; continue; } diff --git a/base/cpp_server/event_loop_kqueue.cpp b/base/cpp_server/event_loop_kqueue.cpp index b98d690..8852e91 100644 --- a/base/cpp_server/event_loop_kqueue.cpp +++ b/base/cpp_server/event_loop_kqueue.cpp @@ -3,6 +3,7 @@ #include "event_loop_kqueue.h" #include "tcp_layer.h" #include "config.h" +#include "os_error.h" #include #include #include @@ -151,7 +152,7 @@ bool KqueueEventLoop::register_write(TCPConnection* conn) { return true; } // Partial write - register for write event - } else if (n < 0 && errno != EAGAIN && errno != EWOULDBLOCK) { + } else if (n < 0 && classify_errno(errno) != IoOutcome::WouldBlock) { // Error conn->set_writing(false); if (server_ && event_callback_) { @@ -216,7 +217,7 @@ void KqueueEventLoop::run() { // SYSCALL: kevent() - wait for kernel events int nev = kevent(kq_, nullptr, 0, events, BATCH_SIZE, &timeout); if (nev < 0) { - if (errno == EINTR) { + if (classify_errno(errno) == IoOutcome::Retry) { // EINTR if (!running_ || !TCPServer::global_running_.load()) break; continue; } @@ -318,13 +319,10 @@ void KqueueEventLoop::process_event(const struct kevent& kev) { while (accept_count < max_accepts) { int new_fd = accept(listener_fd_, nullptr, nullptr); if (new_fd < 0) { - if (errno == EAGAIN || errno == EWOULDBLOCK) { - break; // No more connections - } - if (errno == ECONNABORTED) { - continue; // Client aborted, try next - } - break; // Other error + // ECONNABORTED (PeerGone): client bailed before accept — try next. + // WouldBlock (drained) or anything else: stop this accept burst. + if (classify_errno(errno) == IoOutcome::PeerGone) continue; + break; } // Create connection (will be handled by TCPServer) @@ -386,8 +384,8 @@ void KqueueEventLoop::process_event(const struct kevent& kev) { event_callback_(server_, event); } } else { - // Error or EAGAIN/EWOULDBLOCK - if (errno == EAGAIN || errno == EWOULDBLOCK) { + // Error or would-block + if (classify_errno(errno) == IoOutcome::WouldBlock) { // Would block - re-enable read event conn->set_reading(true); struct kevent kev; @@ -437,7 +435,7 @@ void KqueueEventLoop::process_event(const struct kevent& kev) { pending_events_.push_back(kev); } } else if (n < 0) { - if (errno == EAGAIN || errno == EWOULDBLOCK) { + if (classify_errno(errno) == IoOutcome::WouldBlock) { // Would block - re-enable write event struct kevent kev; EV_SET(&kev, fd, EVFILT_WRITE, EV_ADD | EV_ENABLE | EV_ONESHOT, 0, 0, conn); diff --git a/base/cpp_server/event_loop_uring.cpp b/base/cpp_server/event_loop_uring.cpp index 2b8e6f7..026d6a5 100644 --- a/base/cpp_server/event_loop_uring.cpp +++ b/base/cpp_server/event_loop_uring.cpp @@ -3,6 +3,7 @@ #include "event_loop_uring.h" #include "tcp_layer.h" #include "config.h" +#include "os_error.h" #include #include #include @@ -129,7 +130,7 @@ bool IOUringEventLoop::register_read(TCPConnection* conn) { event_callback_(server_, ev); return true; } - if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) { + if (io_should_wait(errno)) { // Try again later; caller can re-arm read. return false; } @@ -177,7 +178,7 @@ bool IOUringEventLoop::register_write(TCPConnection* conn) { event_callback_(server_, ev); return true; } - if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) { + if (n < 0 && io_should_wait(errno)) { // We'll try again later. return false; } @@ -241,7 +242,7 @@ void IOUringEventLoop::run() { if (ret == -ETIME) { continue; // periodic wake to check shutdown flags } - if (ret == -EINTR) { + if (ret < 0 && classify_errno(-ret) == IoOutcome::Retry) { // -EINTR if (!running_ || !TCPServer::global_running_.load()) break; continue; } @@ -300,7 +301,7 @@ void IOUringEventLoop::handle_completion(struct io_uring_cqe* cqe) { } else { // Transient non-blocking conditions are NORMAL for sockets. // Do NOT bubble these up as fatal errors. - if (err == EAGAIN || err == EWOULDBLOCK || err == EINTR) { + if (io_should_wait(err)) { if (conn->is_reading()) { conn->set_reading(false); register_read(conn); diff --git a/base/cpp_server/os_error.h b/base/cpp_server/os_error.h new file mode 100644 index 0000000..a6ee9bd --- /dev/null +++ b/base/cpp_server/os_error.h @@ -0,0 +1,41 @@ +// os_error.h — classify a failed syscall's errno into an application decision. +// +// Single source of truth for "what does this errno mean for us" across the +// kqueue / epoll / io_uring event-loop backends. See MACOS_USERSPACE.md §3. +#pragma once + +#include + +enum class IoOutcome { + Retry, // EINTR — redo the syscall + WouldBlock, // EAGAIN/EWOULDBLOCK — re-arm the filter, wait for readiness + PeerGone, // ECONNRESET/EPIPE/ECONNABORTED — drop this connection + FdExhausted, // EMFILE/ENFILE — shed load, log rate-limited + NoBufferSpace, // ENOBUFS — app/kernel buffer full, back off + Fatal, // anything unexpected — log errno and close +}; + +inline IoOutcome classify_errno(int e) { + switch (e) { + case EINTR: return IoOutcome::Retry; + case EAGAIN: return IoOutcome::WouldBlock; +#if EWOULDBLOCK != EAGAIN + case EWOULDBLOCK: return IoOutcome::WouldBlock; +#endif + case ECONNRESET: + case EPIPE: + case ECONNABORTED: return IoOutcome::PeerGone; + case EMFILE: + case ENFILE: return IoOutcome::FdExhausted; + case ENOBUFS: return IoOutcome::NoBufferSpace; + default: return IoOutcome::Fatal; + } +} + +// Convenience: the very common "would this syscall block / should I just retry" +// test used in the immediate-I/O fast paths. EINTR is grouped here because the +// caller's response is identical: stop now, try again on the next arming. +inline bool io_should_wait(int e) { + IoOutcome o = classify_errno(e); + return o == IoOutcome::WouldBlock || o == IoOutcome::Retry; +} diff --git a/base/cpp_server/tcp_replay_server.cpp b/base/cpp_server/tcp_replay_server.cpp new file mode 100644 index 0000000..98058be --- /dev/null +++ b/base/cpp_server/tcp_replay_server.cpp @@ -0,0 +1,274 @@ +// TCP Replay/Relay Server +// Forwards a TCP stream from machine A (downstream client) to machine B +// (upstream server), piping bytes in both directions. Each accepted client +// gets its own upstream connection and two pump threads. +// +// Usage: +// tcp_replay_server +// tcp_replay_server --selftest +// +// --selftest spins up an in-process echo "machine B", points the relay at it, +// then a client "machine A" pushes synthetic data through and verifies the +// round-trip. No external machines or capture files needed. + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace { + +std::atomic g_running{true}; + +void on_signal(int /*sig*/) { g_running.store(false); } + +void set_nodelay(int fd) { + int flag = 1; + setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag)); +} + +// Create a listening socket bound to `port` on all interfaces. +int make_listener(uint16_t port) { + int fd = socket(AF_INET, SOCK_STREAM, 0); + if (fd < 0) { perror("socket"); return -1; } + + int opt = 1; + setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)); + + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = INADDR_ANY; + addr.sin_port = htons(port); + + if (bind(fd, reinterpret_cast(&addr), sizeof(addr)) < 0) { + perror("bind"); + ::close(fd); + return -1; + } + if (listen(fd, 128) < 0) { + perror("listen"); + ::close(fd); + return -1; + } + return fd; +} + +// Return the port a listener was actually bound to (useful for port 0). +uint16_t bound_port(int fd) { + sockaddr_in addr{}; + socklen_t len = sizeof(addr); + if (getsockname(fd, reinterpret_cast(&addr), &len) < 0) return 0; + return ntohs(addr.sin_port); +} + +// Dial a TCP connection to host:port (host may be a name or dotted-quad). +int connect_to(const char* host, uint16_t port) { + char port_str[16]; + std::snprintf(port_str, sizeof(port_str), "%u", port); + + addrinfo hints{}; + hints.ai_family = AF_INET; + hints.ai_socktype = SOCK_STREAM; + + addrinfo* res = nullptr; + if (getaddrinfo(host, port_str, &hints, &res) != 0 || !res) { + std::fprintf(stderr, "relay: cannot resolve upstream %s:%u\n", host, port); + return -1; + } + + int fd = -1; + for (addrinfo* ai = res; ai; ai = ai->ai_next) { + fd = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol); + if (fd < 0) continue; + if (connect(fd, ai->ai_addr, ai->ai_addrlen) == 0) break; + ::close(fd); + fd = -1; + } + freeaddrinfo(res); + + if (fd < 0) { + std::fprintf(stderr, "relay: cannot connect to upstream %s:%u\n", host, port); + return -1; + } + set_nodelay(fd); + return fd; +} + +// Copy bytes from `src` to `dst` until EOF or error. On exit, half-close the +// write side of `dst` so the peer observes end-of-stream. +void pump(int src, int dst) { + char buf[16384]; + while (g_running.load()) { + ssize_t n = recv(src, buf, sizeof(buf), 0); + if (n <= 0) break; // 0 = peer closed, <0 = error + ssize_t off = 0; + while (off < n) { + ssize_t w = send(dst, buf + off, static_cast(n - off), 0); + if (w <= 0) { shutdown(dst, SHUT_WR); return; } + off += w; + } + } + shutdown(dst, SHUT_WR); +} + +// Relay one accepted client (A) to a fresh upstream connection (B). +void handle_client(int client_fd, std::string upstream_host, uint16_t upstream_port) { + set_nodelay(client_fd); + int upstream_fd = connect_to(upstream_host.c_str(), upstream_port); + if (upstream_fd < 0) { + ::close(client_fd); + return; + } + + // A -> B in this thread, B -> A in a helper thread. + std::thread back(pump, upstream_fd, client_fd); + pump(client_fd, upstream_fd); + back.join(); + + ::close(client_fd); + ::close(upstream_fd); +} + +int run_relay(uint16_t listen_port, const std::string& upstream_host, uint16_t upstream_port) { + int listener = make_listener(listen_port); + if (listener < 0) return 1; + + std::printf("relay: listening on :%u -> %s:%u\n", + bound_port(listener), upstream_host.c_str(), upstream_port); + + std::vector workers; + while (g_running.load()) { + int client = accept(listener, nullptr, nullptr); + if (client < 0) { + if (!g_running.load()) break; + continue; + } + workers.emplace_back(handle_client, client, upstream_host, upstream_port); + // Reap finished workers opportunistically to bound the vector. + for (auto it = workers.begin(); it != workers.end();) { + if (it->joinable()) { ++it; } else { it = workers.erase(it); } + } + } + + ::close(listener); + for (auto& t : workers) if (t.joinable()) t.join(); + std::printf("relay: shutdown\n"); + return 0; +} + +// ---- Self-test: A -> relay -> B(echo) -> relay -> A ----------------------- + +// Minimal echo upstream standing in for "machine B". +void echo_upstream(int listener, std::atomic* stop) { + int client = accept(listener, nullptr, nullptr); + if (client < 0) return; + set_nodelay(client); + char buf[16384]; + while (!stop->load()) { + ssize_t n = recv(client, buf, sizeof(buf), 0); + if (n <= 0) break; + ssize_t off = 0; + while (off < n) { + ssize_t w = send(client, buf + off, static_cast(n - off), 0); + if (w <= 0) break; + off += w; + } + } + ::close(client); +} + +int run_selftest() { + constexpr size_t kPayload = 1 << 20; // 1 MiB of synthetic data + + // Stand up echo "machine B" on an ephemeral port. + int up_listener = make_listener(0); + if (up_listener < 0) return 1; + uint16_t up_port = bound_port(up_listener); + + std::atomic echo_stop{false}; + std::thread echo(echo_upstream, up_listener, &echo_stop); + + // Stand up the relay on its own ephemeral port, pointed at B. + int relay_listener = make_listener(0); + if (relay_listener < 0) return 1; + uint16_t relay_port = bound_port(relay_listener); + + std::thread relay([&] { + int client = accept(relay_listener, nullptr, nullptr); + if (client >= 0) handle_client(client, "127.0.0.1", up_port); + }); + + // "Machine A": connect to the relay, push synthetic data, read it back. + int a = connect_to("127.0.0.1", relay_port); + if (a < 0) return 1; + + std::vector sent(kPayload), got(kPayload); + for (size_t i = 0; i < kPayload; ++i) sent[i] = static_cast(i * 1103515245u + 12345u); + + std::thread writer([&] { + size_t off = 0; + while (off < kPayload) { + ssize_t w = send(a, sent.data() + off, kPayload - off, 0); + if (w <= 0) break; + off += static_cast(w); + } + shutdown(a, SHUT_WR); // signal end-of-stream so the echo side finishes + }); + + size_t recvd = 0; + while (recvd < kPayload) { + ssize_t n = recv(a, got.data() + recvd, kPayload - recvd, 0); + if (n <= 0) break; + recvd += static_cast(n); + } + writer.join(); + + // Tear everything down. + g_running.store(false); + echo_stop.store(true); + ::close(a); + relay.join(); + echo.join(); + ::close(relay_listener); + ::close(up_listener); + + bool ok = (recvd == kPayload) && (std::memcmp(sent.data(), got.data(), kPayload) == 0); + std::printf("selftest: A->relay->B->relay->A sent=%zu recvd=%zu %s\n", + kPayload, recvd, ok ? "OK" : "MISMATCH"); + return ok ? 0 : 1; +} + +} // namespace + +int main(int argc, char* argv[]) { + signal(SIGINT, on_signal); + signal(SIGTERM, on_signal); + signal(SIGPIPE, SIG_IGN); // writes to a closed peer return EPIPE instead + + if (argc == 2 && std::strcmp(argv[1], "--selftest") == 0) { + return run_selftest(); + } + if (argc == 4) { + uint16_t listen_port = static_cast(std::atoi(argv[1])); + uint16_t upstream_port = static_cast(std::atoi(argv[3])); + return run_relay(listen_port, argv[2], upstream_port); + } + + std::fprintf(stderr, + "Usage:\n" + " %s \n" + " %s --selftest\n", + argv[0], argv[0]); + return 1; +}