Merge pull request #306 from sccn/fix/sync-idle-disconnect

Fix idle synchronous consumer disconnect detection

Author: cboulay

src/tcp_server.cpp

@@ -27,6 +27,9 @@
 #include <thread>
 #include <utility>
 #include <vector>
+#ifndef _WIN32
+#include <poll.h>
+#endif
 
 #ifdef _MSC_VER
 // (inefficiently converting int to bool in portable_oarchive instantiation...)
@@ -157,12 +160,37 @@ class sync_write_handler {
 	}
 
 	/// Check if there are any connected consumers
-	bool have_consumers() const {
+	bool have_consumers() {
 		std::lock_guard<std::mutex> lock(mutex_);
+		// Sync sockets have no running IO context. Check on demand, without changing
+		// their native blocking mode (which SO_SNDTIMEO relies on).
+		prune_disconnected(sockets_native_);
+		prune_disconnected(sockets_swapped_);
 		return !sockets_native_.empty() || !sockets_swapped_.empty();
 	}
 
 private:
+	/// Called under mutex_, which also serializes socket writes and registration.
+	static void prune_disconnected(std::vector<tcp_socket_p> &sockets) {
+		sockets.erase(std::remove_if(sockets.begin(), sockets.end(), [](const tcp_socket_p &sock) {
+			if (!sock || !sock->is_open()) return true;
+			// After the handshake the feed is one-way: readability means EOF,
+			// a reset, or unexpected client data. All end the feed, as in the
+			// asynchronous peer-close watcher. Never issue a blocking receive.
+#ifdef _WIN32
+			fd_set readable;
+			FD_ZERO(&readable);
+			FD_SET(sock->native_handle(), &readable);
+			timeval timeout{};
+			return ::select(0, &readable, nullptr, nullptr, &timeout) > 0;
+#else
+			// poll also handles descriptors beyond select's FD_SETSIZE limit.
+			pollfd fd{sock->native_handle(), POLLIN, 0};
+			return ::poll(&fd, 1, 0) > 0 && (fd.revents & (POLLIN | POLLHUP | POLLERR));
+#endif
+		}), sockets.end());
+	}
+
 	/// Apply the configured blocking-send timeout to a socket (no-op if disabled). SO_SNDTIMEO
 	/// only takes effect on a blocking socket, so we force blocking mode first. On expiry a
 	/// synchronous write returns try_again/would_block/timed_out, which write_to_group treats

testing/ext/outlet.cpp

@@ -19,7 +19,9 @@ bool wait_until_no_consumers(lsl::stream_outlet &outlet, double timeout_sec) {
 TEST_CASE("have_consumers becomes false after idle inlet disconnects", "[outlet][basic]") {
 	lsl::stream_info info("have_consumers_idle", "Markers", 1, lsl::IRREGULAR_RATE, lsl::cf_int32,
 		"have_consumers_idle");
-	lsl::stream_outlet outlet(info);
+	const auto transport = GENERATE(transp_default, transp_sync_blocking);
+	CAPTURE(transport);
+	lsl::stream_outlet outlet(info, 0, 360, transport);
 	auto found = lsl::resolve_stream("name", info.name(), 1, 2.0);
 	REQUIRE(found.size() == 1);
 
@@ -31,13 +33,23 @@ TEST_CASE("have_consumers becomes false after idle inlet disconnects", "[outlet]
 		// Intentionally do not push samples; disconnect must still be detected (#267).
 	}
 
-	CHECK(wait_until_no_consumers(outlet, 2.0));
+	SECTION("have_consumers detects the disconnect") {
+		CHECK(wait_until_no_consumers(outlet, 2.0));
+	}
+	SECTION("wait_for_consumers does not accept the stale connection") {
+		auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
+		while (outlet.wait_for_consumers(0.0) && std::chrono::steady_clock::now() < deadline)
+			std::this_thread::sleep_for(std::chrono::milliseconds(10));
+		CHECK_FALSE(outlet.wait_for_consumers(0.05));
+	}
 }
 
 TEST_CASE("have_consumers becomes false after disconnect during push", "[outlet][basic]") {
 	lsl::stream_info info("have_consumers_busy", "Markers", 1, lsl::IRREGULAR_RATE, lsl::cf_int32,
 		"have_consumers_busy");
-	lsl::stream_outlet outlet(info);
+	const auto transport = GENERATE(transp_default, transp_sync_blocking);
+	CAPTURE(transport);
+	lsl::stream_outlet outlet(info, 0, 360, transport);
 	auto found = lsl::resolve_stream("name", info.name(), 1, 2.0);
 	REQUIRE(found.size() == 1);
 
@@ -76,7 +88,9 @@ TEST_CASE("have_consumers becomes false after disconnect during push", "[outlet]
 TEST_CASE("have_consumers tracks explicit close and reopen", "[outlet][basic]") {
 	lsl::stream_info info("have_consumers_reopen", "Markers", 1, lsl::IRREGULAR_RATE,
 		lsl::cf_int32, "have_consumers_reopen");
-	lsl::stream_outlet outlet(info);
+	const auto transport = GENERATE(transp_default, transp_sync_blocking);
+	CAPTURE(transport);
+	lsl::stream_outlet outlet(info, 0, 360, transport);
 	auto found = lsl::resolve_stream("name", info.name(), 1, 2.0);
 	REQUIRE(found.size() == 1);
 
@@ -94,7 +108,9 @@ TEST_CASE("have_consumers tracks explicit close and reopen", "[outlet][basic]")
 TEST_CASE("have_consumers stays true when one of two inlets disconnects", "[outlet][basic]") {
 	lsl::stream_info info("have_consumers_two", "Markers", 1, lsl::IRREGULAR_RATE, lsl::cf_int32,
 		"have_consumers_two");
-	lsl::stream_outlet outlet(info);
+	const auto transport = GENERATE(transp_default, transp_sync_blocking);
+	CAPTURE(transport);
+	lsl::stream_outlet outlet(info, 0, 360, transport);
 	auto found = lsl::resolve_stream("name", info.name(), 1, 2.0);
 	REQUIRE(found.size() == 1);