From abd91da5ebea6705efb046523d6426283560fe41 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Henrik=20Rydg=C3=A5rd?= Date: Mon, 31 Aug 2026 00:57:17 +0200 Subject: [PATCH] WebSocket: teach the input sink about EOF, and stop losing why a read failed Follow-up to the previous commit, from Nemoumbra's questions - which found a worse spin than the one that fix addressed. InputSink couldn't tell "nothing right now" from "peer is gone": Fill() treats recv() == 0 as no data and only sets hasError_ on a real error. Block() then waits with WaitUntilReady(), which reports a closed socket as ready immediately and forever, so TakeExact() looped on it without ever returning. A client that disconnects with half a frame buffered - easy to do while blasting messages - put the server in an infinite loop inside TakeExact, never even returning to Process(). Measured 7.95 CPU-seconds over 8 seconds; 0.08 after. So: track EOF explicitly (sticky atEnd_, exposed as AtEnd()), and have Block() give up when nothing more can arrive. That information was being thrown away in three more places: * Process() only tried to fill when the sink was already empty, so a disconnect went unnoticed for as long as there were leftovers - and if those leftovers were a partial frame, the read above never completed. Always fill, and close once the peer is gone and we've consumed what it sent. * ReadPending() uses TakeAtMost(), which returns 0 both for "nothing right now" and "nothing ever again", and then reported success having consumed nothing. Ask the sink which it was. * Both TakeExact() call sites answered a failed read with POLICY_VIOLATION, blaming the client for a protocol error when it had simply disconnected. Check the sink and report ABNORMAL when that's what happened. Also stop queueing data once our own close frame is queued. RFC 6455 5.5.1 forbids data frames after a close, and beyond the protocol, anything appended afterwards keeps the buffers non-empty and starves the "everything is flushed" check that ends the connection. Observed the server pumping 167MB of log broadcasts after being asked to close. The repeated close-and-discard is now one helper. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01DCPmm7FoQUoqrbMdhfqhQ2 --- Common/Net/Sinks.cpp | 14 +++++- Common/Net/Sinks.h | 4 ++ Common/Net/WebsocketServer.cpp | 80 ++++++++++++++++++++++++++-------- Common/Net/WebsocketServer.h | 7 +++ 4 files changed, 86 insertions(+), 19 deletions(-) diff --git a/Common/Net/Sinks.cpp b/Common/Net/Sinks.cpp index e711d61bc6..2325fb2b19 100644 --- a/Common/Net/Sinks.cpp +++ b/Common/Net/Sinks.cpp @@ -213,6 +213,13 @@ void InputSink::Fill() { return; } + if (bytes == 0 && avail > 0) { + // recv() returning 0 on a non-empty request means the peer closed. Sticky: nothing more + // will ever arrive, and anything waiting for more has to stop rather than poll forever. + atEnd_ = true; + return; + } + // Okay, move forward (might be by zero.) valid_ += bytes; write_ += bytes; @@ -223,12 +230,17 @@ void InputSink::Fill() { } bool InputSink::Block() { + if (atEnd_ || hasError_) { + // Nothing can arrive any more. Without this we'd spin: a closed socket is always "ready", + // so WaitUntilReady returns immediately and Fill() reads nothing, forever. + return false; + } if (!fd_util::WaitUntilReady((int)fd_, 5.0)) { return false; } Fill(); - return true; + return !atEnd_; } void InputSink::AccountDrain(size_t bytes) { diff --git a/Common/Net/Sinks.h b/Common/Net/Sinks.h index 09abd1ed51..4fd621b03e 100644 --- a/Common/Net/Sinks.h +++ b/Common/Net/Sinks.h @@ -28,6 +28,9 @@ public: bool Empty() const; bool TryFill(); bool HasError() const { return hasError_; } + // True once the peer has closed its end. Sticky - no more data can ever arrive, which is not + // the same as "nothing right now", and callers that wait for more need to tell them apart. + bool AtEnd() const { return atEnd_; } size_t ValidAmount() const { return valid_; @@ -58,6 +61,7 @@ private: size_t write_; size_t valid_; bool hasError_ = false; + bool atEnd_ = false; }; class OutputSink { diff --git a/Common/Net/WebsocketServer.cpp b/Common/Net/WebsocketServer.cpp index 875255a9cc..5a3f31a25d 100644 --- a/Common/Net/WebsocketServer.cpp +++ b/Common/Net/WebsocketServer.cpp @@ -123,6 +123,8 @@ WebSocketServer *WebSocketServer::CreateAsUpgrade(const http::ServerRequest &req void WebSocketServer::Send(const std::string &str) { _assert_(open_); + if (SendingIsOver()) + return; _assert_(fragmentOpcode_ == -1); SendHeader(true, (int)Opcode::TEXT, str.size()); SendBytes(str.c_str(), str.size()); @@ -130,6 +132,8 @@ void WebSocketServer::Send(const std::string &str) { void WebSocketServer::Send(const std::vector &payload) { _assert_(open_); + if (SendingIsOver()) + return; _assert_(fragmentOpcode_ == -1); SendHeader(true, (int)Opcode::BINARY, payload.size()); SendBytes((const char *)payload.data(), payload.size()); @@ -137,6 +141,8 @@ void WebSocketServer::Send(const std::vector &payload) { void WebSocketServer::AddFragment(bool finish, const std::string &str) { _assert_(open_); + if (SendingIsOver()) + return; if (fragmentOpcode_ == -1) { SendHeader(finish, (int)Opcode::TEXT, str.size()); fragmentOpcode_ = (int)Opcode::TEXT; @@ -153,6 +159,8 @@ void WebSocketServer::AddFragment(bool finish, const std::string &str) { void WebSocketServer::AddFragment(bool finish, const std::vector &payload) { _assert_(open_); + if (SendingIsOver()) + return; if (fragmentOpcode_ == -1) { SendHeader(finish, (int)Opcode::BINARY, payload.size()); fragmentOpcode_ = (int)Opcode::BINARY; @@ -169,6 +177,8 @@ void WebSocketServer::AddFragment(bool finish, const std::vector &paylo void WebSocketServer::Ping(const std::vector &payload) { _assert_(open_); + if (SendingIsOver()) + return; _assert_(payload.size() <= 125); SendHeader(true, (int)Opcode::PING, payload.size()); SendBytes((const char *)payload.data(), payload.size()); @@ -176,12 +186,18 @@ void WebSocketServer::Ping(const std::vector &payload) { void WebSocketServer::Pong(const std::vector &payload) { _assert_(open_); + if (SendingIsOver()) + return; _assert_(payload.size() <= 125); SendHeader(true, (int)Opcode::PONG, payload.size()); SendBytes((const char *)payload.data(), payload.size()); } void WebSocketServer::Close(WebSocketClose reason) { + if (sentClose_) { + // Already closing - a second close frame would just be more data we can't send. + return; + } closeReason_ = reason; if (reason == WebSocketClose::NO_STATUS) { // This means we received a CLOSE without a code. @@ -200,6 +216,16 @@ void WebSocketServer::Close(WebSocketClose reason) { sentClose_ = true; } +// The connection is gone or unusable - there's nothing left to send and no point waiting for +// anything, so drop whatever is queued and let the caller's loop end. +void WebSocketServer::CloseAbnormally() { + closeReason_ = WebSocketClose::ABNORMAL; + open_ = false; + out_->Discard(); + outBuf_.clear(); + outBufOffset_ = 0; +} + bool WebSocketServer::Process(float timeout) { if (!open_) { return false; @@ -212,11 +238,7 @@ bool WebSocketServer::Process(float timeout) { // keep waiting for it to empty: select() reports an errored socket as ready every single // time and SendFlush() can't make progress, so we'd return true forever and the caller // would sit in a tight loop burning a core. - closeReason_ = WebSocketClose::ABNORMAL; - open_ = false; - out_->Discard(); - outBuf_.clear(); - outBufOffset_ = 0; + CloseAbnormally(); return false; } @@ -256,11 +278,7 @@ bool WebSocketServer::Process(float timeout) { // Anything else isn't going to fix itself (a bad fd, say), and returning true on a call // that fails immediately means the caller busy-loops instead of being paced by the timeout. ERROR_LOG(Log::IO, "WebSocket select() failed: %d - closing connection", err); - closeReason_ = WebSocketClose::ABNORMAL; - open_ = false; - out_->Discard(); - outBuf_.clear(); - outBufOffset_ = 0; + CloseAbnormally(); return false; } @@ -273,17 +291,27 @@ bool WebSocketServer::Process(float timeout) { SendFlush(); } if (FD_ISSET(fd_, &read)) { - if (in_->Empty() && !in_->TryFill()) { - // Since select said it was readable, we assume this means disconnect. - closeReason_ = WebSocketClose::ABNORMAL; - open_ = false; - // Kill any remaining output too. - out_->Discard(); + // Fill even when the sink still holds bytes. Otherwise a disconnect goes unnoticed for as + // long as there are leftovers, and if those leftovers are half a frame, ReadFrames() ends + // up waiting inside TakeExact() for bytes that can never arrive. + if (!in_->TryFill()) { + // select() said readable and there's still nothing, so the peer is gone. + CloseAbnormally(); return false; } + // Note this before draining - the peer can close with data still in flight, and we want to + // hand over what it did send before acting on the disconnect. + const bool atEnd = in_->AtEnd(); + while (ReadFrames() && !in_->Empty()) continue; + + if (atEnd && in_->Empty()) { + // Consumed everything they sent, and nothing more is coming. + CloseAbnormally(); + return false; + } } return true; @@ -297,6 +325,16 @@ bool WebSocketServer::ReadFrames() { return ReadFrame(); } +// A read came up short. TakeExact() only tells us it failed, not why - so ask the sink whether +// the peer is simply gone, rather than blaming it for a protocol violation it didn't commit. +void WebSocketServer::CloseForReadFailure() { + if (in_->AtEnd() || in_->HasError()) { + Close(WebSocketClose::ABNORMAL); + } else { + Close(WebSocketClose::POLICY_VIOLATION); + } +} + bool WebSocketServer::ReadFrame() { _assert_(pendingLeft_ == 0); @@ -304,7 +342,7 @@ bool WebSocketServer::ReadFrame() { auto readExact = [&](void *p, size_t sz) { if (!in_->TakeExact((char *)p, sz)) { // TODO: Failing on too slow trickle timeout for now. - Close(WebSocketClose::POLICY_VIOLATION); + CloseForReadFailure(); return false; } return true; @@ -422,6 +460,12 @@ bool WebSocketServer::ReadPending() { // Truncate out the unread bytes for next time. pendingBuf_.resize(pos + readBytes); + + if (in_->AtEnd() || in_->HasError()) { + // TakeAtMost() returns 0 both for "nothing right now" and "nothing ever again", so we + // have to ask the sink which it was. The rest of this message is never arriving. + return false; + } return true; } @@ -454,7 +498,7 @@ bool WebSocketServer::ReadControlFrame(int opcode, size_t sz) { // Just block here to read the payload. if (!in_->TakeExact((char *)payload.data(), sz)) { // TODO: Failing on too slow trickle timeout for now. - Close(WebSocketClose::POLICY_VIOLATION); + CloseForReadFailure(); return false; } diff --git a/Common/Net/WebsocketServer.h b/Common/Net/WebsocketServer.h index a7aecbe38f..398fa32dff 100644 --- a/Common/Net/WebsocketServer.h +++ b/Common/Net/WebsocketServer.h @@ -73,6 +73,13 @@ protected: void SendHeader(bool fin, int opcode, size_t sz); void SendBytes(const void *p, size_t sz); + // True once we've queued our own close frame. RFC 6455 5.5.1: no data frames may follow it. + // Just as importantly, anything queued after that point keeps outBuf_/out_ non-empty, which + // starves the "everything is flushed" check in Process() so the connection never finishes + // closing - and since the socket stays writable, select() stops blocking and we spin. + bool SendingIsOver() const { return sentClose_; } + void CloseAbnormally(); + void CloseForReadFailure(); void SendFlush(); void CompactOutBuf(); size_t OutBufPending() const { return outBuf_.size() - outBufOffset_; }