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.