From 696b94b5eddb90ced4bf28762edf2f1db291a8bb Mon Sep 17 00:00:00 2001 From: yhirose Date: Sun, 18 Oct 2026 17:23:05 -0400 Subject: [PATCH] Wake a WebSocket read on a pong timeout on every platform When max_missed_pongs ran out, the heartbeat thread shut down the read side of the socket to end a read() blocked on the silent peer. On Windows that does not wake the wait, so read() stayed blocked until its read timeout and then returned Timeout. The heartbeat test never ran on Windows, so CI did not catch it. The heartbeat now sets a flag that the stream's read interrupt checks, the mechanism SSEClient::stop() uses, and WebSocketSSLStream takes the interrupt too. It is set only when a pong timeout can happen, so other connections wait as before. read() reports a timeout on a connection that has closed as Fail, since Timeout means it is still usable. Build and run the heartbeat test in the Windows jobs, and add a TLS pong timeout test. --- .github/workflows/test.yaml | 4 +++ httplib.h | 42 +++++++++++++++++---------- test/CMakeLists.txt | 8 ++++++ test/test_websocket_heartbeat.cc | 49 ++++++++++++++++++++++++-------- 4 files changed, 76 insertions(+), 27 deletions(-) diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index ba7cc50f..93fbf758 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -623,6 +623,10 @@ jobs: } if ($failed) { exit 1 } Write-Host "All shards passed." + - name: Run WebSocket heartbeat test ${{ matrix.config.name }} + if: ${{ matrix.config.run_tests }} + working-directory: build/test + run: ./Release/httplib-heartbeat-test.exe --gtest_color=yes env: VCPKG_ROOT: "C:/vcpkg" diff --git a/httplib.h b/httplib.h index 2c709d52..c2d46ff2 100644 --- a/httplib.h +++ b/httplib.h @@ -4645,6 +4645,8 @@ private: int max_missed_pongs_; int unacked_pings_ = 0; std::atomic closed_{false}; + // Set when the peer stopped answering pings, to end a read() waiting on it + std::atomic pong_timed_out_{false}; // Set once the caller has bounded read() through set_read_timeout(). Until // then the timeout in effect is the compile-time default, and elapsing it // is a failure that closes the connection, not a Timeout. @@ -10766,6 +10768,7 @@ public: socket_t socket() const override; time_t duration() const override; void set_read_timeout(time_t sec, time_t usec = 0) override; + void set_read_interrupt(std::function interrupted) override; private: mutable std::mutex session_mutex_; @@ -10780,6 +10783,7 @@ private: time_t write_timeout_sec_; time_t write_timeout_usec_; const std::chrono::time_point start_time_; + std::function read_interrupted_; }; #ifdef CPPHTTPLIB_OPENSSL_SUPPORT @@ -13386,7 +13390,8 @@ inline bool WebSocketSSLStream::is_readable() const { } inline bool WebSocketSSLStream::wait_readable() const { - return select_read(sock_, read_timeout_sec_, read_timeout_usec_) > 0; + return select_read_interruptible(sock_, read_timeout_sec_, read_timeout_usec_, + read_interrupted_) > 0; } inline bool WebSocketSSLStream::wait_writable() const { @@ -13486,6 +13491,11 @@ inline void WebSocketSSLStream::set_read_timeout(time_t sec, time_t usec) { read_timeout_usec_ = usec; } +inline void +WebSocketSSLStream::set_read_interrupt(std::function interrupted) { + read_interrupted_ = std::move(interrupted); +} + } // namespace detail #endif // CPPHTTPLIB_SSL_ENABLED @@ -23331,12 +23341,15 @@ inline ReadResult WebSocket::read(std::string &msg) { impl::FrameRead r = impl::read_websocket_frame(strm_, opcode, payload, fin, is_server_, CPPHTTPLIB_WEBSOCKET_MAX_PAYLOAD_LENGTH); - // A timeout landed on a frame boundary: the connection is untouched and - // still usable, so hand control back without closing it. That is only - // useful to a caller who asked for the timeout; the compile-time default - // is a backstop against a peer gone quiet, and elapsing it closes the - // connection so a plain `while (ws.read(msg))` loop ends. - if (r == impl::FrameRead::Timeout && read_timeout_set_) { return Timeout; } + // A timeout landed on a frame boundary of an open connection: it is + // untouched and still usable, so hand control back without closing it. + // That is only useful to a caller who asked for the timeout; the + // compile-time default is a backstop against a peer gone quiet, and + // elapsing it closes the connection so a plain `while (ws.read(msg))` loop + // ends. A wait that a pong timeout cut short finds the connection closed. + if (r == impl::FrameRead::Timeout && read_timeout_set_ && !closed_) { + return Timeout; + } if (r != impl::FrameRead::Ok) { closed_ = true; return Fail; @@ -23465,10 +23478,16 @@ inline WebSocket::~WebSocket() { } ping_cv_.notify_all(); if (ping_thread_.joinable()) { ping_thread_.join(); } + // A server's stream outlives this object + strm_.set_read_interrupt(nullptr); } inline void WebSocket::start_heartbeat() { if (ping_interval_sec_ <= 0) { return; } + // A pong timeout ends a read() blocked on the silent peer + if (max_missed_pongs_ > 0) { + strm_.set_read_interrupt([this] { return pong_timed_out_.load(); }); + } ping_thread_ = std::thread([this]() { std::unique_lock lock(ping_mutex_); // The predicate keeps a spurious wakeup from sending a ping early @@ -23480,14 +23499,7 @@ inline void WebSocket::start_heartbeat() { if (max_missed_pongs_ > 0 && unacked_pings_ >= max_missed_pongs_) { lock.unlock(); close(CloseStatus::GoingAway, "pong timeout"); - // Wake a read() blocked on the unresponsive peer. Only the read side - // is shut down: a TLS backend answers the EOF with an alert, and - // writing it to a socket closed for writing raises SIGPIPE. -#ifdef _WIN32 - shutdown(strm_.socket(), SD_RECEIVE); -#else - shutdown(strm_.socket(), SHUT_RD); -#endif + pong_timed_out_ = true; return; } lock.unlock(); diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 63b96ab3..1ae9cdb6 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -31,6 +31,14 @@ target_compile_options(httplib-test PRIVATE "$<$:/utf-8;/b target_link_libraries(httplib-test PRIVATE httplib GTest::gtest_main CURL::libcurl) gtest_discover_tests(httplib-test) +# It overrides the heartbeat macros, which a compiled library has fixed +if(NOT HTTPLIB_COMPILE) + add_executable(httplib-heartbeat-test test_websocket_heartbeat.cc) + target_compile_options(httplib-heartbeat-test PRIVATE "$<$:/utf-8;/bigobj>") + target_link_libraries(httplib-heartbeat-test PRIVATE httplib GTest::gtest_main) + gtest_discover_tests(httplib-heartbeat-test) +endif() + file( COPY www www2 www3 ca-bundle.crt image.jpg DESTINATION ${CMAKE_CURRENT_BINARY_DIR} diff --git a/test/test_websocket_heartbeat.cc b/test/test_websocket_heartbeat.cc index 4242a495..45646fb4 100644 --- a/test/test_websocket_heartbeat.cc +++ b/test/test_websocket_heartbeat.cc @@ -146,16 +146,21 @@ TEST_F(WebSocketServerPingIntervalTest, ServerRuntimeInterval) { // ping count reaches max_missed_pongs. class WebSocketPongTimeoutTest : public ::testing::Test { protected: + virtual std::unique_ptr make_server() { + return std::unique_ptr(new Server()); + } + void SetUp() override { - svr_.set_websocket_ping_interval(0); - svr_.WebSocket("/ws", [this](const Request &, ws::WebSocket &) { + svr_ = make_server(); + svr_->set_websocket_ping_interval(0); + svr_->WebSocket("/ws", [this](const Request &, ws::WebSocket &) { std::unique_lock lock(handler_mutex_); handler_cv_.wait(lock, [this]() { return release_; }); }); - port_ = svr_.bind_to_any_port("localhost"); - thread_ = std::thread([this]() { svr_.listen_after_bind(); }); - svr_.wait_until_ready(); + port_ = svr_->bind_to_any_port("localhost"); + thread_ = std::thread([this]() { svr_->listen_after_bind(); }); + svr_->wait_until_ready(); } void TearDown() override { @@ -164,11 +169,11 @@ protected: release_ = true; } handler_cv_.notify_all(); - svr_.stop(); + svr_->stop(); thread_.join(); } - Server svr_; + std::unique_ptr svr_; int port_; std::thread thread_; std::mutex handler_mutex_; @@ -176,8 +181,10 @@ protected: bool release_ = false; }; -TEST_F(WebSocketPongTimeoutTest, ClientDetectsNonResponsivePeer) { - ws::WebSocketClient client("ws://localhost:" + std::to_string(port_) + "/ws"); +// Client pings every 1s (compile-time default in this test file). With +// max_missed_pongs = 2, the heartbeat thread should self-close within roughly +// 3s, and that has to end a read() already waiting on the peer. +static void expect_pong_timeout_ends_read(ws::WebSocketClient &client) { client.set_websocket_max_missed_pongs(2); // A read timeout asked for at runtime is reported as Timeout, so it cannot // be mistaken for the Fail a pong timeout produces. @@ -185,9 +192,6 @@ TEST_F(WebSocketPongTimeoutTest, ClientDetectsNonResponsivePeer) { ASSERT_TRUE(client.connect()); ASSERT_TRUE(client.is_open()); - // Client pings every 1s (compile-time default in this test file). - // With max_missed_pongs = 2, the heartbeat thread should self-close within - // roughly 3s, and that has to end a read() already waiting on the peer. auto start = std::chrono::steady_clock::now(); std::string msg; EXPECT_EQ(client.read(msg), ws::Fail); @@ -196,6 +200,27 @@ TEST_F(WebSocketPongTimeoutTest, ClientDetectsNonResponsivePeer) { EXPECT_FALSE(client.is_open()); } +TEST_F(WebSocketPongTimeoutTest, ClientDetectsNonResponsivePeer) { + ws::WebSocketClient client("ws://localhost:" + std::to_string(port_) + "/ws"); + expect_pong_timeout_ends_read(client); +} + +#ifdef CPPHTTPLIB_SSL_ENABLED +class WebSocketPongTimeoutSSLTest : public WebSocketPongTimeoutTest { +protected: + std::unique_ptr make_server() override { + return std::unique_ptr(new SSLServer("./cert.pem", "./key.pem")); + } +}; + +TEST_F(WebSocketPongTimeoutSSLTest, ClientDetectsNonResponsivePeer) { + ws::WebSocketClient client("wss://localhost:" + std::to_string(port_) + + "/ws"); + client.enable_server_certificate_verification(false); + expect_pong_timeout_ends_read(client); +} +#endif + // The compile-time client read timeout (3s here) was never asked for through // set_read_timeout(), so when it elapses read() reports Fail and closes the // connection rather than handing back a Timeout on a still-open one.