Merge pull request #14110 from jordan-woyak/mgba-joybus-threading

HW/GBACore: Adjust joybus interthread communication to use WorkQueueThreadSP.

5 files changed, 89 insertions, 125 deletions

diff --git a/Source/Core/Common/WorkQueueThread.h b/Source/Core/Common/WorkQueueThread.h
index 8816ddda16..826ee778be 100644
--- a/Source/Core/Common/WorkQueueThread.h
+++ b/Source/Core/Common/WorkQueueThread.h
@@ -98,6 +98,8 @@ public:
       m_items.WaitForEmpty();
   }
 
+  bool IsRunning() { return m_thread.joinable(); }
+
 private:
   using CommandFunction = std::function<void()>;
 
@@ -142,8 +144,6 @@ private:
       return std::lock_guard{m_mutex};
   }
 
-  bool IsRunning() { return m_thread.joinable(); }
-
   void ThreadLoop(const std::string& thread_name, const FunctionType& function)
   {
     Common::SetCurrentThreadName(thread_name.c_str());
diff --git a/Source/Core/Core/HW/GBACore.cpp b/Source/Core/Core/HW/GBACore.cpp
index c0dc6de0aa..6d1d14b239 100644
--- a/Source/Core/Core/HW/GBACore.cpp
+++ b/Source/Core/Core/HW/GBACore.cpp
@@ -254,9 +254,8 @@ bool Core::Start(u64 gc_ticks)
 
   if (Config::Get(Config::MAIN_GBA_THREADS))
   {
-    m_idle = true;
-    m_exit_loop = false;
-    m_thread = std::make_unique<std::thread>([this] { ThreadLoop(); });
+    m_event_thread.Reset(fmt::format("GBA{}", m_device_number + 1),
+                         std::bind_front(&Core::HandleEvent, this));
   }
 
   return true;
@@ -264,17 +263,8 @@ bool Core::Start(u64 gc_ticks)
 
 void Core::Stop()
 {
-  if (m_thread)
-  {
-    Flush();
-    m_exit_loop = true;
-    {
-      std::lock_guard<std::mutex> lock(m_queue_mutex);
-      m_command_cv.notify_one();
-    }
-    m_thread->join();
-    m_thread.reset();
-  }
+  m_event_thread.Shutdown();
+
   if (m_core)
   {
     mCoreConfigDeinit(&m_core->config);
@@ -481,103 +471,76 @@ void Core::SetupEvent()
   m_event.priority = 0x80;
 }
 
+void Core::SyncJoybus(u64 gc_ticks, u16 keys)
+{
+  PushEvent({
+      .run_until_ticks = gc_ticks,
+      .keys = keys,
+      .event_type = JoybusEventType::TimeSync,
+  });
+}
+
 void Core::SendJoybusCommand(u64 gc_ticks, int transfer_time, u8* buffer, u16 keys)
 {
   if (!IsStarted())
     return;
 
-  Command command{};
-  command.ticks = gc_ticks;
-  command.transfer_time = transfer_time;
-  command.sync_only = buffer == nullptr;
-  if (buffer)
-    std::copy_n(buffer, command.buffer.size(), command.buffer.begin());
-  command.keys = keys;
+  m_joybus_command_transfer_time = transfer_time;
+  m_joybus_command = GBASIOJOYCommand(buffer[0]);
+  std::copy_n(buffer + 1, m_joybus_buffer.size(), m_joybus_buffer.data());
 
-  if (m_thread)
-  {
-    std::lock_guard<std::mutex> lock(m_queue_mutex);
-    m_command_queue.push(command);
-    m_idle = false;
-    m_command_cv.notify_one();
-  }
-  else
-  {
-    RunCommand(command);
-  }
+  m_command_pending.store(true, std::memory_order_relaxed);
+
+  PushEvent({
+      .run_until_ticks = gc_ticks,
+      .keys = keys,
+      .event_type = JoybusEventType::RunCommand,
+  });
 }
 
-std::vector<u8> Core::GetJoybusResponse()
+int Core::GetJoybusResponse(u8* data_out)
 {
-  if (!IsStarted())
-    return {};
+  m_command_pending.wait(true, std::memory_order_acquire);
 
-  if (m_thread)
-  {
-    std::unique_lock<std::mutex> lock(m_response_mutex);
-    m_response_cv.wait(lock, [&] { return m_response_ready; });
-  }
-  m_response_ready = false;
-  return m_response;
+  std::copy_n(m_joybus_buffer.data(), m_response_size, data_out);
+  return m_response_size;
 }
 
 void Core::Flush()
 {
-  if (!IsStarted() || !m_thread)
-    return;
-  std::unique_lock<std::mutex> lock(m_queue_mutex);
-  m_response_cv.wait(lock, [&] { return m_idle; });
+  m_event_thread.WaitForCompletion();
 }
 
-void Core::ThreadLoop()
+void Core::PushEvent(JoybusEvent event)
 {
-  Common::SetCurrentThreadName(fmt::format("GBA{}", m_device_number + 1).c_str());
-  std::unique_lock<std::mutex> queue_lock(m_queue_mutex);
-  while (true)
-  {
-    m_command_cv.wait(queue_lock, [&] { return !m_command_queue.empty() || m_exit_loop; });
-    if (m_exit_loop)
-      break;
-    Command command{m_command_queue.front()};
-    m_command_queue.pop();
-    queue_lock.unlock();
-
-    RunCommand(command);
-
-    queue_lock.lock();
-    if (m_command_queue.empty())
-      m_idle = true;
-    m_response_cv.notify_one();
-  }
+  if (m_event_thread.IsRunning())
+    m_event_thread.Push(event);
+  else
+    HandleEvent(event);
 }
 
-void Core::RunCommand(Command& command)
+void Core::HandleEvent(JoybusEvent event)
 {
-  m_keys = command.keys;
-  RunUntil(command.ticks);
-  if (!command.sync_only)
-  {
-    m_response.clear();
-    if (m_link_enabled && !m_force_disconnect)
-    {
-      int recvd = GBASIOJOYSendCommand(
-          &m_sio_driver, static_cast<GBASIOJOYCommand>(command.buffer[0]), &command.buffer[1]);
-      std::copy_n(command.buffer.begin() + 1, recvd, std::back_inserter(m_response));
-    }
+  m_keys = event.keys;
+  RunUntil(event.run_until_ticks);
 
-    if (m_thread && !m_response_ready)
-    {
-      std::lock_guard<std::mutex> response_lock(m_response_mutex);
-      m_response_ready = true;
-      m_response_cv.notify_one();
-    }
-    else
-    {
-      m_response_ready = true;
-    }
+  if (event.event_type != JoybusEventType::RunCommand)
+    return;
+
+  if (m_link_enabled && !m_force_disconnect)
+  {
+    m_response_size =
+        u8(GBASIOJOYSendCommand(&m_sio_driver, m_joybus_command, m_joybus_buffer.data()));
   }
-  if (command.transfer_time)
-    RunFor(command.transfer_time);
+  else
+  {
+    m_response_size = 0;
+  }
+
+  m_command_pending.store(false, std::memory_order_release);
+  m_command_pending.notify_one();
+
+  RunFor(m_joybus_command_transfer_time);
 }
 
 void Core::RunUntil(u64 gc_ticks)
@@ -700,8 +663,8 @@ void Core::DoState(PointerWrap& p)
   p.Do(m_gc_ticks_remainder);
   p.Do(m_keys);
   p.Do(m_link_enabled);
-  p.Do(m_response_ready);
-  p.Do(m_response);
+  p.Do(m_response_size);
+  p.Do(m_joybus_buffer);
 
   std::vector<u8> core_state;
   core_state.resize(m_core->stateSize(m_core));
diff --git a/Source/Core/Core/HW/GBACore.h b/Source/Core/Core/HW/GBACore.h
index 72faa1bec5..64f61a0eb7 100644
--- a/Source/Core/Core/HW/GBACore.h
+++ b/Source/Core/Core/HW/GBACore.h
@@ -6,13 +6,9 @@
 #ifdef HAS_LIBMGBA
 
 #include <array>
-#include <condition_variable>
 #include <memory>
-#include <mutex>
-#include <queue>
 #include <string>
 #include <string_view>
-#include <thread>
 #include <vector>
 
 #define PYCPARSE  // Remove static functions from the header
@@ -23,6 +19,7 @@
 
 #include "Common/Buffer.h"
 #include "Common/CommonTypes.h"
+#include "Common/WorkQueueThread.h"
 
 class GBAHostInterface;
 class PointerWrap;
@@ -75,8 +72,9 @@ public:
   void SetForceDisconnect(bool force_disconnect);
   void EReaderQueueCard(std::string_view card_path);
 
+  void SyncJoybus(u64 gc_ticks, u16 keys);
   void SendJoybusCommand(u64 gc_ticks, int transfer_time, u8* buffer, u16 keys);
-  std::vector<u8> GetJoybusResponse();
+  int GetJoybusResponse(u8* data_out);
 
   void ImportState(std::string_view state_path);
   void ExportState(std::string_view state_path);
@@ -88,20 +86,23 @@ public:
   static std::string GetSavePath(std::string_view rom_path, int device_number);
 
 private:
-  void ThreadLoop();
   void RunUntil(u64 gc_ticks);
   void RunFor(u64 gc_ticks);
   void Flush();
 
-  struct Command
+  enum class JoybusEventType : u8
   {
-    u64 ticks;
-    int transfer_time;
-    bool sync_only;
-    std::array<u8, 6> buffer;
-    u16 keys;
+    TimeSync,
+    RunCommand,
   };
-  void RunCommand(Command& command);
+  struct JoybusEvent
+  {
+    u64 run_until_ticks{};
+    u16 keys{};
+    JoybusEventType event_type{};
+  };
+  void PushEvent(JoybusEvent event);
+  void HandleEvent(JoybusEvent event);
 
   bool LoadBIOS(const char* bios_path);
   bool LoadSave(const char* save_path);
@@ -136,17 +137,19 @@ private:
 
   std::weak_ptr<GBAHostInterface> m_host;
 
-  std::unique_ptr<std::thread> m_thread;
-  bool m_exit_loop = false;
-  bool m_idle = false;
-  std::mutex m_queue_mutex;
-  std::condition_variable m_command_cv;
-  std::queue<Command> m_command_queue;
-
-  std::mutex m_response_mutex;
-  std::condition_variable m_response_cv;
-  bool m_response_ready = false;
-  std::vector<u8> m_response;
+  // Set by the GC thread before issuing a JoybusEventType::RunCommand.
+  int m_joybus_command_transfer_time{};
+  GBASIOJOYCommand m_joybus_command{};
+
+  // Commands are synchronous. This buffer is used for the command and the response.
+  std::array<u8, 5> m_joybus_buffer{};  // State saved.
+
+  // Set by the GBA thread after filling in the above buffer.
+  u8 m_response_size{};  // State saved.
+  std::atomic_bool m_command_pending{};
+
+  // The entire threaded GBA runs within events pushed to this queue.
+  Common::WorkQueueThreadSP<JoybusEvent> m_event_thread;
 
   ::Core::System& m_system;
 };
diff --git a/Source/Core/Core/HW/SI/SI_DeviceGBAEmu.cpp b/Source/Core/Core/HW/SI/SI_DeviceGBAEmu.cpp
index eb7c06343d..36ccc15bc0 100644
--- a/Source/Core/Core/HW/SI/SI_DeviceGBAEmu.cpp
+++ b/Source/Core/Core/HW/SI/SI_DeviceGBAEmu.cpp
@@ -91,11 +91,9 @@ int CSIDevice_GBAEmu::RunBuffer(u8* buffer, int request_length)
   case NextAction::ReceiveResponse:
   {
     m_next_action = NextAction::SendCommand;
-
-    std::vector<u8> response = m_core->GetJoybusResponse();
-    if (response.empty())
+    const auto response_length = m_core->GetJoybusResponse(buffer);
+    if (response_length == 0)
       return -1;
-    std::ranges::copy(response, buffer);
 
 #ifdef _DEBUG
     const Common::Log::LogLevel log_level =
@@ -105,10 +103,10 @@ int CSIDevice_GBAEmu::RunBuffer(u8* buffer, int request_length)
     GENERIC_LOG_FMT(Common::Log::LogType::SERIALINTERFACE, log_level,
                     "{}                              [< {:02x}{:02x}{:02x}{:02x}{:02x}] ({})",
                     m_device_number, buffer[0], buffer[1], buffer[2], buffer[3], buffer[4],
-                    response.size());
+                    response_length);
 #endif
 
-    return static_cast<int>(response.size());
+    return response_length;
   }
   }
 
@@ -169,7 +167,7 @@ void CSIDevice_GBAEmu::DoState(PointerWrap& p)
 
 void CSIDevice_GBAEmu::OnEvent(u64 userdata, s64 cycles_late)
 {
-  m_core->SendJoybusCommand(m_system.GetCoreTiming().GetTicks() + userdata, 0, nullptr, m_keys);
+  m_core->SyncJoybus(m_system.GetCoreTiming().GetTicks() + userdata, m_keys);
 
   const auto num_cycles = userdata + GetSyncInterval(m_system.GetSystemTimers());
   m_system.GetSerialInterface().ScheduleEvent(m_device_number, num_cycles);
diff --git a/Source/Core/Core/State.cpp b/Source/Core/Core/State.cpp
index 3819a2c0fb..b9f655e578 100644
--- a/Source/Core/Core/State.cpp
+++ b/Source/Core/Core/State.cpp
@@ -95,7 +95,7 @@ struct CompressAndDumpStateArgs
 static Common::WorkQueueThreadSP<CompressAndDumpStateArgs> s_compress_and_dump_thread;
 
 // Don't forget to increase this after doing changes on the savestate system
-constexpr u32 STATE_VERSION = 183;  // Last changed in PR 14501
+constexpr u32 STATE_VERSION = 184;  // Last changed in PR 14110
 
 // Increase this if the StateExtendedHeader definition changes
 constexpr u32 EXTENDED_HEADER_VERSION = 1;  // Last changed in PR 12217