From 03fa890a2e1206b4237ddb37fa7850fe1e85cfdd Mon Sep 17 00:00:00 2001 From: yhirose Date: Sun, 18 Oct 2026 13:19:15 -0400 Subject: [PATCH] Revert "Let SSEClient::stop() interrupt a blocked read" This reverts commit 3744227de2cc25be2e65d5c4620e5485ac3527a7. Shutting down the socket does not wake a read waiting in WSAPoll on Windows while the peer stays silent, so stop() still waited out the read timeout there and the new tests failed. Faster stop() on the other platforms also let SSEIntegrationTest.ManualStop return while its server handler still read a local of the test body (stack-use-after-return). The fix will come back built on a wait that works on every platform. --- httplib.h | 32 ----------------------- test/test.cc | 72 ---------------------------------------------------- 2 files changed, 104 deletions(-) diff --git a/httplib.h b/httplib.h index eff46786..6ad8d1b5 100644 --- a/httplib.h +++ b/httplib.h @@ -4494,10 +4494,6 @@ private: std::atomic connected_{false}; std::string last_event_id_; - // Socket of the open stream, for stop() to shut down - std::mutex stream_mutex_; - socket_t stream_sock_ = INVALID_SOCKET; - // Async support std::thread async_thread_; }; @@ -5005,20 +5001,6 @@ inline void SSEClient::start_async() { inline void SSEClient::stop() { running_.store(false); - { - // Ends a blocked read. The loop withdraws the socket under this lock - // before closing it, so it is still open here. - std::lock_guard lock(stream_mutex_); - if (stream_sock_ != INVALID_SOCKET) { -#ifdef _WIN32 - ::shutdown(stream_sock_, SD_BOTH); -#else - // Only the read side, so that the alert TLS sends on the unexpected EOF - // does not raise SIGPIPE - ::shutdown(stream_sock_, SHUT_RD); -#endif - } - } client_.stop(); // Cancel any pending operations if (async_thread_.joinable()) { async_thread_.join(); } } @@ -5093,7 +5075,6 @@ 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; - auto sock = handle.connection_ ? handle.connection_->sock : INVALID_SOCKET; stream::Result result{std::move(handle)}; // Connection error handling @@ -5126,19 +5107,6 @@ inline void SSEClient::run_event_loop() { continue; } - // Hand the socket to stop(), unless it has already been called - { - std::lock_guard lock(stream_mutex_); - if (!running_.load()) { break; } - stream_sock_ = sock; - } - - // Destroyed before result, so the socket is withdrawn before it is closed - detail::scope_exit withdraw_sock([this] { - std::lock_guard lock(stream_mutex_); - stream_sock_ = INVALID_SOCKET; - }); - // Connection successful connected_.store(true); reconnect_count = 0; diff --git a/test/test.cc b/test/test.cc index 7bedb460..40c7dd1d 100644 --- a/test/test.cc +++ b/test/test.cc @@ -25637,78 +25637,6 @@ TEST_F(SSEIntegrationTest, ManualStop) { 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