From 3c7ccf809ea2e2314dfee35af9d92565f839f175 Mon Sep 17 00:00:00 2001 From: Exzap <13877693+Exzap@users.noreply.github.com> Date: Fri, 8 May 2026 06:04:13 +0200 Subject: [PATCH] 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 | 96 +++++++-------------- src/input/api/DSU/DSUControllerProvider.h | 8 +- src/util/helpers/ConcurrentQueue.h | 4 +- 3 files changed, 37 insertions(+), 71 deletions(-) diff --git a/src/input/api/DSU/DSUControllerProvider.cpp b/src/input/api/DSU/DSUControllerProvider.cpp index 885f3284..a94aaae0 100644 --- a/src/input/api/DSU/DSUControllerProvider.cpp +++ b/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 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{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(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(m_uid, 4, std::array{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(m_uid, 1, std::array{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(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(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 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 @@ -479,4 +447,4 @@ DSUControllerProvider::ControllerState& DSUControllerProvider::ControllerState:: this->operator=(static_cast(data_response)); data = data_response.GetData(); return *this; -} +} \ No newline at end of file diff --git a/src/input/api/DSU/DSUControllerProvider.h b/src/input/api/DSU/DSUControllerProvider.h index 692da619..a0a933da 100644 --- a/src/input/api/DSU/DSUControllerProvider.h +++ b/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 @@ -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 m_state{}; std::array m_prev_state{}; mutable std::array m_mutex; mutable std::array m_wait_cond; - std::queue> m_writer_jobs; + ConcurrentQueue> m_writerJobs; std::array m_motion_handler; std::array m_last_motion_timestamp{}; diff --git a/src/util/helpers/ConcurrentQueue.h b/src/util/helpers/ConcurrentQueue.h index 2e7347b7..7cd370ee 100644 --- a/src/util/helpers/ConcurrentQueue.h +++ b/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(); }