From 1564a8275f1ec6720cba78c7a4766947b8e2c208 Mon Sep 17 00:00:00 2001 From: Ben Maurer Date: Fri, 10 Jul 2026 20:49:37 -0700 Subject: [PATCH 1/3] Fix busy-loop when a socket closes during waitOnSocketData waitOnSocketData treated EBADF/EINVAL from select() like EINTR and returned false. Callers (SocketHandler::readAll, RawSocketUtils::readAll) treat false as "retry", but a closed fd never becomes readable again, so an fd closed by another thread spun at 100% CPU. Throw instead, so retry loops treat it as socket death like any other read failure. The etterminal and jumphost init loops read packets outside any try/catch; give them one so a router that dies during init produces a logged fatal instead of an uncaught exception. --- src/base/Headers.hpp | 14 ++++++++++++-- src/terminal/UserJumphostHandler.cpp | 11 +++++++++-- src/terminal/UserTerminalHandler.cpp | 11 +++++++++-- 3 files changed, 30 insertions(+), 6 deletions(-) diff --git a/src/base/Headers.hpp b/src/base/Headers.hpp index c3c5468f8..d85f8eb3b 100644 --- a/src/base/Headers.hpp +++ b/src/base/Headers.hpp @@ -367,6 +367,12 @@ inline bool waitOnSocketData(int fd) { if (errno == EINTR) { // Interrupted by the signal, the caller will retry. return false; + } else if (errno == EBADF || errno == EINVAL) { + // The fd was closed under us (e.g. peer disconnected). It will never + // become readable, and callers use this helper in retry loops, so + // returning false would spin at 100% CPU. Throw so the loop treats + // it as socket death. + throw std::runtime_error("Socket closed while waiting for data"); } else { FATAL_FAIL(selectResult); } @@ -389,8 +395,12 @@ inline bool isSocketWritable(int fd) { tv.tv_usec = 0; const int selectResult = select(fd + 1, NULL, &fdset, NULL, &tv); if (selectResult < 0) { - if (errno == EINTR) { - // Interrupted by the signal, the caller will retry. + if (errno == EINTR || errno == EBADF || errno == EINVAL) { + // EINTR: interrupted by a signal. EBADF/EINVAL: the fd was closed + // under us (e.g. client disconnected). Returning false is safe in + // both cases: callers use this as a non-blocking "can I write more?" + // poll and stop on false; a dead socket is then surfaced by the next + // read/write on the connection. return false; } else { FATAL_FAIL(selectResult); diff --git a/src/terminal/UserJumphostHandler.cpp b/src/terminal/UserJumphostHandler.cpp index 38c11ca29..167754047 100644 --- a/src/terminal/UserJumphostHandler.cpp +++ b/src/terminal/UserJumphostHandler.cpp @@ -40,8 +40,15 @@ void UserJumphostHandler::run() { InitialPayload payload; while (true) { Packet initPacket; - if (!routerSocketHandler->readPacket(routerFd, &initPacket)) { - continue; + try { + if (!routerSocketHandler->readPacket(routerFd, &initPacket)) { + continue; + } + } catch (const std::runtime_error& re) { + // The router connection died before init (e.g. etserver shut down). + // Fail with a logged fatal instead of an uncaught exception. + STFATAL << "Router connection died waiting for jumphost init: " + << re.what(); } if (initPacket.getHeader() != TerminalPacketType::JUMPHOST_INIT) { STFATAL << "Invalid jumphost init packet header: " diff --git a/src/terminal/UserTerminalHandler.cpp b/src/terminal/UserTerminalHandler.cpp index b82ad9c3c..c148e2de6 100644 --- a/src/terminal/UserTerminalHandler.cpp +++ b/src/terminal/UserTerminalHandler.cpp @@ -40,8 +40,15 @@ UserTerminalHandler::UserTerminalHandler( void UserTerminalHandler::run() { while (true) { Packet termInitPacket; - if (!socketHandler->readPacket(routerFd, &termInitPacket)) { - continue; + try { + if (!socketHandler->readPacket(routerFd, &termInitPacket)) { + continue; + } + } catch (const std::runtime_error& re) { + // The router connection died before init (e.g. etserver shut down). + // Fail with a logged fatal instead of an uncaught exception. + STFATAL << "Router connection died waiting for terminal init: " + << re.what(); } if (termInitPacket.getHeader() != TerminalPacketType::TERMINAL_INIT) { STFATAL << "Invalid terminal init packet header: " From 3b5ab10f1b1ffff8a7c70f16cc2499b50d597e81 Mon Sep 17 00:00:00 2001 From: Ben Maurer Date: Fri, 10 Jul 2026 20:51:46 -0700 Subject: [PATCH 2/3] Add opt-in flow control: --flow-control {none,backpressure,discard} (#631) When a process produces output faster than the network can deliver it, the excess queues in kernel buffers - mostly the server's TCP send buffer, which autotunes to many MB. On a saturated link the display falls steadily behind real time and Ctrl-C appears dead: the interrupt is delivered, but the prompt has to wait behind the entire queue. On a 100KB/s test link, Ctrl-C took 56-116s to take effect. This adds a per-session flow control mode, chosen by the client: - none (default): the existing behavior, unchanged. The mode field is not even serialized when unset, so new binaries interoperate with old ones exactly as before, and users who don't opt in see zero change. - backpressure: terminal output is staged in a small (64KB) buffer that is drained only when select() reports the socket writable; when it fills, the server stops reading the PTY and the remote process pauses, like plain ssh. Lossless - safe for consumers that need every byte, such as tmux -CC. - discard: same buffer, but when it fills the oldest chunk is dropped. The remote process never stalls - even while the client is disconnected - and the display stays near real time. Output that exceeded the link rate is missing from scrollback. Application-level buffering only helps if the backlog actually waits in the application buffer, so opted-in sessions also shrink the kernel queues via SocketHandler::minimizeKernelBuffering (a no-op by default): - TCP_NOTSENT_LOWAT=32KB on the client connection: select() reports writable only when <32KB is unsent, keeping the backlog in the WriteBuffer where it can be gated or dropped. In-flight (sent-but-unacked) data is not limited, so throughput on high-BDP links is unaffected. It is reapplied every loop iteration because the connection gets a new socket on reconnect and fd numbers are reused. - SO_SNDBUF=64KB on the etterminal->etserver unix socket (pure queue, no in-flight component, so clamping costs nothing locally). etterminal learns the mode via a new TermInit field. In the jumphost path, discard only drops TERMINAL_BUFFER packets; control packets (port forwarding, responses, keepalives) are always delivered, and reads pause when the queue is dominated by them so it stays bounded. The client suppresses its missed-keepalive disconnect while it is intentionally not reading (console buffer full), since the keepalive echo may be sitting unread in the socket. Measured on a 100KB/s throttled link with a 2.7MB/s producer: display lag @30s producer Ctrl-C -> prompt none (unchanged) ~31s, growing throttled 56-116s backpressure ~2.5s, bounded throttled 2-3s discard ~1.2s, bounded full speed 2.1s In discard mode the producer also runs at full speed through a 60s network outage, and on reconnect the display recovers to ~1.2s of lag instead of replaying the backlog. --- proto/ETerminal.proto | 18 ++ src/base/PipeSocketHandler.cpp | 14 ++ src/base/PipeSocketHandler.hpp | 2 + src/base/SocketHandler.hpp | 9 + src/base/TcpSocketHandler.cpp | 19 ++ src/base/TcpSocketHandler.hpp | 2 + src/base/WriteBuffer.hpp | 150 +++++++++++++++ src/terminal/TerminalClient.cpp | 67 ++++++- src/terminal/TerminalClient.hpp | 5 +- src/terminal/TerminalClientMain.cpp | 27 ++- src/terminal/TerminalServer.cpp | 177 +++++++++++++++-- src/terminal/UserTerminalHandler.cpp | 5 + test/integration_tests/TerminalServerTest.cpp | 48 +++++ test/unit_tests/WriteBufferTest.cpp | 181 ++++++++++++++++++ 14 files changed, 700 insertions(+), 24 deletions(-) create mode 100644 src/base/WriteBuffer.hpp create mode 100644 test/unit_tests/WriteBufferTest.cpp diff --git a/proto/ETerminal.proto b/proto/ETerminal.proto index 6b831869c..deaa5b3fa 100644 --- a/proto/ETerminal.proto +++ b/proto/ETerminal.proto @@ -57,10 +57,25 @@ message PortForwardData { optional bool closed = 5; } +enum FlowControlMode { + // Legacy behavior: no application-level buffering or kernel buffer + // tuning. This is the default and what servers assume for old clients + // that don't send flow_control_mode: current users are unchanged. + FLOW_CONTROL_NONE = 0; + // Bounded buffering; when full, stop reading from the source. Lossless + // (like plain ssh, but with small, tuned buffers so Ctrl-C stays + // responsive on a saturated link). + FLOW_CONTROL_BACKPRESSURE = 1; + // Bounded buffering; when full, drop the oldest pending output. The + // producing process never stalls and the display stays near real time. + FLOW_CONTROL_DISCARD = 2; +} + message InitialPayload { optional bool jumphost = 1 [default = false]; repeated PortForwardSourceRequest reversetunnels = 2; map environmentvariables = 3; + optional FlowControlMode flow_control_mode = 4 [default = FLOW_CONTROL_NONE]; } message InitialResponse { @@ -75,6 +90,9 @@ message ConfigParams { message TermInit { repeated string environmentnames = 1; repeated string environmentvalues = 2; + // Tells etterminal whether the attached client opted into flow control, + // so it can shrink the kernel buffer on its socket to the server. + optional FlowControlMode flow_control_mode = 3 [default = FLOW_CONTROL_NONE]; } message TerminalUserInfo { diff --git a/src/base/PipeSocketHandler.cpp b/src/base/PipeSocketHandler.cpp index 509a05d65..0702f6188 100644 --- a/src/base/PipeSocketHandler.cpp +++ b/src/base/PipeSocketHandler.cpp @@ -144,4 +144,18 @@ void PipeSocketHandler::stopListening(const SocketEndpoint& endpoint) { FATAL_FAIL(::close(sockFd)); #endif } + +void PipeSocketHandler::minimizeKernelBuffering(int fd) { +#ifndef WIN32 + // Bound the kernel buffer on this unix socket. Terminal output waits + // here when the server applies backpressure (stops reading); the default + // (~200KB on Linux) is seconds of stale output on a slow link. 64KB does + // not limit throughput on a local socket. + int sndbuf = 64 * 1024; + if (setsockopt(fd, SOL_SOCKET, SO_SNDBUF, (char*)&sndbuf, sizeof(sndbuf)) < + 0) { + LOG(WARNING) << "Failed to set SO_SNDBUF: " << strerror(errno); + } +#endif +} } // namespace et diff --git a/src/base/PipeSocketHandler.hpp b/src/base/PipeSocketHandler.hpp index 2afb99e55..7a2199574 100644 --- a/src/base/PipeSocketHandler.hpp +++ b/src/base/PipeSocketHandler.hpp @@ -30,6 +30,8 @@ class PipeSocketHandler : public UnixSocketHandler { */ virtual void stopListening(const SocketEndpoint& endpoint); + virtual void minimizeKernelBuffering(int fd); + protected: /** @brief Tracks path -> listening socket descriptors for each pipe. */ map> pipeServerSockets; diff --git a/src/base/SocketHandler.hpp b/src/base/SocketHandler.hpp index 232c32969..168fb16ca 100644 --- a/src/base/SocketHandler.hpp +++ b/src/base/SocketHandler.hpp @@ -20,6 +20,15 @@ class SocketHandler { * descriptor. */ virtual bool hasData(int fd) = 0; + + /** + * @brief Shrinks the kernel's outgoing queue on fd so that pending data + * waits in application-level buffers, where flow control can apply + * backpressure or discard stale output. Only called for sessions that + * opt into flow control (--flow-control backpressure|discard); the + * default is a no-op so legacy sessions keep stock kernel buffering. + */ + virtual void minimizeKernelBuffering(int fd) {} /** * @brief Reads up to count bytes from fd. */ diff --git a/src/base/TcpSocketHandler.cpp b/src/base/TcpSocketHandler.cpp index da8a90ebe..a8feeb6ce 100644 --- a/src/base/TcpSocketHandler.cpp +++ b/src/base/TcpSocketHandler.cpp @@ -293,4 +293,23 @@ void TcpSocketHandler::initSocket(int fd) { fd, SOL_SOCKET, SO_LINGER, (const char*)&so_linger, sizeof so_linger)); } } + +void TcpSocketHandler::minimizeKernelBuffering(int fd) { +#ifdef TCP_NOTSENT_LOWAT + // Keep the kernel's not-yet-sent queue small so pending terminal output + // stays in application-level buffers, where flow control can apply + // backpressure or discard stale data. Without this, select()/poll() + // report the socket writable whenever the autotuned (multi-MB) send + // buffer has room, and on a slow link many seconds of stale output pile + // up in the kernel where they can no longer be dropped (issue #631). + // TCP_NOTSENT_LOWAT only limits *unsent* data; sent-but-unacked + // (in-flight) data is unaffected, so throughput on high-BDP links is + // preserved. Best effort: not supported on all platforms. + int lowat = 32 * 1024; + if (setsockopt(fd, IPPROTO_TCP, TCP_NOTSENT_LOWAT, (char*)&lowat, + sizeof(lowat)) < 0) { + LOG(WARNING) << "Failed to set TCP_NOTSENT_LOWAT: " << strerror(errno); + } +#endif +} } // namespace et diff --git a/src/base/TcpSocketHandler.hpp b/src/base/TcpSocketHandler.hpp index d81decf30..9f657eee9 100644 --- a/src/base/TcpSocketHandler.hpp +++ b/src/base/TcpSocketHandler.hpp @@ -31,6 +31,8 @@ class TcpSocketHandler : public UnixSocketHandler { */ virtual void stopListening(const SocketEndpoint& endpoint); + virtual void minimizeKernelBuffering(int fd); + protected: /** @brief Tracks all listening sockets created per TCP port. */ map> portServerSockets; diff --git a/src/base/WriteBuffer.hpp b/src/base/WriteBuffer.hpp new file mode 100644 index 000000000..1dd39c937 --- /dev/null +++ b/src/base/WriteBuffer.hpp @@ -0,0 +1,150 @@ +#ifndef __ET_WRITE_BUFFER__ +#define __ET_WRITE_BUFFER__ + +#include "Headers.hpp" + +namespace et { + +enum class WriteBufferMode { + BACKPRESSURE, // Stop accepting data when buffer is full + DISCARD // Drop oldest data when buffer is full +}; + +/** + * @brief Bounded buffer for pending write data, enabling flow control. + * + * Supports two modes: + * - BACKPRESSURE: canAcceptMore() returns false when full, causing callers + * to stop reading from the source. Processes stall when consumer is slow. + * - DISCARD: canAcceptMore() always returns true. When the buffer exceeds + * MAX_BUFFER_SIZE, oldest data is dropped. Processes never stall. + */ +class WriteBuffer { + public: + /** @brief Maximum bytes to buffer before applying backpressure or + * discarding. Every byte held here (or in any buffer downstream) is stale + * data the user must wait through on a slow link, so keep this small: at + * 100KB/s, 64KB is ~0.6s of output. It only needs to absorb short bursts + * between drain opportunities (reads are 16KB), not smooth over the + * network. This is a soft bound: canAcceptMore() admits a whole chunk + * whenever size() is below the limit, so the buffer can exceed it by up + * to one chunk. */ + static constexpr size_t MAX_BUFFER_SIZE = 64 * 1024; // 64KB + + explicit WriteBuffer(WriteBufferMode mode = WriteBufferMode::BACKPRESSURE) + : mode(mode), totalBytes(0), writeOffset(0) {} + + /** + * @brief Returns true if the buffer has room for more data. + * In BACKPRESSURE mode, returns false when buffer >= MAX_BUFFER_SIZE. + * In DISCARD mode, always returns true (old data will be dropped). + */ + bool canAcceptMore() const { + if (mode == WriteBufferMode::DISCARD) return true; + return totalBytes < MAX_BUFFER_SIZE; + } + + /** + * @brief Returns true if there is data waiting to be written. + */ + bool hasPendingData() const { return !pending.empty(); } + + /** + * @brief Returns the current amount of buffered data in bytes. + */ + size_t size() const { return totalBytes; } + + /** + * @brief Returns the buffer mode. + */ + WriteBufferMode getMode() const { return mode; } + + /** + * @brief Adds data to the end of the buffer. + * In DISCARD mode, drops oldest chunks if the buffer would exceed + * MAX_BUFFER_SIZE. + * @param data The data to enqueue. + */ + void enqueue(const string& data) { + if (data.empty()) return; + pending.push_back(data); + totalBytes += data.size(); + + if (mode == WriteBufferMode::DISCARD) { + discardOldest(); + } + } + + /** + * @brief Returns a pointer to the next bytes to write and the count. + * @param count Output: number of bytes available for writing. + * @return Pointer to the data, or nullptr if buffer is empty. + */ + const char* peekData(size_t* count) const { + if (pending.empty()) { + *count = 0; + return nullptr; + } + const string& front = pending.front(); + *count = front.size() - writeOffset; + return front.data() + writeOffset; + } + + /** + * @brief Removes bytesWritten from the front of the buffer. + * @param bytesWritten Number of bytes successfully written to socket. + */ + void consume(size_t bytesWritten) { + if (bytesWritten == 0) return; + + while (bytesWritten > 0 && !pending.empty()) { + string& front = pending.front(); + size_t available = front.size() - writeOffset; + + if (bytesWritten >= available) { + // Consumed the entire front chunk + bytesWritten -= available; + totalBytes -= available; + writeOffset = 0; + pending.pop_front(); + } else { + // Partial consumption + writeOffset += bytesWritten; + totalBytes -= bytesWritten; + bytesWritten = 0; + } + } + } + + /** + * @brief Clears all pending data. + */ + void clear() { + pending.clear(); + totalBytes = 0; + writeOffset = 0; + } + + private: + /** + * @brief Drops oldest chunks until buffer is within MAX_BUFFER_SIZE. + * Only called in DISCARD mode after enqueue. + */ + void discardOldest() { + while (totalBytes > MAX_BUFFER_SIZE && !pending.empty()) { + string& front = pending.front(); + size_t frontSize = front.size() - writeOffset; + totalBytes -= frontSize; + writeOffset = 0; + pending.pop_front(); + } + } + + WriteBufferMode mode; + std::deque pending; + size_t totalBytes; + size_t writeOffset; // Offset into the front chunk for partial writes +}; +} // namespace et + +#endif // __ET_WRITE_BUFFER__ diff --git a/src/terminal/TerminalClient.cpp b/src/terminal/TerminalClient.cpp index b3c777fb4..2df893b10 100644 --- a/src/terminal/TerminalClient.cpp +++ b/src/terminal/TerminalClient.cpp @@ -2,6 +2,7 @@ #include "TelemetryService.hpp" #include "TunnelUtils.hpp" +#include "WriteBuffer.hpp" namespace et { @@ -12,14 +13,20 @@ TerminalClient::TerminalClient( const string& passkey, shared_ptr _console, bool jumphost, const string& tunnels, const string& reverseTunnels, bool forwardSshAgent, const string& identityAgent, int _keepaliveDuration, - const vector>& envVars) + const vector>& envVars, + et::FlowControlMode _flowControlMode) : console(_console), shuttingDown(false), - keepaliveDuration(_keepaliveDuration) { + keepaliveDuration(_keepaliveDuration), + flowControlMode(_flowControlMode) { portForwardHandler = shared_ptr( new PortForwardHandler(_socketHandler, _pipeSocketHandler)); InitialPayload payload; payload.set_jumphost(jumphost); + if (flowControlMode != et::FLOW_CONTROL_NONE) { + // Left unset for NONE so the wire format matches old clients exactly. + payload.set_flow_control_mode(flowControlMode); + } for (const auto& envVar : envVars) { (*payload.mutable_environmentvariables())[envVar.first] = envVar.second; @@ -172,6 +179,14 @@ void TerminalClient::run(const string& command, const bool noexit) { TerminalInfo lastTerminalInfo; + const bool flowControlEnabled = (flowControlMode != et::FLOW_CONTROL_NONE); + // Flow control (opt-in): terminal output is staged here and written to the + // console at the top of each loop iteration. Unused when flow control is + // off: output is written straight to the console (legacy behavior). + WriteBuffer consoleOutputBuffer(flowControlMode == et::FLOW_CONTROL_DISCARD + ? WriteBufferMode::DISCARD + : WriteBufferMode::BACKPRESSURE); + if (!console.get()) { // NOTE: ../../scripts/ssh-et relies on the wording of this message, so if // you change it please update it as well. @@ -200,8 +215,12 @@ void TerminalClient::run(const string& command, const bool noexit) { } int clientFd = connection->getSocketFd(); if (clientFd > 0) { - FD_SET(clientFd, &rfd); - maxfd = max(maxfd, clientFd); + // Only read from the server if the console output buffer has room; + // this propagates backpressure when the console is slow. + if (!flowControlEnabled || consoleOutputBuffer.canAcceptMore()) { + FD_SET(clientFd, &rfd); + maxfd = max(maxfd, clientFd); + } } // Include port forward sockets in select for low-latency forwarding. set pfFds; @@ -215,6 +234,24 @@ void TerminalClient::run(const string& command, const bool noexit) { select(maxfd + 1, &rfd, NULL, NULL, &tv); try { + // First, drain buffered terminal output to the console. Coalesce all + // pending chunks into a single write to avoid intermediate renders + // (flicker), matching the non-buffered path below. + if (console && consoleOutputBuffer.hasPendingData()) { + string pending; + size_t count; + const char* data; + while ((data = consoleOutputBuffer.peekData(&count)) != nullptr && + count > 0) { + pending.append(data, count); + consoleOutputBuffer.consume(count); + } + if (!pending.empty()) { + // May block if the console is slow; bounded by the buffer size. + console->write(pending); + } + } + if (console) { // Check for data to send. if (FD_ISSET(consoleFd, &rfd)) { @@ -287,7 +324,12 @@ void TerminalClient::run(const string& command, const bool noexit) { // arriving in one write followed by the repaint in the next), which // produces visible flicker. string coalesced; - while (connection->hasData()) { + // Stop pulling packets once the console buffer is full: in + // backpressure mode it would otherwise grow past its bound by + // however much the kernel had queued. Unprocessed packets stay in + // the socket buffer until the next iteration. + while (connection->hasData() && + (!flowControlEnabled || consoleOutputBuffer.canAcceptMore())) { VLOG(4) << "connection has data"; Packet packet; if (!connection->read(&packet)) { @@ -326,17 +368,26 @@ void TerminalClient::run(const string& command, const bool noexit) { } } if (console && !coalesced.empty()) { - console->write(coalesced); + if (flowControlEnabled) { + // Staged for the flow-controlled drain at the top of the loop + consoleOutputBuffer.enqueue(coalesced); + } else { + console->write(coalesced); + } } } if (clientFd > 0 && keepaliveTime < time(NULL)) { keepaliveTime = time(NULL) + keepaliveDuration; - if (waitingOnKeepalive) { + // While the console buffer is full we are intentionally not reading + // the connection, so the keepalive echo may be sitting unread in the + // socket; don't treat that as a dead connection. + if (waitingOnKeepalive && + !(flowControlEnabled && consoleOutputBuffer.hasPendingData())) { LOG(INFO) << "Missed a keepalive, killing connection."; connection->closeSocketAndMaybeReconnect(); waitingOnKeepalive = false; - } else { + } else if (!waitingOnKeepalive) { LOG(INFO) << "Writing keepalive packet"; connection->writePacket(Packet(TerminalPacketType::KEEP_ALIVE, "")); waitingOnKeepalive = true; diff --git a/src/terminal/TerminalClient.hpp b/src/terminal/TerminalClient.hpp index a56b635f8..9906af129 100644 --- a/src/terminal/TerminalClient.hpp +++ b/src/terminal/TerminalClient.hpp @@ -32,7 +32,8 @@ class TerminalClient { bool jumphost, const string& tunnels, const string& reverseTunnels, bool forwardSshAgent, const string& identityAgent, int _keepaliveDuration, - const vector>& envVars); + const vector>& envVars, + et::FlowControlMode _flowControlMode = et::FLOW_CONTROL_NONE); /** @brief Tears down the client, closing sockets and stopping background * threads. */ virtual ~TerminalClient(); @@ -60,6 +61,8 @@ class TerminalClient { recursive_mutex shutdownMutex; /** @brief Keepalive interval (seconds) sent to the server. */ int keepaliveDuration; + /** @brief Flow control mode for output buffering. */ + et::FlowControlMode flowControlMode; }; } // namespace et diff --git a/src/terminal/TerminalClientMain.cpp b/src/terminal/TerminalClientMain.cpp index 570bfe806..5f0310ecd 100644 --- a/src/terminal/TerminalClientMain.cpp +++ b/src/terminal/TerminalClientMain.cpp @@ -185,7 +185,16 @@ int main(int argc, char** argv) { "If set, communicate to etserver on the matching fifo name", cxxopts::value()->default_value("")) // ("ssh-option", "Options to pass down to `ssh -o`", - cxxopts::value>()); + cxxopts::value>()) // + ("flow-control", + "Flow control mode: 'none' (default; legacy behavior, unchanged), " + "'backpressure' (lossless: bounded, tuned buffers keep Ctrl-C " + "responsive on a slow link; when the client can't keep up the " + "remote process is paused, like plain ssh), or 'discard' (drop " + "the oldest pending output so the remote process never stalls " + "and the display stays close to real time; not for consumers " + "that require every byte, e.g. tmux -CC)", + cxxopts::value()->default_value("none")); options.parse_positional({"host"}); auto result = options.parse(argc, argv); @@ -439,10 +448,24 @@ int main(int argc, char** argv) { jServerFifo, result.count("x") > 0, result["verbose"].as(), etterminal_path, serverFifo, ssh_options); + et::FlowControlMode flowControlMode = et::FLOW_CONTROL_NONE; + string flowControlStr = result["flow-control"].as(); + if (flowControlStr == "backpressure") { + flowControlMode = et::FLOW_CONTROL_BACKPRESSURE; + } else if (flowControlStr == "discard") { + flowControlMode = et::FLOW_CONTROL_DISCARD; + } else if (flowControlStr != "none") { + CLOG(INFO, "stdout") << "Invalid flow-control mode: " << flowControlStr + << ". Must be 'none', 'backpressure', or 'discard'." + << endl; + exit(1); + } + TerminalClient terminalClient( clientSocket, clientPipeSocket, socketEndpoint, idpasskeypair.first, idpasskeypair.second, console, is_jumphost, tunnel_arg, r_tunnel_arg, - forwardAgent, sshSocket, keepaliveDuration, sshConfigOptions.env_vars); + forwardAgent, sshSocket, keepaliveDuration, sshConfigOptions.env_vars, + flowControlMode); terminalClient.run( result.count("command") ? result["command"].as() : "", result.count("noexit")); diff --git a/src/terminal/TerminalServer.cpp b/src/terminal/TerminalServer.cpp index b86ac1e09..9ba035315 100644 --- a/src/terminal/TerminalServer.cpp +++ b/src/terminal/TerminalServer.cpp @@ -2,6 +2,7 @@ #include "TerminalServer.hpp" #include "TelemetryService.hpp" +#include "WriteBuffer.hpp" #define BUF_SIZE (16 * 1024) @@ -124,6 +125,21 @@ void TerminalServer::runJumpHost( terminalFd, Packet(TerminalPacketType::JUMPHOST_INIT, protoToString(payload))); + // Flow control (only when the client opted in): queue of pending jumphost + // packets for the client. + const et::FlowControlMode jumphostMode = payload.flow_control_mode(); + const bool jumphostFlowControlEnabled = + (jumphostMode != et::FLOW_CONTROL_NONE); + const bool jumphostDiscard = (jumphostMode == et::FLOW_CONTROL_DISCARD); + std::deque pendingPackets; + size_t pendingBytes = 0; + // Bytes of TERMINAL_BUFFER packets in pendingPackets. Only those are + // droppable in discard mode; when the queue is dominated by non-droppable + // control packets (e.g. port-forward data), reads must stop so the queue + // stays bounded. + size_t droppableBytes = 0; + const size_t MAX_PENDING_BYTES = WriteBuffer::MAX_BUFFER_SIZE; + while (true) { { lock_guard guard(terminalThreadMutex); @@ -132,33 +148,103 @@ void TerminalServer::runJumpHost( } } - fd_set rfd; + fd_set rfd, wfd; timeval tv; FD_ZERO(&rfd); + FD_ZERO(&wfd); int maxfd = -1; // Only drain the terminal while the client connection can absorb the // data, so backpressure reaches the terminal instead of this loop - // blocking inside writePacket() - if (serverClientState->canBufferWrite(2 * BUF_SIZE)) { + // blocking inside writePacket(). In flow-control modes the bounded + // pending queue gates reads instead; in discard mode room can always be + // made by dropping old terminal output, so only the non-droppable + // (control) backlog gates reading. + bool readTerminal; + if (!jumphostFlowControlEnabled) { + readTerminal = serverClientState->canBufferWrite(2 * BUF_SIZE); + } else if (jumphostDiscard) { + readTerminal = (pendingBytes - droppableBytes) < MAX_PENDING_BYTES; + } else { + readTerminal = pendingBytes < MAX_PENDING_BYTES; + } + if (readTerminal) { FD_SET(terminalFd, &rfd); maxfd = terminalFd; } int serverClientFd = serverClientState->getSocketFd(); if (serverClientFd > 0) { + if (jumphostFlowControlEnabled) { + // Reapply every iteration; see runTerminal for why fd tracking is + // not reliable across reconnects. + getSocketHandler()->minimizeKernelBuffering(serverClientFd); + } FD_SET(serverClientFd, &rfd); maxfd = max(maxfd, serverClientFd); + + // Wake as soon as the socket can take more pending packets. + if (!pendingPackets.empty()) { + FD_SET(serverClientFd, &wfd); + } } tv.tv_sec = 0; tv.tv_usec = 100000; - select(maxfd + 1, &rfd, NULL, NULL, &tv); + if (select(maxfd + 1, &rfd, &wfd, NULL, &tv) < 0 && errno == EINTR) { + // See runTerminal: retry only on EINTR; other errors fall through so + // the read/write paths surface the dead fd. + continue; + } try { + // Drain pending packets while the socket can take more + if (serverClientFd > 0 && FD_ISSET(serverClientFd, &wfd) && + !pendingPackets.empty()) { + while (!pendingPackets.empty()) { + serverClientState->writePacket(pendingPackets.front()); + pendingBytes -= pendingPackets.front().length(); + if (pendingPackets.front().getHeader() == + TerminalPacketType::TERMINAL_BUFFER) { + droppableBytes -= pendingPackets.front().length(); + } + pendingPackets.pop_front(); + + if (!isSocketWritable(serverClientFd)) { + break; // Kernel queue is full enough; stop draining + } + } + } + if (FD_ISSET(terminalFd, &rfd)) { try { Packet packet; if (terminalSocketHandler->readPacket(terminalFd, &packet)) { - serverClientState->writePacket(packet); + if (!jumphostFlowControlEnabled) { + serverClientState->writePacket(packet); + } else { + pendingPackets.push_back(packet); + pendingBytes += packet.length(); + if (packet.getHeader() == TerminalPacketType::TERMINAL_BUFFER) { + droppableBytes += packet.length(); + } + + // In discard mode, drop the oldest droppable packets when + // over the limit. Only terminal output is safe to drop: + // control packets (port forwarding, responses, keepalives) + // must be delivered or the protocol state desyncs. + if (jumphostDiscard) { + auto it = pendingPackets.begin(); + while (pendingBytes > MAX_PENDING_BYTES && + it != pendingPackets.end()) { + if (it->getHeader() == TerminalPacketType::TERMINAL_BUFFER) { + pendingBytes -= it->length(); + droppableBytes -= it->length(); + it = pendingPackets.erase(it); + } else { + ++it; + } + } + } + } } } catch (const std::runtime_error& ex) { LOG(INFO) << "Terminal session ended" << ex.what(); @@ -263,7 +349,13 @@ void TerminalServer::runTerminal( shared_ptr terminalSocketHandler = terminalRouter->getSocketHandler(); + const et::FlowControlMode flowControlMode = payload.flow_control_mode(); + const bool flowControlEnabled = (flowControlMode != et::FLOW_CONTROL_NONE); + TermInit termInit; + if (flowControlEnabled) { + termInit.set_flow_control_mode(flowControlMode); + } for (auto& it : environmentVariables) { *(termInit.add_environmentnames()) = it.first; *(termInit.add_environmentvalues()) = it.second; @@ -272,6 +364,14 @@ void TerminalServer::runTerminal( terminalFd, Packet(TerminalPacketType::TERMINAL_INIT, protoToString(termInit))); + // Flow control (opt-in): terminal output is staged here and drained only + // while the socket reports writable, so the backlog stays where it can be + // gated (backpressure) or dropped (discard) instead of piling up in the + // kernel. Unused when the client didn't opt in (FLOW_CONTROL_NONE). + WriteBuffer terminalOutputBuffer(flowControlMode == et::FLOW_CONTROL_DISCARD + ? WriteBufferMode::DISCARD + : WriteBufferMode::BACKPRESSURE); + while (run) { { lock_guard guard(terminalThreadMutex); @@ -282,22 +382,38 @@ void TerminalServer::runTerminal( // Data structures needed for select() and // non-blocking I/O. - fd_set rfd; + fd_set rfd, wfd; timeval tv; FD_ZERO(&rfd); + FD_ZERO(&wfd); int maxfd = -1; // Only drain the terminal while the client connection can absorb the // data, so backpressure reaches the shell instead of this loop blocking - // inside writePacket() - if (serverClientState->canBufferWrite(2 * BUF_SIZE)) { + // inside writePacket(). In flow-control modes the bounded WriteBuffer + // gates reads instead (in discard mode it always has room: old output + // is dropped). + if (flowControlEnabled ? terminalOutputBuffer.canAcceptMore() + : serverClientState->canBufferWrite(2 * BUF_SIZE)) { FD_SET(terminalFd, &rfd); maxfd = terminalFd; } int serverClientFd = serverClientState->getSocketFd(); if (serverClientFd > 0) { + if (flowControlEnabled) { + // Reapply every iteration: the connection gets a brand-new socket + // on reconnect, kernel tuning is per-socket, and fd numbers are + // reused, so there is no reliable way to detect the swap from the + // fd alone. The setsockopt is idempotent and costs ~1us. + serverSocketHandler->minimizeKernelBuffering(serverClientFd); + } FD_SET(serverClientFd, &rfd); maxfd = max(maxfd, serverClientFd); + + // Wake as soon as the socket can take more buffered output. + if (flowControlEnabled && terminalOutputBuffer.hasPendingData()) { + FD_SET(serverClientFd, &wfd); + } } // Include port forward sockets in select for low-latency forwarding. set pfFds; @@ -308,9 +424,39 @@ void TerminalServer::runTerminal( } tv.tv_sec = 0; tv.tv_usec = 100000; - select(maxfd + 1, &rfd, NULL, NULL, &tv); + if (select(maxfd + 1, &rfd, &wfd, NULL, &tv) < 0 && errno == EINTR) { + // Interrupted by a signal: the fd sets are unspecified, so + // re-evaluate rather than acting on them. Other errors (e.g. a fd + // closed by another thread) fall through: the read/write paths then + // surface the dead fd as a session-ending error. + continue; + } try { + // First, drain buffered terminal output while the kernel's unsent + // queue is below the low-water mark (TCP_NOTSENT_LOWAT). + if (flowControlEnabled && serverClientFd > 0 && + FD_ISSET(serverClientFd, &wfd) && + terminalOutputBuffer.hasPendingData()) { + while (terminalOutputBuffer.hasPendingData()) { + size_t count; + const char* data = terminalOutputBuffer.peekData(&count); + if (data == nullptr || count == 0) break; + + et::TerminalBuffer tb; + tb.set_buffer(string(data, count)); + VLOG(2) << "Draining buffered bytes to client: " << count << " " + << serverClientState->getWriter()->getSequenceNumber(); + serverClientState->writePacket( + Packet(TerminalPacketType::TERMINAL_BUFFER, protoToString(tb))); + terminalOutputBuffer.consume(count); + + if (!isSocketWritable(serverClientFd)) { + break; // Kernel queue is full enough; stop draining + } + } + } + // Check for data to receive; the received // data includes also the data previously sent // on the same master descriptor (line 90). @@ -322,10 +468,15 @@ void TerminalServer::runTerminal( VLOG(2) << "Sending bytes from terminal: " << rc << " " << serverClientState->getWriter()->getSequenceNumber(); string s(b, rc); - et::TerminalBuffer tb; - tb.set_buffer(s); - serverClientState->writePacket( - Packet(TerminalPacketType::TERMINAL_BUFFER, protoToString(tb))); + if (flowControlEnabled) { + // Stage for flow-controlled draining (above) + terminalOutputBuffer.enqueue(s); + } else { + et::TerminalBuffer tb; + tb.set_buffer(s); + serverClientState->writePacket( + Packet(TerminalPacketType::TERMINAL_BUFFER, protoToString(tb))); + } } else if (rc == 0) { LOG(INFO) << "Terminal session ended"; run = false; diff --git a/src/terminal/UserTerminalHandler.cpp b/src/terminal/UserTerminalHandler.cpp index c148e2de6..011a223a3 100644 --- a/src/terminal/UserTerminalHandler.cpp +++ b/src/terminal/UserTerminalHandler.cpp @@ -59,6 +59,11 @@ void UserTerminalHandler::run() { setenv(ti.environmentnames(a).c_str(), ti.environmentvalues(a).c_str(), true); } + if (ti.flow_control_mode() != et::FLOW_CONTROL_NONE) { + // The client opted into flow control: shrink the kernel buffer on the + // etterminal->etserver hop so backpressure holds less stale output. + socketHandler->minimizeKernelBuffering(routerFd); + } break; } diff --git a/test/integration_tests/TerminalServerTest.cpp b/test/integration_tests/TerminalServerTest.cpp index 5adab3a2c..63dcdd5b9 100644 --- a/test/integration_tests/TerminalServerTest.cpp +++ b/test/integration_tests/TerminalServerTest.cpp @@ -1,4 +1,5 @@ #include +#include #include "FakeConsole.hpp" #include "FakeSshSetupHandler.hpp" @@ -479,6 +480,53 @@ TEST_CASE_METHOD(ServerEndToEndTestFixture, "ServerDataTransferTest", sshSetupHandler->shutdownHandler(); } +// Sessions that opt into flow control (backpressure or discard) must +// still deliver data intact in both directions. The volumes here stay far +// below WriteBuffer::MAX_BUFFER_SIZE, so even discard mode must not drop +// anything. +TEST_CASE_METHOD(ServerEndToEndTestFixture, "ServerFlowControlDataTransferTest", + "[ServerFlowControlDataTransferTest][integration]") { + const et::FlowControlMode mode = + GENERATE(et::FLOW_CONTROL_BACKPRESSURE, et::FLOW_CONTROL_DISCARD); + CAPTURE(mode); + + auto [id, passkey] = sshSetupHandler->SetupSsh( + "", "localhost", "localhost", 2022, "", "", false, 0, "", "", {}); + + sleep(1); + + shared_ptr terminalClient(new TerminalClient( + clientSocketHandler, clientPipeSocketHandler, serverEndpoint, id, passkey, + fakeConsole, false, "", "", false, "", MAX_CLIENT_KEEP_ALIVE_DURATION, {}, + mode)); + thread terminalClientThread( + [terminalClient]() { terminalClient->run("", false); }); + sleep(3); + + // Keyboard input path (console -> terminal) + string s = "flow_control_input"; + for (size_t a = 0; a < s.size(); a++) { + fakeConsole->simulateKeystrokes(string(1, s[a])); + } + string resultConcat; + for (size_t a = 0; a < s.size(); a++) { + resultConcat = resultConcat.append(fakeUserTerminal->getKeystrokes(1)); + } + REQUIRE(resultConcat == s); + + // Terminal output path (terminal -> console); this is the direction the + // flow-control buffering applies to. + string out = "flow_control_output"; + fakeUserTerminal->simulateTerminalResponse(out); + REQUIRE(fakeConsole->getTerminalData(out.length()) == out); + + terminalClient->shutdown(); + terminalClientThread.join(); + terminalClient.reset(); + + sshSetupHandler->shutdownHandler(); +} + TEST_CASE_METHOD(ServerEndToEndTestFixture, "ServerJumphostTest", "[ServerJumphostTest][integration]") { // The destination terminal is already set up in the fixture at diff --git a/test/unit_tests/WriteBufferTest.cpp b/test/unit_tests/WriteBufferTest.cpp new file mode 100644 index 000000000..5ee6fa266 --- /dev/null +++ b/test/unit_tests/WriteBufferTest.cpp @@ -0,0 +1,181 @@ +#include "TestHeaders.hpp" +#include "WriteBuffer.hpp" + +using namespace et; + +TEST_CASE("WriteBuffer basic operations", "[WriteBuffer]") { + WriteBuffer buffer; + + SECTION("Empty buffer state") { + REQUIRE(buffer.canAcceptMore() == true); + REQUIRE(buffer.hasPendingData() == false); + REQUIRE(buffer.size() == 0); + + size_t count; + const char* data = buffer.peekData(&count); + REQUIRE(data == nullptr); + REQUIRE(count == 0); + } + + SECTION("Enqueue and peek") { + buffer.enqueue("hello"); + REQUIRE(buffer.hasPendingData() == true); + REQUIRE(buffer.size() == 5); + + size_t count; + const char* data = buffer.peekData(&count); + REQUIRE(data != nullptr); + REQUIRE(count == 5); + REQUIRE(string(data, count) == "hello"); + } + + SECTION("Consume partial") { + buffer.enqueue("hello"); + buffer.consume(2); + REQUIRE(buffer.size() == 3); + + size_t count; + const char* data = buffer.peekData(&count); + REQUIRE(count == 3); + REQUIRE(string(data, count) == "llo"); + } + + SECTION("Consume full chunk") { + buffer.enqueue("hello"); + buffer.enqueue("world"); + REQUIRE(buffer.size() == 10); + + buffer.consume(5); + REQUIRE(buffer.size() == 5); + + size_t count; + const char* data = buffer.peekData(&count); + REQUIRE(count == 5); + REQUIRE(string(data, count) == "world"); + } + + SECTION("Consume across chunks") { + buffer.enqueue("abc"); + buffer.enqueue("defgh"); + REQUIRE(buffer.size() == 8); + + buffer.consume(5); // Consumes "abc" + "de" + REQUIRE(buffer.size() == 3); + + size_t count; + const char* data = buffer.peekData(&count); + REQUIRE(count == 3); + REQUIRE(string(data, count) == "fgh"); + } + + SECTION("Clear buffer") { + buffer.enqueue("hello"); + buffer.enqueue("world"); + buffer.clear(); + + REQUIRE(buffer.hasPendingData() == false); + REQUIRE(buffer.size() == 0); + REQUIRE(buffer.canAcceptMore() == true); + } + + SECTION("Empty string enqueue is ignored") { + buffer.enqueue(""); + REQUIRE(buffer.hasPendingData() == false); + REQUIRE(buffer.size() == 0); + } +} + +TEST_CASE("WriteBuffer backpressure", "[WriteBuffer]") { + WriteBuffer buffer(WriteBufferMode::BACKPRESSURE); + + SECTION("canAcceptMore returns false when buffer is full") { + // Fill the buffer to capacity + string largeChunk(WriteBuffer::MAX_BUFFER_SIZE, 'x'); + buffer.enqueue(largeChunk); + + REQUIRE(buffer.canAcceptMore() == false); + REQUIRE(buffer.size() == WriteBuffer::MAX_BUFFER_SIZE); + + // Consume some data + buffer.consume(1024); + REQUIRE(buffer.canAcceptMore() == true); + } + + SECTION("default mode is backpressure") { + WriteBuffer defaultBuffer; + REQUIRE(defaultBuffer.getMode() == WriteBufferMode::BACKPRESSURE); + } +} + +TEST_CASE("WriteBuffer discard mode", "[WriteBuffer]") { + WriteBuffer buffer(WriteBufferMode::DISCARD); + + SECTION("canAcceptMore always returns true") { + string largeChunk(WriteBuffer::MAX_BUFFER_SIZE, 'x'); + buffer.enqueue(largeChunk); + + // In discard mode, canAcceptMore is always true + REQUIRE(buffer.canAcceptMore() == true); + } + + SECTION("old data is discarded when buffer overflows") { + // Fill with chunk A (128KB) + const size_t halfSize = WriteBuffer::MAX_BUFFER_SIZE / 2; + string chunkA(halfSize, 'A'); + buffer.enqueue(chunkA); + REQUIRE(buffer.size() == halfSize); + + // Fill with chunk B (128KB) -- total now equals MAX + string chunkB(halfSize, 'B'); + buffer.enqueue(chunkB); + REQUIRE(buffer.size() == WriteBuffer::MAX_BUFFER_SIZE); + + // Add chunk C (128KB) -- should discard chunk A + string chunkC(halfSize, 'C'); + buffer.enqueue(chunkC); + + // Buffer should be within MAX_BUFFER_SIZE + REQUIRE(buffer.size() <= WriteBuffer::MAX_BUFFER_SIZE); + + // The oldest data (chunk A) should be gone + // The front of the buffer should now be chunk B + size_t count; + const char* data = buffer.peekData(&count); + REQUIRE(data != nullptr); + REQUIRE(data[0] == 'B'); + } + + SECTION("buffer stays bounded with many enqueues") { + const size_t chunkSize = 4096; + // Enqueue much more than MAX_BUFFER_SIZE + for (size_t i = 0; i < WriteBuffer::MAX_BUFFER_SIZE * 4; i += chunkSize) { + string chunk(chunkSize, 'A' + ((i / chunkSize) % 26)); + buffer.enqueue(chunk); + // Buffer should never exceed MAX_BUFFER_SIZE + one chunk + REQUIRE(buffer.size() <= WriteBuffer::MAX_BUFFER_SIZE + chunkSize); + } + } + + SECTION("newest data is preserved") { + // Fill buffer well past capacity + const size_t chunkSize = 1024; + string lastChunk; + for (size_t i = 0; i < WriteBuffer::MAX_BUFFER_SIZE * 2; i += chunkSize) { + lastChunk = string(chunkSize, 'A' + ((i / chunkSize) % 26)); + buffer.enqueue(lastChunk); + } + + // Drain the buffer and verify the last chunk is in the output + string allData; + while (buffer.hasPendingData()) { + size_t count; + const char* data = buffer.peekData(&count); + allData.append(data, count); + buffer.consume(count); + } + + // The tail of the drained data should be our last chunk + REQUIRE(allData.size() >= lastChunk.size()); + REQUIRE(allData.substr(allData.size() - lastChunk.size()) == lastChunk); + } +} From 201fdb4cb2494d6727e5290b5de2b86b4ff8e5c7 Mon Sep 17 00:00:00 2001 From: Ben Maurer Date: Fri, 10 Jul 2026 20:53:25 -0700 Subject: [PATCH 3/3] Add E2E flow-control test harness A userspace throttle proxy (100KB/s, with a clamped receive buffer so it models a real slow link instead of absorbing megabytes into its own autotuned rcvbuf) sits between etserver and the et client, with a timestamp-printing workload behind a PTY driven from tmux. - do_scenario.sh: per-mode scenario (trunk/none/backpressure/discard) measuring display lag, producer lag (is the process stalled?), throughput, and Ctrl-C-to-prompt latency, with an optional 60s disconnect simulation. Emits JSONL metrics. Refuses to run on a dirty tree because its cleanup restores source dirs from HEAD. - throughput_test.sh: bulk throughput on a fast link (no throttle). - REPORT.md: methodology and measured results. To run the components standalone, et gains --idpasskey to skip the SSH handshake and connect directly to a manually started etterminal, and etterminal's --idpasskey handling is fixed to actually use the passed value (it was assigned to a shadowing local and dropped). --- src/terminal/TerminalClientMain.cpp | 31 ++- src/terminal/TerminalMain.cpp | 2 +- test/e2e/README.md | 141 ++++++++++++++ test/e2e/REPORT.md | 165 ++++++++++++++++ test/e2e/do_scenario.sh | 285 ++++++++++++++++++++++++++++ test/e2e/generate_report.py | 194 +++++++++++++++++++ test/e2e/print_timestamps.py | 27 +++ test/e2e/run_all_scenarios.sh | 56 ++++++ test/e2e/run_e2e_test.sh | 102 ++++++++++ test/e2e/throttle_proxy.py | 159 ++++++++++++++++ test/e2e/throughput_test.sh | 72 +++++++ 11 files changed, 1225 insertions(+), 9 deletions(-) create mode 100644 test/e2e/README.md create mode 100644 test/e2e/REPORT.md create mode 100644 test/e2e/do_scenario.sh create mode 100644 test/e2e/generate_report.py create mode 100644 test/e2e/print_timestamps.py create mode 100644 test/e2e/run_all_scenarios.sh create mode 100644 test/e2e/run_e2e_test.sh create mode 100644 test/e2e/throttle_proxy.py create mode 100644 test/e2e/throughput_test.sh diff --git a/src/terminal/TerminalClientMain.cpp b/src/terminal/TerminalClientMain.cpp index 5f0310ecd..ced425ba0 100644 --- a/src/terminal/TerminalClientMain.cpp +++ b/src/terminal/TerminalClientMain.cpp @@ -186,6 +186,10 @@ int main(int argc, char** argv) { cxxopts::value()->default_value("")) // ("ssh-option", "Options to pass down to `ssh -o`", cxxopts::value>()) // + ("idpasskey", + "If set, skip SSH and use this id/passkey directly (format: " + "id/passkey). Start etterminal manually before connecting.", + cxxopts::value()) // ("flow-control", "Flow control mode: 'none' (default; legacy behavior, unchanged), " "'backpressure' (lossless: bounded, tuned buffers keep Ctrl-C " @@ -300,7 +304,7 @@ int main(int argc, char** argv) { exit(0); } - { + if (!result.count("idpasskey")) { char* home_dir = ssh_get_user_home_dir(); const char* host_from_command = destinationHost.c_str(); ssh_options_set(&sshConfigOptions, SSH_OPTIONS_HOST, @@ -378,7 +382,7 @@ int main(int argc, char** argv) { shared_ptr clientSocket(new TcpSocketHandler()); shared_ptr clientPipeSocket(new PipeSocketHandler()); - if (!ping(socketEndpoint, clientSocket)) { + if (!result.count("idpasskey") && !ping(socketEndpoint, clientSocket)) { CLOG(INFO, "stdout") << "Could not reach the ET server: " << socketEndpoint.name() << ":" << socketEndpoint.port() << endl; @@ -441,12 +445,23 @@ int main(int argc, char** argv) { } } - auto subprocessUtils = make_shared(); - SshSetupHandler sshSetupHandler(subprocessUtils); - pair idpasskeypair = sshSetupHandler.SetupSsh( - username, destinationHost, host_alias, destinationPort, jumphost, - jServerFifo, result.count("x") > 0, result["verbose"].as(), - etterminal_path, serverFifo, ssh_options); + pair idpasskeypair; + if (result.count("idpasskey")) { + auto tokens = split(result["idpasskey"].as(), '/'); + if (tokens.size() != 2 || tokens[0].empty() || tokens[1].empty()) { + CLOG(INFO, "stdout") + << "Invalid --idpasskey format. Expected: id/passkey" << endl; + exit(1); + } + idpasskeypair = {tokens[0], tokens[1]}; + } else { + auto subprocessUtils = make_shared(); + SshSetupHandler sshSetupHandler(subprocessUtils); + idpasskeypair = sshSetupHandler.SetupSsh( + username, destinationHost, host_alias, destinationPort, jumphost, + jServerFifo, result.count("x") > 0, result["verbose"].as(), + etterminal_path, serverFifo, ssh_options); + } et::FlowControlMode flowControlMode = et::FLOW_CONTROL_NONE; string flowControlStr = result["flow-control"].as(); diff --git a/src/terminal/TerminalMain.cpp b/src/terminal/TerminalMain.cpp index 173f27cfa..1feed6d31 100644 --- a/src/terminal/TerminalMain.cpp +++ b/src/terminal/TerminalMain.cpp @@ -124,7 +124,7 @@ int main(int argc, char** argv) { STFATAL << "Invalid number of tokens: " << tokens.size(); } } else { - string idpasskey = result["idpasskey"].as(); + idpasskey = result["idpasskey"].as(); if (result.count("idpasskeyfile")) { // Check for passkey file std::ifstream t(result["idpasskeyfile"].as().c_str()); diff --git a/test/e2e/README.md b/test/e2e/README.md new file mode 100644 index 000000000..d9ac6fae5 --- /dev/null +++ b/test/e2e/README.md @@ -0,0 +1,141 @@ +# End-to-End Flow Control Testing + +This directory contains tools for testing ET's flow control behavior with a +simulated slow network. + +## Overview + +The test uses a TCP throttle proxy between the ET client and server to simulate +a bandwidth-limited network. The proxy applies TCP backpressure to the server +(reads slowly), which forces the server's flow control to kick in. + +``` +print_timestamps.py -> PTY -> etterminal -> etserver -> [throttle proxy] -> et client -> terminal + (100KB/s) +``` + +The `--idpasskey` flag on both `et` and `etterminal` bypasses SSH, so you can +run all three processes on the same machine without SSH access. + +## Prerequisites + +Build ET first: + +```bash +cd build && cmake .. && ninja +``` + +## Quick Start + +Run the test script (must run outside process supervisors that kill children, +e.g. via cron or a dedicated terminal): + +```bash +# From the repo root: +bash test/e2e/run_e2e_test.sh discard 100000 +``` + +The script starts etserver (port 4444), a throttle proxy (port 4445 -> 4444 at +100KB/s), etterminal, and the et client in a tmux session. It then runs +`print_timestamps.py` and measures: + +- **display_lag**: how far behind the timestamps on screen are from real time +- **process_lag**: how far behind the process producing output is from real time + (0 = running freely, >1 = stalled by backpressure) + +## Manual Setup + +If you prefer to run each component in a separate terminal: + +```bash +# Generate credentials (id must be 16 chars, key must be 32 chars) +ID="TestID0123456789" +KEY="TestPasskey0123456789012345678AB" + +# Terminal 1: etserver +./build/etserver --serverfifo=/tmp/test_et.fifo --port=4444 + +# Terminal 2: throttle proxy (100KB/s) +python3 test/e2e/throttle_proxy.py 4445 4444 100000 + +# Terminal 3: etterminal +./build/etterminal --idpasskey="$ID/$KEY" --serverfifo=/tmp/test_et.fifo + +# Terminal 4: et client +./build/et --idpasskey="$ID/$KEY" 127.0.0.1:4445 --flow-control discard + +# Inside the ET session, run: +python3 test/e2e/print_timestamps.py +``` + +Then compare the timestamps on screen with `date` to measure lag. + +## What to Expect + +### Without flow control (baseline, before this PR) + +The server reads from the PTY and writes directly to the TCP socket. When the +proxy is slow, `writePacket()` blocks, which blocks PTY reads, which stalls the +process. Data already in the TCP kernel buffer is "old" and gets delivered to the +client over time. + +- **display_lag**: grows linearly (~1s per second) +- **process_lag**: grows (process stalls) +- **Ctrl-C**: may be slow because old data must drain first + +### With `--flow-control backpressure` + +Same as baseline: when the 256KB WriteBuffer fills and the socket is not +writable, the server stops reading from the PTY. The process stalls. + +- **display_lag**: grows (bounded by TCP buffer + WriteBuffer = ~400KB) +- **process_lag**: grows (process stalls when buffer fills) +- **Ctrl-C**: responsive because buffered data is bounded + +### With `--flow-control discard` + +The WriteBuffer always accepts data and discards oldest chunks when full. The +server keeps reading from the PTY even when the socket is slow. The process +never stalls. Old data is discarded, so when the socket becomes writable, the +server sends *recent* data. + +- **display_lag**: grows (bounded by TCP buffer, smaller than backpressure) +- **process_lag**: near zero (process runs freely) +- **Ctrl-C**: responsive + +## Throttle Proxy Details + +`throttle_proxy.py` is a TCP proxy that rate-limits the server→client direction +by reading from the server socket at a controlled rate. This causes the kernel +TCP send buffer on the server side to fill up, which makes the server's +`writePacket()` calls block — exactly what happens on a slow network. + +The client→server direction (keystrokes, Ctrl-C) is forwarded at full speed. + +The proxy binds on `[::]` with `IPV6_V6ONLY=0` (dual-stack) to accept both IPv4 +and IPv6-mapped IPv4 connections, which is required because ET's +`TcpSocketHandler::connect()` may use either address family depending on +`getaddrinfo()` results. + +## Running via Cron (for automated environments) + +If your environment kills background processes (e.g. Claude Code), schedule the +test via cron: + +```bash +echo "* * * * * bash /path/to/test/e2e/run_e2e_test.sh discard 100000 > /tmp/e2e_result.log 2>&1" | crontab - +# Wait ~2 minutes, then: +cat /tmp/e2e_result.log +crontab -r +``` + +## Troubleshooting + +- **"proxy saw no connection"**: ET connected to a different port (likely the + system etserver on port 2022). Make sure `--idpasskey` is being used so SSH + config parsing is skipped. +- **etserver crashes with EINVAL**: Fixed by handling EBADF/EINVAL in + `waitOnSocketWritable()` (see Headers.hpp). The fd becomes invalid when the + client disconnects mid-drain. +- **"Connection refused" on proxy**: The proxy process died. Check if port 4445 + is already in use (`lsof -i:4445`). diff --git a/test/e2e/REPORT.md b/test/e2e/REPORT.md new file mode 100644 index 000000000..4fb9caf19 --- /dev/null +++ b/test/e2e/REPORT.md @@ -0,0 +1,165 @@ +# ET Flow Control: E2E Test Report + +## Problem Statement (Issue #631) + +When a process inside an ET session produces output faster than the network can +deliver it, ET has no mechanism to manage the mismatch. The result: + +1. The terminal display falls progressively behind real time +2. Ctrl-C appears to do nothing: the interrupt is delivered, but tens of + seconds of stale queued output must drain before the prompt reappears +3. In the worst case (laptop sleep, wifi drop), long-running jobs freeze + entirely because the PTY kernel buffer fills up and `write()` blocks + +## The fix, in two parts + +**Part 1: application-level flow control.** Terminal output is staged in a +bounded `WriteBuffer` between the PTY and the client socket. Flow control +is strictly opt-in; users who don't opt in are completely unchanged: + +- `--flow-control none` (default): the exact pre-feature code path. Output + is written straight to the connection with blocking writes; no + application buffering, no kernel socket tuning. +- `--flow-control backpressure`: when the buffer fills, stop reading from + the PTY. The remote process pauses, exactly like plain ssh. Lossless — + required for correctness-sensitive consumers like `tmux -CC`. +- `--flow-control discard`: when the buffer fills, drop the *oldest* pending + output. The remote process never stalls and the display stays close to + real time. Old output is lost from scrollback while the link is saturated. + +**Part 2: kernel buffer tuning (opt-in sessions only).** Flow control only +helps if pending data actually waits in the application buffer. An audit of +every buffer on the path found kernel-side reservoirs that were absorbing +the backlog downstream of any point where ET could drop or gate it. These +are tuned per-session, only when the client opted in (the server learns the +mode from InitialPayload; etterminal learns it via TermInit): + +| Buffer | Untuned size | Fix | +|---|---|---| +| TCP send buffer (server->client) | autotunes to multi-MB (20MB max on this host) | `TCP_NOTSENT_LOWAT=32KB`: select() only reports writable when <32KB is unsent, so the backlog stays in the WriteBuffer. In-flight (sent-but-unacked) data is not limited, preserving high-BDP throughput. | +| ET WriteBuffer | 256KB | 64KB — only needs to absorb bursts between drain opportunities | +| etterminal->etserver unix socket | ~200KB (net.core.wmem_default) | `SO_SNDBUF=64KB` on the sender side | +| PTY kernel buffer | ~64KB | not tunable from userspace; part of the fixed lag floor | +| BackedWriter backup (64MB) | n/a | reconnect recovery only; never delays live data | + +## Test Setup + +``` +print_timestamps.py -> PTY -> etterminal -> etserver --[TCP]--> throttle_proxy --[TCP]--> et client -> terminal + 100KB/s +``` + +All components run on a single machine via `--idpasskey` (no SSH). A userspace +TCP throttle proxy between etserver and the et client limits server-to-client +throughput to 100KB/s. The proxy clamps `SO_RCVBUF` on its server-facing +socket to 64KB so it models a real slow link instead of silently absorbing +megabytes in its own autotuned receive buffer. + +**Workload**: `print_timestamps.py` prints 1000 timestamped lines in a burst +every 10ms (~2.7MB/s raw output — 27x the proxy's capacity). + +**Metrics** (sampled every 5s; Ctrl-C sent after the workload): + +- **display_lag**: wall clock minus the newest timestamp visible on the + client's terminal — how stale the screen is. +- **process_lag**: wall clock minus the timestamp the producer most recently + wrote (via a sidecar file that bypasses ET). Large values mean the process + is blocked in `write()`. +- **ctrl_c_latency**: time from sending Ctrl-C until the shell prompt is + visible again. This is the original issue 631 complaint. + +## Results (2026-07-10, devvm, Linux 6.13; re-validated on the v7.0.0 base) + +Upstream v7.0.0 independently added disconnect buffering (#731, #762: the +connection buffers up to 64MB while disconnected so processes don't freeze +immediately) and a PTY-input deadlock fix (#765). Those address the +disconnect freeze, but not the connected saturated link: with a live but +slow connection, writes still land in the autotuned kernel send buffer, so +the `none` baseline below reproduces issue #631 unchanged on v7.0.0. + +### Slow link, 30s saturation + +| scenario | display lag @30s | process | Ctrl-C -> prompt | +|---|---|---|---| +| `none` (default — status quo, v7.0.0 base) | 32.3s, growing ~1s/s | throttled to link rate | **115.8s** | +| backpressure, no kernel tuning (pre-v7 base) | 31.5s, growing | throttled | 53.8s | +| discard, no kernel tuning (pre-v7 base) | 24.4s, growing | full speed | 28.0s | +| **backpressure + tuning (opt-in)** | **~2s, bounded** | throttled (lossless) | **2.7s** | +| **discard + tuning (opt-in)** | **~1.1s, bounded** | **full speed** | **2.1s** | + +The `none` row is the default build with no flag: unbounded kernel queue, +growing lag, and Ctrl-C takes however long the queue takes to drain (high +variance run to run; 56-116s observed across bases). This is deliberate: +current users see zero change unless they opt in. + +The no-kernel-tuning rows show why the tuning is essential: with the default +autotuned TCP send buffer, megabytes of stale output pile up in the kernel +where the WriteBuffer can neither gate nor drop them, and both modes are +barely better than trunk. With `TCP_NOTSENT_LOWAT`, display lag stops growing +entirely: it is bounded by the small fixed pipeline (WriteBuffer + unsent +bytes + link queue) instead of scaling with how long the link has been +saturated. + +### 60s disconnect at t=30s (tuned) + +| scenario | process during disconnect | after reconnect | Ctrl-C | +|---|---|---|---| +| discard | **full speed, never stalls** | fresh output in <5s, no stale replay | 1.6-2.1s | +| backpressure | stalls (lossless contract, like ssh; the v7 default instead buffers up to 64MB before stalling) | resumes, display recovers in <5s | 2.6s | + +etserver stayed alive through all scenarios. + +### Bulk throughput (localhost, no throttle, 30MB through the full pipeline) + +Measurements are noisy on a shared devvm (trunk build: 40-47 MB/s across +runs; this branch: 20-41 MB/s across modes, with `none`, `backpressure`, +and `discard` overlapping run-to-run). Two things hold: all modes stay in +the tens of MB/s, far beyond any realistic terminal workload, and the +default `none` path is the identical pre-feature code, so non-opted-in +users cannot regress. On real WAN links throughput is BDP-bound, which +`TCP_NOTSENT_LOWAT` does not restrict. + +## Why the default is `none` + +The requirement is that current users are no worse off — the strongest +form of that is byte-identical behavior, so the default (CLI flag and +proto field alike, which also covers old clients talking to new servers) +is the legacy path with stock kernel buffering. + +`backpressure` is the conservative opt-in: it keeps today's lossless +semantics (a slow client pauses the producer, exactly like plain ssh; safe +for `tmux -CC`) while bounding the queue so Ctrl-C takes ~3s instead of +minutes. `discard` is the opt-in for freshness over completeness — the +`cat /dev/zero | base64` case from the issue, ML training jobs, builds: +the process never stalls (even fully disconnected) and the display tracks +real time. The cost is that saturated-link output is missing from +scrollback. + +## What this means for real-world use + +- **ML training / long build, laptop closed**: with `none` or + `backpressure`, the job pauses when buffers fill — same as today. With + `--flow-control discard`, the job keeps running at full speed and you see + current output when you reconnect. +- **`cat` a huge file over a slow link, then Ctrl-C**: prompt returns in + ~2-3s in either opt-in mode (default/today: a minute or more). +- **tmux -CC**: use `none` (default) or `backpressure`. Every byte is + delivered. + +## Reproducing These Results + +```bash +cd /path/to/EternalTerminal/build +cmake -DDISABLE_VCPKG=ON -GNinja .. && ninja -j4 + +# One scenario (trunk|none|backpressure|discard) [--disconnect] +bash test/e2e/do_scenario.sh discard --outdir /tmp/et_e2e +bash test/e2e/do_scenario.sh discard --disconnect --outdir /tmp/et_e2e + +# Bulk throughput on a fast link +bash test/e2e/throughput_test.sh +bash test/e2e/throughput_test.sh --flow-control discard +``` + +See `test/e2e/README.md` for manual setup and `test/e2e/throttle_proxy.py` for +details on the proxy implementation. diff --git a/test/e2e/do_scenario.sh b/test/e2e/do_scenario.sh new file mode 100644 index 000000000..e86bc8083 --- /dev/null +++ b/test/e2e/do_scenario.sh @@ -0,0 +1,285 @@ +#!/bin/bash +# Runs one E2E flow control scenario. Outputs JSONL metrics. +# +# Usage: bash do_scenario.sh [--disconnect] [--outdir DIR] +# mode: trunk | backpressure | discard +# --disconnect: simulate 60s TCP disconnect at t=30s +# --outdir: directory for JSONL output and logs +# +# Commits: +# trunk: ee8ddc21c (base) + f8930e169 (harness: --idpasskey, crash fix) +# backpressure: 497000e08 (--flow-control backpressure) +# discard: 497000e08 (--flow-control discard) + +MODE="$1"; shift +DISCONNECT=false +OUTDIR="/tmp/et_e2e_results" +while [ $# -gt 0 ]; do + case "$1" in + --disconnect) DISCONNECT=true ;; + --outdir) OUTDIR="$2"; shift ;; + esac + shift +done + +DIR="$(cd "$(dirname "$0")" && pwd)" +REPO="$(cd "$DIR/../.." && pwd)" +BUILD="$REPO/build" +RATE=100000 +ID="E2E$(date +%s | tail -c 8)" +KEY="E2ETestKey123456789012345678901A" +SIDECAR="/tmp/et_e2e_sidecar_$$.txt" +ETMUX="/tmp/et_e2e_client_$$.sock" +PROXY_LOG="/tmp/et_e2e_proxy_$$.log" + +# The cleanup trap restores src/, proto/, and test dirs to HEAD (the trunk +# scenario checks out old sources there). Any uncommitted work in those +# paths would be silently destroyed, so refuse to run on a dirty tree. +if ! git -C "$REPO" diff --quiet -- src/ proto/ test/integration_tests/ test/unit_tests/ 2>/dev/null; then + echo "ERROR: uncommitted changes under src/, proto/, or test/. Commit them first:" + git -C "$REPO" status --short -- src/ proto/ test/integration_tests/ test/unit_tests/ | head + exit 1 +fi + +mkdir -p "$OUTDIR" + +SCENARIO="$MODE" +[ "$DISCONNECT" = "true" ] && SCENARIO="${MODE}-disconnect" +JSONL="$OUTDIR/${SCENARIO}.jsonl" +> "$JSONL" + +C='\033[1;36m'; G='\033[1;32m'; R='\033[1;31m'; Y='\033[33m'; N='\033[0m' + +cleanup() { + tmux -S "$ETMUX" kill-server 2>/dev/null + kill $ETSERVER_PID $PROXY_PID 2>/dev/null + kill -9 $(lsof -t -i:4444 2>/dev/null) $(lsof -t -i:4445 2>/dev/null) 2>/dev/null + pkill -9 -f "etterminal.*$ID" 2>/dev/null + rm -f /tmp/et_e2e_demo.fifo "$ETMUX" "$SIDECAR" "$PROXY_LOG" + # Restore only the dirs the trunk scenario checks out. Do NOT use + # "git checkout HEAD -- ." here: it clobbers unrelated uncommitted work + # in the tree (this exact mistake once reverted the flow-control + # implementation itself). + cd "$REPO" && git checkout HEAD -- src/ proto/ test/integration_tests/ test/unit_tests/ 2>/dev/null +} +trap cleanup EXIT + +# --- Checkout and build --- + +COMMIT_DESC="" +ET_CMD="" +case "$MODE" in + trunk) + echo -e "${C}=== Scenario: trunk (no flow control) ===${N}" + COMMIT_DESC="ee8ddc21c + f8930e169 (harness only, no WriteBuffer)" + ET_CMD="./et --idpasskey=ID/KEY 127.0.0.1:4445" + cd "$REPO" + git checkout ee8ddc21c -- src/ proto/ test/integration_tests/ test/unit_tests/ 2>/dev/null + git checkout f8930e169 -- src/terminal/TerminalClientMain.cpp src/terminal/TerminalMain.cpp src/base/Headers.hpp 2>/dev/null + rm -f src/base/WriteBuffer.hpp test/unit_tests/WriteBufferTest.cpp + ;; + none) + echo -e "${C}=== Scenario: none (HEAD, no --flow-control: must match trunk) ===${N}" + COMMIT_DESC="HEAD ($(cd "$REPO" && git rev-parse --short HEAD)), no flag" + ET_CMD="./et --idpasskey=ID/KEY 127.0.0.1:4445" + # Use HEAD as-is (no checkout needed) + ;; + backpressure) + echo -e "${C}=== Scenario: backpressure ===${N}" + COMMIT_DESC="HEAD ($(cd "$REPO" && git rev-parse --short HEAD))" + ET_CMD="./et --idpasskey=ID/KEY --flow-control backpressure 127.0.0.1:4445" + # Use HEAD as-is (no checkout needed) + ;; + discard) + echo -e "${C}=== Scenario: discard ===${N}" + COMMIT_DESC="HEAD ($(cd "$REPO" && git rev-parse --short HEAD))" + ET_CMD="./et --idpasskey=ID/KEY --flow-control discard 127.0.0.1:4445" + # Use HEAD as-is (no checkout needed) + ;; + *) echo "Usage: $0 [--disconnect]"; exit 1 ;; +esac +FC_FLAG="" +[ "$MODE" = "backpressure" ] && FC_FLAG="--flow-control backpressure" +[ "$MODE" = "discard" ] && FC_FLAG="--flow-control discard" + +echo " commit: $COMMIT_DESC" +echo " client: $ET_CMD" +echo " proxy: ${RATE}B/s throttle on port 4445" +[ "$DISCONNECT" = "true" ] && echo " disconnect: 60s at t=30s" +echo "" + +echo -e "${C}--- Build ---${N}" +cd "$BUILD" +cmake -DDISABLE_VCPKG=ON -GNinja .. 2>&1 | tail -1 +ninja -j4 2>&1 | tail -3 +echo "" + +# --- Start services --- + +echo -e "${C}--- Services ---${N}" +rm -f /tmp/et_e2e_demo.fifo + +"$BUILD/etserver" --serverfifo=/tmp/et_e2e_demo.fifo --port=4444 --logdir="$OUTDIR" &>/dev/null & +ETSERVER_PID=$! +sleep 3 + +PYTHONUNBUFFERED=1 python3 -u "$DIR/throttle_proxy.py" 4445 4444 "$RATE" >"$PROXY_LOG" 2>&1 & +PROXY_PID=$! +sleep 2 + +"$BUILD/etterminal" --idpasskey="$ID/$KEY" --serverfifo=/tmp/et_e2e_demo.fifo --logdir="$OUTDIR" &>/dev/null & +sleep 3 + +for i in $(seq 1 15); do + ES=$(lsof -i:4444 2>/dev/null | grep -c LISTEN) + PX=$(lsof -i:4445 2>/dev/null | grep -c LISTEN) + [ "$ES" -ge 1 ] && [ "$PX" -ge 1 ] && break + sleep 1 +done +[ "$ES" -lt 1 ] || [ "$PX" -lt 1 ] && { echo -e "${R}FAILED: etserver=$ES proxy=$PX${N}"; exit 1; } +echo -e "${G}etserver=:4444 proxy=:4445 (${RATE}B/s)${N}" + +# --- Connect --- + +echo -e "${C}--- Connect ---${N}" +tmux -S "$ETMUX" new-session -d -s et -x 200 -y 50 +for i in $(seq 1 30); do + tmux -S "$ETMUX" capture-pane -p 2>/dev/null | grep -q '\$' && break + sleep 1 +done +tmux -S "$ETMUX" send-keys "$BUILD/et --idpasskey='$ID/$KEY' 127.0.0.1:4445 $FC_FLAG" Enter + +for i in $(seq 1 30); do + grep -q connect "$PROXY_LOG" 2>/dev/null && break + sleep 1 + [ "$i" -eq 30 ] && { echo -e "${R}FAILED: no proxy connection${N}"; cat "$PROXY_LOG"; exit 1; } +done +echo -e "${G}Connected through proxy${N}" + +for i in $(seq 1 30); do + tmux -S "$ETMUX" capture-pane -p 2>/dev/null | grep -qF '~]$' && break + sleep 1 +done +echo -e "${G}Remote shell ready${N}" +echo "" + +# --- Run workload --- + +TOTAL_DURATION=30 +[ "$DISCONNECT" = "true" ] && TOTAL_DURATION=120 +DISCONNECT_AT=30 +RECONNECT_AT=90 + +echo -e "${C}--- print_timestamps.py (${TOTAL_DURATION}s) ---${N}" +tmux -S "$ETMUX" send-keys "SIDECAR_FILE=$SIDECAR python3 $DIR/print_timestamps.py" Enter + +PREV_LINES=0 +PREV_ELAPSED="0" + +printf "%-6s %-12s %-12s %-12s %-10s %-10s\n" "TIME" "DISPLAY_LAG" "PROCESS_LAG" "LINES/SEC" "CONNECTED" "STATUS" +printf "%-6s %-12s %-12s %-12s %-10s %-10s\n" "----" "-----------" "-----------" "---------" "---------" "------" + +ELAPSED=0 +while [ "$ELAPSED" -lt "$TOTAL_DURATION" ]; do + sleep 5 + ELAPSED=$((ELAPSED + 5)) + + if [ "$DISCONNECT" = "true" ]; then + [ "$ELAPSED" -eq "$DISCONNECT_AT" ] && { echo -e "${R}>>> DISCONNECT <<<${N}"; kill -USR1 $PROXY_PID 2>/dev/null; } + [ "$ELAPSED" -eq "$RECONNECT_AT" ] && { echo -e "${G}>>> RECONNECT <<<${N}"; kill -USR2 $PROXY_PID 2>/dev/null; } + fi + + CONNECTED="true" + [ "$DISCONNECT" = "true" ] && [ "$ELAPSED" -gt "$DISCONNECT_AT" ] && [ "$ELAPSED" -le "$RECONNECT_AT" ] && CONNECTED="false" + + SCREEN=$(tmux -S "$ETMUX" capture-pane -p 2>/dev/null | grep -oE '[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]+' | tail -1) + REAL=$(python3 -c "from datetime import datetime; print(datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f'))") + + DLAG="null" + [ -n "$SCREEN" ] && DLAG=$(python3 -c " +from datetime import datetime +s=datetime.strptime('${SCREEN}'.strip(),'%Y-%m-%d %H:%M:%S.%f') +r=datetime.strptime('${REAL}'.strip(),'%Y-%m-%d %H:%M:%S.%f') +print(f'{(r-s).total_seconds():.1f}')" 2>/dev/null || true) + + PLAG="null"; LINES_SEC="null"; CUR_LINES=0; CUR_ELAPSED="0" + if [ -f "$SIDECAR" ]; then + IFS=',' read -r PROCESS_TS CUR_LINES CUR_ELAPSED < "$SIDECAR" + [ -n "$PROCESS_TS" ] && PLAG=$(python3 -c " +from datetime import datetime +s=datetime.strptime('${PROCESS_TS}'.strip(),'%Y-%m-%d %H:%M:%S.%f') +r=datetime.strptime('${REAL}'.strip(),'%Y-%m-%d %H:%M:%S.%f') +print(f'{(r-s).total_seconds():.1f}')" 2>/dev/null || true) + [ -n "$CUR_ELAPSED" ] && [ -n "$CUR_LINES" ] && LINES_SEC=$(python3 -c " +cl=int('${CUR_LINES}'); pl=int('${PREV_LINES}') +ct=float('${CUR_ELAPSED}'); pt=float('${PREV_ELAPSED}') +dt=ct-pt +print(f'{(cl-pl)/dt:.0f}' if dt > 0.1 else '0')" 2>/dev/null || true) + fi + PREV_LINES="${CUR_LINES:-0}"; PREV_ELAPSED="${CUR_ELAPSED:-0}" + + PSTATUS="running" + [ "$PLAG" != "null" ] && ! python3 -c "exit(0 if float('$PLAG') < 1.0 else 1)" 2>/dev/null && PSTATUS="STALLED" + + DLAG_S="${DLAG}s"; [ "$DLAG" = "null" ] && DLAG_S="?" + PLAG_S="${PLAG}s"; [ "$PLAG" = "null" ] && PLAG_S="?" + LS_S="$LINES_SEC"; [ "$LINES_SEC" = "null" ] && LS_S="?" + + COLOR="$G"; [ "$PSTATUS" = "STALLED" ] && COLOR="$R" + printf "t=%-3ss %-12s %-12s %-12s %-10s ${COLOR}%-10s${N}\n" "$ELAPSED" "$DLAG_S" "$PLAG_S" "$LS_S" "$CONNECTED" "$PSTATUS" + + echo "{\"t\":$ELAPSED,\"display_lag\":$DLAG,\"process_lag\":$PLAG,\"lines_sec\":$LINES_SEC,\"connected\":$CONNECTED}" >> "$JSONL" +done + +# --- Ctrl-C responsiveness --- +# This is the original issue 631 complaint: with a saturated link, ^C takes +# a very long time to visibly take effect because stale queued output must +# drain before the prompt reappears. +echo "" +echo -e "${C}--- Ctrl-C responsiveness ---${N}" +CC_T0=$(python3 -c 'import time; print(f"{time.time():.3f}")') +tmux -S "$ETMUX" send-keys C-c +CTRL_C_LATENCY="null" +for i in $(seq 1 240); do + sleep 0.5 + LAST_LINES=$(tmux -S "$ETMUX" capture-pane -p 2>/dev/null | sed '/^$/d' | tail -2) + if echo "$LAST_LINES" | grep -qF '~]$'; then + CTRL_C_LATENCY=$(python3 -c "import time; print(f'{time.time() - $CC_T0:.1f}')") + break + fi +done +if [ "$CTRL_C_LATENCY" = "null" ]; then + echo -e "${R}Ctrl-C: prompt did not return within 120s${N}" +else + echo -e "Ctrl-C to prompt: ${G}${CTRL_C_LATENCY}s${N}" +fi +echo "{\"event\":\"ctrl_c\",\"latency\":$CTRL_C_LATENCY}" >> "$JSONL" + +echo "" +ES_ALIVE=$(lsof -i:4444 2>/dev/null | grep -c LISTEN) +[ "$ES_ALIVE" -gt 0 ] && echo -e "etserver: ${G}alive${N}" || echo -e "etserver: ${R}CRASHED${N}" + +cp "$PROXY_LOG" "$OUTDIR/${SCENARIO}_proxy.log" 2>/dev/null + +python3 -c " +import json +meta = { + 'scenario': '$SCENARIO', + 'mode': '$MODE', + 'disconnect': $( [ \"$DISCONNECT\" = \"true\" ] && echo True || echo False ), + 'commit': '$COMMIT_DESC', + 'et_command': '$ET_CMD', + 'proxy_rate': $RATE, + 'ctrl_c_latency': $CTRL_C_LATENCY, + 'etserver_alive': $( [ "$ES_ALIVE" -gt 0 ] && echo True || echo False ), +} +with open('$OUTDIR/${SCENARIO}_meta.json', 'w') as f: + json.dump(meta, f, indent=2) +" + +echo "Results: $JSONL" + +tmux -S "$ETMUX" send-keys C-c 2>/dev/null +sleep 1 +tmux -S "$ETMUX" send-keys "exit" Enter 2>/dev/null +sleep 1 diff --git a/test/e2e/generate_report.py b/test/e2e/generate_report.py new file mode 100644 index 000000000..6cd0bb2c4 --- /dev/null +++ b/test/e2e/generate_report.py @@ -0,0 +1,194 @@ +#!/usr/bin/env python3 +"""Generate a self-contained HTML report with inline SVG charts from E2E JSONL data. + +Usage: python3 generate_report.py +""" +import json +import sys +from datetime import datetime +from pathlib import Path + + +def load_scenario(results_dir, name): + jsonl = Path(results_dir) / f"{name}.jsonl" + meta_file = Path(results_dir) / f"{name}_meta.json" + if not jsonl.exists(): + return None + samples = [] + for line in jsonl.read_text().strip().split("\n"): + if line.strip(): + samples.append(json.loads(line)) + meta = json.loads(meta_file.read_text()) if meta_file.exists() else {} + return {"name": name, "samples": samples, "meta": meta} + + +def svg_chart(scenarios, key, title, ylabel, width=700, height=300, events=None): + colors = { + "trunk": "#e74c3c", "backpressure": "#f39c12", + "discard": "#2ecc71", "discard-disconnect": "#3498db", + "trunk-disconnect": "#c0392b", "backpressure-disconnect": "#e67e22", + } + ml, mr, mt, mb = 70, 140, 40, 50 + pw, ph = width - ml - mr, height - mt - mb + + all_t, all_v = [], [] + for sc in scenarios: + for s in sc["samples"]: + all_t.append(s["t"]) + v = s.get(key) + if v is not None and str(v) != "null": + try: + all_v.append(float(v)) + except (ValueError, TypeError): + pass + if not all_v: + return f"

No data for {title}

" + + t_min, t_max = min(all_t), max(all_t) + v_max = max(all_v) * 1.1 or 1 + + def tx(t): return ml + (t - t_min) / max(t_max - t_min, 1) * pw + def ty(v): return mt + ph - v / v_max * ph + + out = [f'', + f'', + f'{title}'] + + for i in range(6): + y = mt + i * ph / 5 + v = v_max * (5 - i) / 5 + out.append(f'') + out.append(f'{v:.1f}') + + step = max(int((t_max - t_min) / 8), 5) + for t in range(0, int(t_max) + 1, step): + x = tx(t) + out.append(f'') + out.append(f'{t}s') + + out.append(f'') + out.append(f'') + out.append(f'Time (seconds)') + out.append(f'{ylabel}') + + if events: + for evt_t, lbl, clr in events: + x = tx(evt_t) + out.append(f'') + out.append(f'{lbl}') + + ly = mt + 10 + for sc in scenarios: + color = colors.get(sc["name"], "#999") + pts = [(s["t"], float(s[key])) for s in sc["samples"] + if s.get(key) is not None and str(s[key]) != "null" + and _safe_float(s[key]) is not None] + if not pts: + continue + path = " ".join(f"{'M' if i==0 else 'L'} {tx(t):.1f} {ty(v):.1f}" for i,(t,v) in enumerate(pts)) + out.append(f'') + for t, v in pts: + out.append(f'') + lx = ml + pw + 10 + out.append(f'') + out.append(f'{sc["name"]}') + ly += 18 + + out.append("") + return "\n".join(out) + + +def _safe_float(v): + try: + return float(v) + except (ValueError, TypeError): + return None + + +def generate_html(results_dir, output_path): + scenarios = [s for s in [load_scenario(results_dir, n) + for n in ["trunk", "backpressure", "discard", + "trunk-disconnect", "backpressure-disconnect", + "discard-disconnect"]] if s] + if not scenarios: + print(f"No data in {results_dir}") + sys.exit(1) + + events = [(30, "disconnect", "#e74c3c"), (90, "reconnect", "#2ecc71")] + has_dc = any(s["name"] == "discard-disconnect" for s in scenarios) + + h = ['', + 'ET Flow Control E2E Report', + '''''', + '', + '

ET Flow Control E2E Test Report

', + f'

Generated: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}

', + '

Test Scenarios

', + ''] + + for sc in scenarios: + m = sc["meta"] + alive = 'alive' if m.get("etserver_alive", True) else 'CRASHED' + rate = m.get("proxy_rate", "?") + rate_str = f"{rate:,} B/s" if isinstance(rate, int) else str(rate) + h.append(f'' + f'' + f'') + h.append('
ScenarioGit CommitClient CommandProxy RateServer
{sc["name"]}{m.get("commit","?")}{m.get("et_command","?")}{rate_str}{alive}
') + + h.append('
Setup: etserver :4444 → throttle proxy :4445 → et client. ' + 'Workload: print_timestamps.py ~2.7MB/s. Proxy limits to 100KB/s.
') + + h.append('

Display Lag

') + h.append('

How stale the terminal output is. Grows ~1s/s in all modes (TCP buffer bottleneck).

') + h.append(f'
{svg_chart(scenarios, "display_lag", "Display Lag Over Time", "Seconds behind real time", events=events if has_dc else None)}
') + + h.append('

Process Lag

') + h.append('

Whether the producing process is stalled. 0 = running freely, >1 = blocked on write().

') + h.append(f'
{svg_chart(scenarios, "process_lag", "Process Lag Over Time", "Seconds behind real time", events=events if has_dc else None)}
') + + h.append('

Process Throughput

') + h.append('

Lines/sec written by the producing process. Higher = process running freely.

') + h.append(f'
{svg_chart(scenarios, "lines_sec", "Process Throughput", "Lines/sec", events=events if has_dc else None)}
') + + h.append('

Interpretation

') + h.append('

Display lag is a network property

') + h.append('

All modes show ~1s/s growth because the proxy limits throughput. TCP is ordered — ' + 'data in the kernel buffer must be delivered in sequence. No application-level change can fix this.

') + h.append('

Process lag is the key differentiator

') + h.append('

Trunk & backpressure: process stalls intermittently (process_lag > 0) when TCP buffer fills.

') + h.append('

Discard: process_lag stays at 0. Old data dropped, PTY always drained.

') + + if has_dc: + h.append('

Disconnect scenario

') + h.append('

At t=30s, proxy kills TCP connections for 60s. At t=90s, proxy resumes.

') + h.append('

In discard mode, the process keeps running during disconnect. ' + 'WriteBuffer discards old output. On reconnect, client sees recent data.

') + + h.append('

When to use each mode

') + h.append('') + h.append('') + h.append('
ModeBest forTrade-off
discard (default)Long-running jobs, buildsOld output lost
backpressuretmux -CC, stateful protocolsProcess stalls
') + h.append('') + + Path(output_path).write_text("\n".join(h)) + print(f"Report: {output_path}") + + +if __name__ == "__main__": + if len(sys.argv) < 3: + print("Usage: python3 generate_report.py ") + sys.exit(1) + generate_html(sys.argv[1], sys.argv[2]) diff --git a/test/e2e/print_timestamps.py b/test/e2e/print_timestamps.py new file mode 100644 index 000000000..b047f0241 --- /dev/null +++ b/test/e2e/print_timestamps.py @@ -0,0 +1,27 @@ +"""Prints timestamped output at high volume to test flow control. + +Produces ~2.7MB/s of output (1000 lines per burst, bursts every 10ms). + +If SIDECAR_FILE is set, writes the current timestamp and cumulative line +count to that file after each burst, enabling external measurement of: + - process_lag: is the process keeping up with wall clock? + - lines_per_sec: how many lines has the process written? +""" +import os +import sys +from datetime import datetime +import time + +sidecar = os.environ.get("SIDECAR_FILE", "") +total_lines = 0 +start_time = time.monotonic() + +while True: + ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f") + print((ts + "\n") * 1000) + total_lines += 1000 + if sidecar: + elapsed = time.monotonic() - start_time + with open(sidecar, "w") as f: + f.write(f"{ts},{total_lines},{elapsed:.3f}\n") + time.sleep(0.01) diff --git a/test/e2e/run_all_scenarios.sh b/test/e2e/run_all_scenarios.sh new file mode 100644 index 000000000..689701b32 --- /dev/null +++ b/test/e2e/run_all_scenarios.sh @@ -0,0 +1,56 @@ +#!/bin/bash +# Run all E2E flow control scenarios and generate HTML report. +# +# Usage: bash run_all_scenarios.sh [output_dir] +# +# Scenarios: trunk (~60s), backpressure (~60s), discard (~60s), +# discard-disconnect (~150s). Total: ~6 minutes. +# +# Output: /report.html, /*.jsonl, /*_meta.json + +set -e + +DIR="$(cd "$(dirname "$0")" && pwd)" +REPO="$(cd "$DIR/../.." && pwd)" +OUTDIR="${1:-$DIR/results}" +LOCKFILE="/tmp/et_e2e_run.lock" + +exec 200>"$LOCKFILE" +flock -n 200 || { echo "Another run in progress"; exit 1; } + +mkdir -p "$OUTDIR" +echo "=== ET Flow Control E2E Test Suite ===" +echo "Output: $OUTDIR" +echo "Started: $(date)" +echo "" + +kill -9 $(lsof -t -i:4444 2>/dev/null) $(lsof -t -i:4445 2>/dev/null) 2>/dev/null || true +pkill -9 -f throttle_proxy 2>/dev/null || true +sleep 2 + +for SCENARIO in "trunk" "backpressure" "discard" "discard --disconnect"; do + MODE=$(echo "$SCENARIO" | awk '{print $1}') + EXTRA=$(echo "$SCENARIO" | awk '{$1=""; print $0}' | xargs) + echo "================================================================" + echo " Scenario: $MODE $EXTRA" + echo "================================================================" + + kill -9 $(lsof -t -i:4444 2>/dev/null) $(lsof -t -i:4445 2>/dev/null) 2>/dev/null || true + pkill -9 -f throttle_proxy 2>/dev/null || true + sleep 3 + + bash "$DIR/do_scenario.sh" $SCENARIO --outdir "$OUTDIR" + echo "" +done + +cd "$REPO" && git checkout HEAD -- . 2>/dev/null +cd build && cmake -DDISABLE_VCPKG=ON -GNinja .. 2>&1 | tail -1 && ninja -j4 2>&1 | tail -1 + +echo "================================================================" +echo " Generating HTML report" +echo "================================================================" +python3 "$DIR/generate_report.py" "$OUTDIR" "$OUTDIR/report.html" + +echo "" +echo "=== Complete at $(date) ===" +ls -la "$OUTDIR"/*.html "$OUTDIR"/*.jsonl 2>/dev/null diff --git a/test/e2e/run_e2e_test.sh b/test/e2e/run_e2e_test.sh new file mode 100644 index 000000000..ea4d0a713 --- /dev/null +++ b/test/e2e/run_e2e_test.sh @@ -0,0 +1,102 @@ +#!/bin/bash +# End-to-end flow control test. Runs etserver, a throttle proxy, etterminal, +# and et client in tmux, then measures timestamp lag. +# +# Usage: bash run_e2e_test.sh [discard|backpressure] [rate_bytes_per_sec] +# +# Must be run outside of any process supervisor that kills children +# (e.g. via cron). See test/e2e/README.md for details. + +MODE="${1:-discard}" +RATE="${2:-100000}" +DIR="$(cd "$(dirname "$0")" && pwd)" +BUILD="$(cd "$DIR/../../build" && pwd)" +SOCK="/tmp/et_e2e_$$.sock" +FIFO="/tmp/et_e2e_$$.fifo" +ID="E2E$(date +%s)" +KEY="E2ETestKey123456789012345678901A" +SIDECAR="/tmp/et_e2e_sidecar_$$.txt" + +cleanup() { + tmux -S "$SOCK" kill-server 2>/dev/null || true + kill -9 $(lsof -t -i:4444 2>/dev/null) 2>/dev/null || true + kill -9 $(lsof -t -i:4445 2>/dev/null) 2>/dev/null || true + pkill -9 -f "etterminal.*$ID" 2>/dev/null || true + rm -f "$FIFO" "$SOCK" "$SIDECAR" || true +} +trap cleanup EXIT +cleanup 2>/dev/null + +# Start services +tmux -S "$SOCK" new-session -d -s t -x 200 -y 50 +tmux -S "$SOCK" send-keys "$BUILD/etserver --serverfifo=$FIFO --port=4444" Enter +sleep 5 +tmux -S "$SOCK" split-window -v +sleep 1 +tmux -S "$SOCK" send-keys "python3 $DIR/throttle_proxy.py 4445 4444 $RATE" Enter +sleep 3 +tmux -S "$SOCK" split-window -v +sleep 1 +tmux -S "$SOCK" send-keys "$BUILD/etterminal --idpasskey='$ID/$KEY' --serverfifo=$FIFO" Enter +sleep 5 + +# Verify +ES=$(lsof -i:4444 2>/dev/null | grep -c LISTEN) +PX=$(lsof -i:4445 2>/dev/null | grep -c LISTEN) +if [ "$ES" -lt 1 ] || [ "$PX" -lt 1 ]; then + echo "FAILED: etserver=$ES proxy=$PX" + exit 1 +fi + +# Connect client +EXTRA="" +if echo "$BUILD/et --help" | grep -q flow-control 2>/dev/null; then + EXTRA="--flow-control $MODE" +fi +tmux -S "$SOCK" new-window -n et +tmux -S "$SOCK" send-keys -t t:et "$BUILD/et --idpasskey='$ID/$KEY' 127.0.0.1:4445 $EXTRA" Enter +sleep 10 + +# Verify proxy connection +PROXY_OUT=$(tmux -S "$SOCK" capture-pane -t t:0.1 -p) +if ! echo "$PROXY_OUT" | grep -q connect; then + echo "FAILED: proxy saw no connection" + echo "$PROXY_OUT" | tail -5 + exit 1 +fi + +# Run timestamp test +tmux -S "$SOCK" send-keys -t t:et "SIDECAR_FILE=$SIDECAR python3 $DIR/print_timestamps.py" Enter + +echo "=== $MODE mode, ${RATE}B/s proxy ===" +for secs in 10 20 30; do + sleep 10 + SCREEN=$(tmux -S "$SOCK" capture-pane -t t:et -p | grep -oE '[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]+' | tail -1) + REAL=$(python3 -c "from datetime import datetime; print(datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f'))") + if [ -n "$SCREEN" ]; then + LAG=$(python3 -c " +from datetime import datetime +s=datetime.strptime('${SCREEN}'.strip(),'%Y-%m-%d %H:%M:%S.%f') +r=datetime.strptime('${REAL}'.strip(),'%Y-%m-%d %H:%M:%S.%f') +print(f'{(r-s).total_seconds():.1f}')") + echo " t=${secs}s: display_lag=${LAG}s" + else + echo " t=${secs}s: no data on screen" + fi + + # Check if the process is stalled by reading the sidecar + if [ -f "$SIDECAR" ]; then + PROCESS_TS=$(cat "$SIDECAR") + PROCESS_LAG=$(python3 -c " +from datetime import datetime +s=datetime.strptime('${PROCESS_TS}'.strip(),'%Y-%m-%d %H:%M:%S.%f') +r=datetime.strptime('${REAL}'.strip(),'%Y-%m-%d %H:%M:%S.%f') +print(f'{(r-s).total_seconds():.1f}')") + echo " process_lag=${PROCESS_LAG}s (0=running, >1=stalled)" + fi +done + +echo "" +echo "Proxy stats:" +tmux -S "$SOCK" capture-pane -t t:0.1 -p | grep -E 'sent=|connect' | tail -3 +echo "etserver alive: $(lsof -i:4444 2>/dev/null | grep -c LISTEN)" diff --git a/test/e2e/throttle_proxy.py b/test/e2e/throttle_proxy.py new file mode 100644 index 000000000..de5da5e66 --- /dev/null +++ b/test/e2e/throttle_proxy.py @@ -0,0 +1,159 @@ +"""TCP throttle proxy for E2E testing. Applies TCP backpressure to the server. + +Reads from the server at a limited rate, causing the kernel TCP send buffer +to fill up. Client->server direction (keystrokes) forwarded at full speed. + +Supports disconnect simulation: + SIGUSR1 -> kill all connections, block new ones for 60s + SIGUSR2 -> resume immediately + +Usage: python3 throttle_proxy.py [bytes_per_sec] +""" +import socket +import sys +import threading +import time +import signal +import select +from datetime import datetime + +LISTEN_PORT, TARGET_PORT = int(sys.argv[1]), int(sys.argv[2]) +RATE = int(sys.argv[3]) if len(sys.argv) > 3 else 100000 + +disconnect_until = 0 +active_connections = [] +lock = threading.Lock() + + +def log(msg): + ts = datetime.now().strftime("%H:%M:%S.%f") + print(f"[proxy {ts}] {msg}", flush=True) + + +def handle_disconnect(signum, frame): + global disconnect_until + disconnect_until = time.monotonic() + 60 + log("DISCONNECT: killing all connections for 60s") + with lock: + for cli, srv, stop in active_connections: + stop.set() + try: + cli.close() + except Exception: + pass + try: + srv.close() + except Exception: + pass + active_connections.clear() + + +def handle_reconnect(signum, frame): + global disconnect_until + disconnect_until = 0 + log("RECONNECT: accepting connections again") + + +signal.signal(signal.SIGUSR1, handle_disconnect) +signal.signal(signal.SIGUSR2, handle_reconnect) + + +def throttled_forward(name, src, dst, rate, stop): + total = 0 + CHUNK = max(rate // 20, 64) + last_t = time.time() + try: + while not stop.is_set(): + data = src.recv(CHUNK) + if not data: + break + dst.sendall(data) + total += len(data) + time.sleep(len(data) / rate) + now = time.time() + if now - last_t > 2: + log(f"{name}: sent={total:,}B") + last_t = now + except Exception: + pass + stop.set() + + +def fast_forward(name, src, dst, stop): + try: + while not stop.is_set(): + data = src.recv(4096) + if not data: + break + dst.sendall(data) + except Exception: + pass + stop.set() + + +def handle(cli, addr): + log(f"connect {addr}") + srv = None + stop = threading.Event() + try: + srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + # Clamp the receive buffer BEFORE connect (window scaling is + # negotiated at handshake). Without this, the proxy's kernel rcvbuf + # autotunes to multiple MB and silently absorbs the server's output, + # hiding buffer bloat that a real slow link would push back on. + # 64KB ~= 0.64s of queue at the default 100KB/s rate. + srv.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 64 * 1024) + srv.connect(("127.0.0.1", TARGET_PORT)) + log(f"srv rcvbuf={srv.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF)}") + with lock: + active_connections.append((cli, srv, stop)) + t1 = threading.Thread( + target=throttled_forward, + args=("s->c", srv, cli, RATE, stop), + daemon=True, + ) + t2 = threading.Thread( + target=fast_forward, args=("c->s", cli, srv, stop), daemon=True + ) + t1.start() + t2.start() + t1.join() + t2.join() + except Exception as e: + log(f"error: {e}") + finally: + for s in (cli, srv): + try: + s.close() + except Exception: + pass + with lock: + active_connections[:] = [ + (c, s, st) for c, s, st in active_connections if st is not stop + ] + log(f"disconnect {addr}") + + +# Dual-stack listener +s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) +s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) +s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) +s.bind(("::", LISTEN_PORT)) +s.listen(5) +s.setblocking(False) +log(f"[::]:{LISTEN_PORT} -> 127.0.0.1:{TARGET_PORT} rate={RATE}B/s") + +while True: + try: + r, _, _ = select.select([s], [], [], 1.0) + for _ in r: + c, a = s.accept() + if time.monotonic() < disconnect_until: + log(f"rejecting {a} (disconnected)") + c.close() + continue + threading.Thread(target=handle, args=(c, a), daemon=True).start() + except KeyboardInterrupt: + break + except Exception: + pass diff --git a/test/e2e/throughput_test.sh b/test/e2e/throughput_test.sh new file mode 100644 index 000000000..c3d8439a8 --- /dev/null +++ b/test/e2e/throughput_test.sh @@ -0,0 +1,72 @@ +#!/bin/bash +# Measures bulk throughput through the full et pipeline on a fast link +# (no throttle proxy). Used to verify that flow control and kernel buffer +# tuning (TCP_NOTSENT_LOWAT, SO_SNDBUF clamps) do not regress throughput. +# +# Usage: bash throughput_test.sh [--flow-control ] [--bytes N] +# Assumes the build/ directory contains current binaries. + +FC_FLAG="" +BYTES=30000000 +while [ $# -gt 0 ]; do + case "$1" in + --flow-control) FC_FLAG="--flow-control $2"; shift ;; + --bytes) BYTES="$2"; shift ;; + esac + shift +done + +DIR="$(cd "$(dirname "$0")" && pwd)" +REPO="$(cd "$DIR/../.." && pwd)" +BUILD="$REPO/build" +PORT=4446 +ID="TPT$(date +%s | tail -c 8)" +KEY="E2ETestKey123456789012345678901A" +ETMUX="/tmp/et_tpt_client_$$.sock" +FIFO="/tmp/et_tpt_$$.fifo" + +cleanup() { + tmux -S "$ETMUX" kill-server 2>/dev/null + kill $ETSERVER_PID 2>/dev/null + kill -9 $(lsof -t -i:$PORT 2>/dev/null) 2>/dev/null + pkill -9 -f "etterminal.*$ID" 2>/dev/null + rm -f "$FIFO" "$ETMUX" +} +trap cleanup EXIT + +"$BUILD/etserver" --serverfifo="$FIFO" --port=$PORT --logdir=/tmp &>/dev/null & +ETSERVER_PID=$! +sleep 2 +"$BUILD/etterminal" --idpasskey="$ID/$KEY" --serverfifo="$FIFO" --logdir=/tmp &>/dev/null & +sleep 2 + +tmux -S "$ETMUX" new-session -d -s et -x 200 -y 50 +for i in $(seq 1 30); do + tmux -S "$ETMUX" capture-pane -p 2>/dev/null | grep -q '\$' && break + sleep 1 +done +tmux -S "$ETMUX" send-keys "$BUILD/et --idpasskey='$ID/$KEY' 127.0.0.1:$PORT $FC_FLAG" Enter +for i in $(seq 1 30); do + tmux -S "$ETMUX" capture-pane -p 2>/dev/null | grep -qF '~]$' && break + sleep 1 +done + +# The marker is split in the typed command so the echoed command line +# itself doesn't match the completion grep. +T0=$(python3 -c 'import time; print(f"{time.time():.3f}")') +tmux -S "$ETMUX" send-keys "base64 -w 200 /dev/zero | head -c $BYTES; echo 'TPT_''END_MARK'" Enter +ELAPSED="timeout" +for i in $(seq 1 1200); do + sleep 0.1 + if tmux -S "$ETMUX" capture-pane -p 2>/dev/null | grep -qF 'TPT_END_MARK'; then + ELAPSED=$(python3 -c "import time; print(f'{time.time() - $T0:.2f}')") + break + fi +done + +if [ "$ELAPSED" = "timeout" ]; then + echo "RESULT: timeout" +else + MBPS=$(python3 -c "print(f'{$BYTES / float('$ELAPSED') / 1e6:.1f}')") + echo "RESULT: ${BYTES} bytes in ${ELAPSED}s = ${MBPS} MB/s (flow-control: ${FC_FLAG:-default})" +fi