From 801bf2e0d23b2145a190845d01db24f2ff4aa434 Mon Sep 17 00:00:00 2001 From: yhirose Date: Sun, 18 Oct 2026 14:01:59 -0400 Subject: [PATCH] Let SSEClient::stop() interrupt a blocked read The event loop reads through open_stream(), which takes the socket out of the Client, so the client_.stop() in stop() could not reach it. stop(), and the destructor with it, waited until the server sent something or the read timeout passed, 300 seconds by default. Shutting the socket down from stop() does not wake a read waiting in WSAPoll on Windows, so 3744227 was reverted. A stream now takes an interrupt check instead: while one is set, it waits for data in 100 ms slices and gives up once the check returns true, as the server's keep-alive wait does with its flag. SSEClient sets one that turns true when stop() is called. Streams without one wait as before. SSEIntegrationTest.ManualStop now ends its stream with the fixture's flag. Its server outlives the test body, and the faster stop() let the stream read a local of the body that had returned. --- httplib.h | 56 ++++++++++++++++++++++++++++++--- test/test.cc | 87 ++++++++++++++++++++++++++++++++++++++++++++++------ 2 files changed, 130 insertions(+), 13 deletions(-) diff --git a/httplib.h b/httplib.h index 6ad8d1b5..2c709d52 100644 --- a/httplib.h +++ b/httplib.h @@ -2020,6 +2020,12 @@ public: // each one. 0 lifts the bound. virtual void set_max_timeout(time_t msec) { (void)msec; } + // Lets a wait for data end early once `interrupted` returns true, so that + // another thread can stop a blocked read on every platform. + virtual void set_read_interrupt(std::function interrupted) { + (void)interrupted; + } + // Bytes already pulled off the socket and sitting in this stream's own // buffer. Exposing them lets a line reader scan for a terminator in one // pass instead of asking for a byte at a time. A stream that does no @@ -5075,6 +5081,10 @@ inline void SSEClient::run_event_loop() { // bounded instead of the whole body. auto handle = client_.open_stream("GET", path_, {}, request_headers); handle.body_reader_.payload_max_length = 0; + // stop() ends a read that is waiting for the next event + if (handle.stream_) { + handle.stream_->set_read_interrupt([this] { return !running_.load(); }); + } stream::Result result{std::move(handle)}; // Connection error handling @@ -6740,6 +6750,26 @@ inline ssize_t select_until(socket_t sock, short events, static_cast(left % 1000000)); } +// select_read(), but waits in slices and gives up once interrupted() returns +// true. A local shutdown() would not wake the wait on Windows. +inline ssize_t +select_read_interruptible(socket_t sock, time_t sec, time_t usec, + const std::function &interrupted) { + if (!interrupted) { return select_read(sock, sec, usec); } + + const auto slice = std::chrono::microseconds(100000); + auto deadline = deadline_after(sec, usec); + while (!interrupted()) { + auto left = std::chrono::duration_cast( + deadline - std::chrono::steady_clock::now()); + auto wait = + (std::max)((std::min)(left, slice), std::chrono::microseconds(0)); + auto ret = select_impl(sock, POLLIN, 0, static_cast(wait.count())); + if (ret != 0 || wait < slice) { return ret; } + } + return 0; +} + inline Error wait_until_socket_is_ready(socket_t sock, time_t sec, time_t usec) { struct pollfd pfd_read; @@ -6798,6 +6828,7 @@ public: time_t duration() const override; void set_read_timeout(time_t sec, time_t usec = 0) override; void set_max_timeout(time_t msec) override; + void set_read_interrupt(std::function interrupted) override; const char *buffered_data(size_t &size) const override; void consume_buffered(size_t size) override; @@ -6819,6 +6850,7 @@ private: time_t write_timeout_usec_; time_t max_timeout_msec_; std::chrono::time_point start_time_; + std::function read_interrupted_; std::vector read_buff_; size_t read_buff_off_ = 0; @@ -10679,6 +10711,7 @@ public: time_t duration() const override; void set_read_timeout(time_t sec, time_t usec = 0) override; void set_max_timeout(time_t msec) override; + void set_read_interrupt(std::function interrupted) override; // See SocketStream::set_readable_hint(). void set_readable_hint() { readable_hint_ = true; } @@ -10698,6 +10731,7 @@ private: time_t write_timeout_usec_; time_t max_timeout_msec_; std::chrono::time_point start_time_; + std::function read_interrupted_; bool readable_hint_ = false; tls::TlsError tls_error_; }; @@ -12610,7 +12644,8 @@ inline bool SocketStream::is_readable() const { inline bool SocketStream::wait_readable() const { if (max_timeout_msec_ <= 0) { - 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; } time_t read_timeout_sec; @@ -12618,7 +12653,8 @@ inline bool SocketStream::wait_readable() const { calc_actual_timeout(max_timeout_msec_, duration(), read_timeout_sec_, read_timeout_usec_, read_timeout_sec, read_timeout_usec); - 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 SocketStream::wait_writable() const { @@ -12749,6 +12785,11 @@ inline void SocketStream::set_max_timeout(time_t msec) { start_time_ = std::chrono::steady_clock::now(); } +inline void +SocketStream::set_read_interrupt(std::function interrupted) { + read_interrupted_ = std::move(interrupted); +} + // Buffer stream implementation inline bool BufferStream::is_readable() const { return true; } @@ -13174,7 +13215,8 @@ inline bool SSLSocketStream::is_readable() const { inline bool SSLSocketStream::wait_readable() const { if (max_timeout_msec_ <= 0) { - 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; } time_t read_timeout_sec; @@ -13182,7 +13224,8 @@ inline bool SSLSocketStream::wait_readable() const { calc_actual_timeout(max_timeout_msec_, duration(), read_timeout_sec_, read_timeout_usec_, read_timeout_sec, read_timeout_usec); - 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 SSLSocketStream::wait_writable() const { @@ -13309,6 +13352,11 @@ inline void SSLSocketStream::set_max_timeout(time_t msec) { start_time_ = std::chrono::steady_clock::now(); } +inline void +SSLSocketStream::set_read_interrupt(std::function interrupted) { + read_interrupted_ = std::move(interrupted); +} + inline WebSocketSSLStream::WebSocketSSLStream(socket_t sock, tls::session_t session, time_t read_timeout_sec, diff --git a/test/test.cc b/test/test.cc index 40c7dd1d..75c9c80f 100644 --- a/test/test.cc +++ b/test/test.cc @@ -25593,19 +25593,16 @@ TEST_F(SSEIntegrationTest, LastEventIdHeader) { // Test: Manual stop TEST_F(SSEIntegrationTest, ManualStop) { - // Endpoint that sends one event and stays open briefly - std::atomic handler_running{true}; - - server_->Get("/stay-open", [&handler_running](const Request &, - Response &res) { + // Endpoint that sends one event and stays open briefly. The server outlives + // this test body, so the provider uses the fixture's flag, not a local. + server_->Get("/stay-open", [this](const Request &, Response &res) { res.set_chunked_content_provider( - "text/event-stream", [&handler_running](size_t offset, DataSink &sink) { + "text/event-stream", [this](size_t offset, DataSink &sink) { if (offset == 0) { std::string event = "data: connected\n\n"; sink.write(event.data(), event.size()); } - // Keep connection open while handler_running is true - for (int i = 0; i < 10 && handler_running.load(); ++i) { + for (int i = 0; i < 10 && !stop_server_.load(); ++i) { std::this_thread::sleep_for(std::chrono::milliseconds(50)); } return false; @@ -25630,13 +25627,85 @@ TEST_F(SSEIntegrationTest, ManualStop) { EXPECT_TRUE(sse.is_connected()); // Signal handler to stop - handler_running.store(false); + stop_server_.store(true); // Stop SSE client sse.stop(); EXPECT_FALSE(sse.is_connected()); } +// Sends one event, then nothing until stop is set +static Server::Handler idle_event_stream(std::atomic &stop) { + return [&stop](const Request &, Response &res) { + res.set_chunked_content_provider( + "text/event-stream", [&stop](size_t offset, DataSink &sink) { + if (offset == 0) { + std::string event = "data: hello\n\n"; + sink.write(event.data(), event.size()); + } + while (!stop.load()) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + return false; + }); + }; +} + +// Both stop() and the destructor end a read blocked on /idle-stream promptly +static void check_stop_interrupts_blocked_read(Client &client) { + client.set_read_timeout(30, 0); + + for (auto by_destructor : {false, true}) { + std::atomic received{false}; + std::chrono::steady_clock::time_point start; + { + sse::SSEClient sse(client, "/idle-stream"); + sse.on_message([&](const sse::SSEMessage &) { received.store(true); }); + sse.start_async(); + + for (int i = 0; i < 500 && !received.load(); ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + ASSERT_TRUE(received.load()); + // Let the client block in the next read + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + start = std::chrono::steady_clock::now(); + if (!by_destructor) { sse.stop(); } + } + EXPECT_LT(std::chrono::steady_clock::now() - start, + std::chrono::seconds(2)); + } +} + +TEST_F(SSEIntegrationTest, StopInterruptsBlockedRead) { + server_->Get("/idle-stream", idle_event_stream(stop_server_)); + + Client client("localhost", get_port()); + check_stop_interrupts_blocked_read(client); +} + +#ifdef CPPHTTPLIB_SSL_ENABLED +TEST(SSEClientSSLTest, StopInterruptsBlockedRead) { + SSLServer svr(SERVER_CERT_FILE, SERVER_PRIVATE_KEY_FILE); + std::atomic done{false}; + svr.Get("/idle-stream", idle_event_stream(done)); + + auto port = svr.bind_to_any_port(HOST); + std::thread t([&] { svr.listen_after_bind(); }); + auto se = detail::scope_exit([&] { + done.store(true); + svr.stop(); + t.join(); + }); + svr.wait_until_ready(); + + Client client("https://localhost:" + std::to_string(port)); + client.enable_server_certificate_verification(false); + check_stop_interrupts_blocked_read(client); +} +#endif + // Test: SSEClient with custom headers TEST_F(SSEIntegrationTest, CustomHeaders) { // Setup a server endpoint that checks for custom header