Revert "Let SSEClient::stop() interrupt a blocked read"

This reverts commit 3744227de2.

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.
This commit is contained in:
yhirose committed 2026-10-18 13:19:15 -04:00
1 parent 1d5f0323b1
commit 03fa890a2e
2 files changed
-104

No files matched your search

-32
View File
@@ -4494,10 +4494,6 @@ private:
std::atomic<bool> connected_{false}; std::atomic<bool> connected_{false};
std::string last_event_id_; 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 // Async support
std::thread async_thread_; std::thread async_thread_;
}; };
@@ -5005,20 +5001,6 @@ inline void SSEClient::start_async() {
inline void SSEClient::stop() { inline void SSEClient::stop() {
running_.store(false); 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<std::mutex> 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 client_.stop(); // Cancel any pending operations
if (async_thread_.joinable()) { async_thread_.join(); } if (async_thread_.joinable()) { async_thread_.join(); }
} }
@@ -5093,7 +5075,6 @@ inline void SSEClient::run_event_loop() {
// bounded instead of the whole body. // bounded instead of the whole body.
auto handle = client_.open_stream("GET", path_, {}, request_headers); auto handle = client_.open_stream("GET", path_, {}, request_headers);
handle.body_reader_.payload_max_length = 0; handle.body_reader_.payload_max_length = 0;
auto sock = handle.connection_ ? handle.connection_->sock : INVALID_SOCKET;
stream::Result result{std::move(handle)}; stream::Result result{std::move(handle)};
// Connection error handling // Connection error handling
@@ -5126,19 +5107,6 @@ inline void SSEClient::run_event_loop() {
continue; continue;
} }
// Hand the socket to stop(), unless it has already been called
{
std::lock_guard<std::mutex> 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<std::mutex> lock(stream_mutex_);
stream_sock_ = INVALID_SOCKET;
});
// Connection successful // Connection successful
connected_.store(true); connected_.store(true);
reconnect_count = 0; reconnect_count = 0;
-72
View File
@@ -25637,78 +25637,6 @@ TEST_F(SSEIntegrationTest, ManualStop) {
EXPECT_FALSE(sse.is_connected()); EXPECT_FALSE(sse.is_connected());
} }
// Sends one event, then nothing until stop is set
static Server::Handler idle_event_stream(std::atomic<bool> &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<bool> 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<bool> 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: SSEClient with custom headers
TEST_F(SSEIntegrationTest, CustomHeaders) { TEST_F(SSEIntegrationTest, CustomHeaders) {
// Setup a server endpoint that checks for custom header // Setup a server endpoint that checks for custom header