DSU: Simplify code and use wakeup packet

Neither SO_RCVTIMEO or closing the socket mid-receive are really cross-platform ways to get control of the reader thread, so lets just send a dummy loopback packet to wake it up

src/input/api/DSU/DSUControllerProvider.cpp

@@ -42,25 +42,24 @@ DSUControllerProvider::~DSUControllerProvider()
 	if (m_running)
 	{
 		m_running = false;
-
-		boost::system::error_code ec;
-		m_socket.shutdown(boost::asio::ip::udp::socket::shutdown_both, ec);
-		m_socket.close(ec); 
-
-		if (m_reader_thread.joinable())
+		// wake up the reader thread by sending a packet to self
+		if (m_socketWakeupEndpoint.port() != 0)
 		{
-			m_reader_thread.join();
-		}
-
-		{
-			std::scoped_lock lock(m_writer_mutex);
-			m_writer_cond.notify_all();
-		}
-
-		if (m_writer_thread.joinable())
-		{
-			m_writer_thread.join();
+			boost::asio::io_context io_context;
+			boost::asio::ip::udp::socket socket(io_context);
+			boost::system::error_code ec;
+			socket.open(m_socketWakeupEndpoint.protocol(), ec);
+			if (!ec)
+			{
+				std::array<char, 1> data{};
+				socket.send_to(boost::asio::buffer(data), m_socketWakeupEndpoint, 0, ec);
+			}
+			else
+				cemuLog_log(LogType::Force, "DSUControllerProvider wakeup failed");
 		}
+		m_reader_thread.join();
+		m_writerJobs.push(nullptr); // wake up writer thread by pushing an empty message
+		m_writer_thread.join();
 	}
 }
 
@@ -101,14 +100,8 @@ bool DSUControllerProvider::connect()
 			m_socket.close();
 
 		m_socket.open(ip::udp::v4());
-
-        // set timeout for our threads to give a chance to exit
-#if BOOST_OS_WINDOWS
-        m_socket.set_option(boost::asio::detail::socket_option::integer<SOL_SOCKET, SO_RCVTIMEO>{200});
-#elif BOOST_OS_LINUX || BOOST_OS_MACOS
-        timeval timeout{.tv_usec = 200 * 1000};
-        setsockopt(m_socket.native_handle(), SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeval));
-#endif
+		m_socket.bind(ip::udp::endpoint(ip::udp::v4(), 0));
+		m_socketWakeupEndpoint = ip::udp::endpoint(boost::asio::ip::address_v4::loopback(), m_socket.local_endpoint().port());
 		// reset data
 		m_state = {};
 		m_prev_state = {};
@@ -206,19 +199,13 @@ uint32_t DSUControllerProvider::get_packet_index(uint8_t index) const
 void DSUControllerProvider::request_version()
 {
 	auto msg = std::make_unique<VersionRequest>(m_uid);
-
-	std::scoped_lock lock(m_writer_mutex);
-	m_writer_jobs.push(std::move(msg));
-	m_writer_cond.notify_one();
+	m_writerJobs.push(std::move(msg));
 }
 
 void DSUControllerProvider::request_pad_info()
 {
 	auto msg = std::make_unique<ListPorts>(m_uid, 4, std::array<uint8_t, 4>{0, 1, 2, 3});
-
-	std::scoped_lock lock(m_writer_mutex);
-	m_writer_jobs.push(std::move(msg));
-	m_writer_cond.notify_one();
+	m_writerJobs.push(std::move(msg));
 }
 
 void DSUControllerProvider::request_pad_info(uint8_t index)
@@ -227,19 +214,13 @@ void DSUControllerProvider::request_pad_info(uint8_t index)
 		return;
 
 	auto msg = std::make_unique<ListPorts>(m_uid, 1, std::array<uint8_t, 4>{index});
-
-	std::scoped_lock lock(m_writer_mutex);
-	m_writer_jobs.push(std::move(msg));
-	m_writer_cond.notify_one();
+	m_writerJobs.push(std::move(msg));
 }
 
 void DSUControllerProvider::request_pad_data()
 {
 	auto msg = std::make_unique<DataRequest>(m_uid);
-
-	std::scoped_lock lock(m_writer_mutex);
-	m_writer_jobs.push(std::move(msg));
-	m_writer_cond.notify_one();
+	m_writerJobs.push(std::move(msg));
 }
 
 void DSUControllerProvider::request_pad_data(uint8_t index)
@@ -248,10 +229,7 @@ void DSUControllerProvider::request_pad_data(uint8_t index)
 		return;
 
 	auto msg = std::make_unique<DataRequest>(m_uid, index);
-
-	std::scoped_lock lock(m_writer_mutex);
-	m_writer_jobs.push(std::move(msg));
-	m_writer_cond.notify_one();
+	m_writerJobs.push(std::move(msg));
 }
 
 MotionSample DSUControllerProvider::get_motion_sample(uint8_t index) const
@@ -276,12 +254,10 @@ void DSUControllerProvider::reader_thread()
 		boost::asio::ip::udp::endpoint sender_endpoint;
 		boost::system::error_code ec{};
 		const size_t len = m_socket.receive_from(boost::asio::buffer(recv_buf), sender_endpoint, 0, ec);
+		if (!m_running.load(std::memory_order_relaxed))
+			break;
 		if (ec)
 		{
-			if (!m_running.load(std::memory_order_relaxed))
-			{
-				break;
-			}
 
 #ifdef DEBUG_DSU_CLIENT
 				printf(" DSUControllerProvider::ReaderThread: exception %s\n", ec.what());
@@ -358,6 +334,7 @@ void DSUControllerProvider::reader_thread()
 				}
 
 				index = info->GetIndex();
+				cemu_assert(index < kMaxClients);
 #ifdef DEBUG_DSU_CLIENT
 			printf(" DSUControllerProvider::ReaderThread: received PortInfo for index %d\n", index);
 #endif
@@ -381,6 +358,7 @@ void DSUControllerProvider::reader_thread()
 				}
 
 				index = rsp->GetIndex();
+				cemu_assert(index < kMaxClients);
 #ifdef DEBUG_DSU_CLIENT
 			printf(" DSUControllerProvider::ReaderThread: received DataResponse for index %d\n", index);
 #endif
@@ -406,20 +384,10 @@ void DSUControllerProvider::writer_thread()
 	SetThreadName("DSU-writer");
 	while (m_running.load(std::memory_order_relaxed))
 	{
-		std::unique_lock lock(m_writer_mutex);
-		while (m_writer_jobs.empty())
-		{
-			if (m_writer_cond.wait_for(lock, std::chrono::milliseconds(250)) == std::cv_status::timeout)
-			{
-				if (!m_running.load(std::memory_order_relaxed))
-					return;
-			}
-		}
-
-		const auto msg = std::move(m_writer_jobs.front());
-		m_writer_jobs.pop();
-		lock.unlock();
-
+		std::unique_ptr<ClientMessage> msg = m_writerJobs.pop();
+		if (!m_running.load(std::memory_order_relaxed))
+			return;
+		cemu_assert_debug(msg.get());
 #ifdef DEBUG_DSU_CLIENT
 		printf(" DSUControllerProvider::WriterThread: sending message: 0x%x (len: 0x%x)\n", (int)msg->GetMessageType(), msg->GetSize());
 #endif

src/input/api/DSU/DSUControllerProvider.h

@@ -4,6 +4,7 @@
 #include "input/api/DSU/DSUMessages.h"
 
 #include "input/api/ControllerProvider.h"
+#include "util/helpers/ConcurrentQueue.h"
 
 #include <boost/asio.hpp>
 
@@ -87,7 +88,6 @@ public:
 	void request_pad_data();
 	void request_pad_data(uint8_t index);
 
-
 private:
 	uint16 m_server_version = 0;
 
@@ -98,20 +98,18 @@ private:
 	void writer_thread();
 	void integrate_motion(uint8_t index, const DataResponse& data_response);
 
-	std::mutex m_writer_mutex;
-	std::condition_variable m_writer_cond;
-
 	uint32 m_uid;
 	boost::asio::io_context m_io_service;
 	boost::asio::ip::udp::endpoint m_receiver_endpoint;
 	boost::asio::ip::udp::socket m_socket;
+	boost::asio::ip::udp::endpoint m_socketWakeupEndpoint;
 
 	std::array<ControllerState, kMaxClients> m_state{};
 	std::array<ControllerState, kMaxClients> m_prev_state{};
 	mutable std::array<std::mutex, kMaxClients> m_mutex;
 	mutable std::array<std::condition_variable, kMaxClients> m_wait_cond;
 
-	std::queue<std::unique_ptr<ClientMessage>> m_writer_jobs;
+	ConcurrentQueue<std::unique_ptr<ClientMessage>> m_writerJobs;
 
 	std::array<WiiUMotionHandler, kMaxClients> m_motion_handler;
 	std::array<uint64, kMaxClients> m_last_motion_timestamp{};

src/util/helpers/ConcurrentQueue.h

@@ -62,7 +62,7 @@ public:
 			m_condVar.wait(mlock);
 		}
 
-		auto val = m_queue.front();
+		auto val = std::move(m_queue.front());
 		m_queue.pop();
 		return val;
 	}
@@ -75,7 +75,7 @@ public:
 			m_condVar.wait(mlock);
 		}
 
-		item = m_queue.front();
+		item = std::move(m_queue.front());
 		m_queue.pop();
 	}