Replace SPSCQueue with lock-free libretro-common queues

Every Common::SPSCQueue goes, and with it the mutexes in front of the
multi-producer ones:

- WorkQueueThread runs on an sthread fed by an mpsc_stack, so any
  thread pushes without a lock, and parks on retro_eventcount.
- AsyncRequests hands requests to the video thread through a
  retro_waitable_spsc of pointers; a blocking request lives on the
  caller's stack.
- CoreTiming's off-thread events and the DVD thread's results move to
  mpsc_stack, dropping the CoreTiming write lock.
- Real Wiimote reports cross threads as fixed 32-byte records in
  retro_spsc rings; the read thread receives straight into the ring.
- Bluetooth passthrough frees transfers through an mpsc_stack whose
  nodes live in the transfers' own buffers, queues incoming packets in
  mirrored retro_spsc rings and hands ACL data to IOS in place.
- PerformanceTracker's frame times move to a retro_spsc ring.

Source/Core/Common/SPSCQueue.h

@@ -1,155 +0,0 @@
-// Copyright 2010 Dolphin Emulator Project
-// SPDX-License-Identifier: GPL-2.0-or-later
-
-#pragma once
-
-// a simple lockless thread-safe,
-// single producer, single consumer queue
-
-#include <atomic>
-#include <cassert>
-
-#if defined(__APPLE__)
-#include <Availability.h>
-#if (defined(__TV_OS_VERSION_MIN_REQUIRED) && __TV_OS_VERSION_MIN_REQUIRED < 140000) || \
-    (defined(__IPHONE_OS_VERSION_MIN_REQUIRED) && __IPHONE_OS_VERSION_MIN_REQUIRED < 140000)
-#define DOLPHIN_NEEDS_ATOMIC_WAIT_FALLBACK 1
-#include <condition_variable>
-#include <mutex>
-#endif
-#endif
-
-#include "Common/TypeUtils.h"
-
-namespace Common
-{
-
-namespace detail
-{
-template <typename T, bool IncludeWaitFunctionality>
-class SPSCQueueBase final
-{
-public:
-  SPSCQueueBase() = default;
-  ~SPSCQueueBase()
-  {
-    Clear();
-    delete m_read_ptr;
-  }
-
-  SPSCQueueBase(const SPSCQueueBase&) = delete;
-  SPSCQueueBase& operator=(const SPSCQueueBase&) = delete;
-
-  std::size_t Size() const { return m_size.load(std::memory_order_acquire); }
-  bool Empty() const { return Size() == 0; }
-
-  // The following are only safe from the "producer thread":
-  void Push(const T& arg) { Emplace(arg); }
-  void Push(T&& arg) { Emplace(std::move(arg)); }
-  template <typename... Args>
-  void Emplace(Args&&... args)
-  {
-    m_write_ptr->value.Construct(std::forward<Args>(args)...);
-
-    Node* const new_ptr = new Node;
-    m_write_ptr->next = new_ptr;
-    m_write_ptr = new_ptr;
-
-    AdjustSize(1);
-  }
-
-  void WaitForEmpty() requires(IncludeWaitFunctionality)
-  {
-#ifdef DOLPHIN_NEEDS_ATOMIC_WAIT_FALLBACK
-    std::unique_lock<std::mutex> lock(m_wait_mutex);
-    while (Size() > 0)
-      m_wait_cv.wait(lock);
-#else
-    while (const std::size_t old_size = Size())
-      m_size.wait(old_size, std::memory_order_acquire);
-#endif
-  }
-
-  // The following are only safe from the "consumer thread":
-  T& Front() { return m_read_ptr->value.Ref(); }
-  const T& Front() const { return m_read_ptr->value.Ref(); }
-
-  void Pop()
-  {
-    assert(!Empty());
-
-    m_read_ptr->value.Destroy();
-
-    Node* const old_node = m_read_ptr;
-    m_read_ptr = old_node->next;
-    delete old_node;
-
-    AdjustSize(-1);
-  }
-
-  bool Pop(T& result)
-  {
-    if (Empty())
-      return false;
-
-    result = std::move(Front());
-    Pop();
-    return true;
-  }
-
-  void WaitForData() requires(IncludeWaitFunctionality)
-  {
-#ifdef DOLPHIN_NEEDS_ATOMIC_WAIT_FALLBACK
-    std::unique_lock<std::mutex> lock(m_wait_mutex);
-    while (Size() == 0)
-      m_wait_cv.wait(lock);
-#else
-    m_size.wait(0, std::memory_order_acquire);
-#endif
-  }
-
-  void Clear()
-  {
-    while (!Empty())
-      Pop();
-  }
-
-private:
-  struct Node
-  {
-    ManuallyConstructedValue<T> value;
-    Node* next;
-  };
-
-  Node* m_write_ptr = new Node;
-  Node* m_read_ptr = m_write_ptr;
-
-  void AdjustSize(std::size_t value)
-  {
-    m_size.fetch_add(value, std::memory_order_release);
-    if constexpr (IncludeWaitFunctionality)
-    {
-#ifdef DOLPHIN_NEEDS_ATOMIC_WAIT_FALLBACK
-      m_wait_cv.notify_one();
-#else
-      m_size.notify_one();
-#endif
-    }
-  }
-
-  std::atomic<std::size_t> m_size = 0;
-
-#ifdef DOLPHIN_NEEDS_ATOMIC_WAIT_FALLBACK
-  std::mutex m_wait_mutex;
-  std::condition_variable m_wait_cv;
-#endif
-};
-}  // namespace detail
-
-template <typename T>
-using SPSCQueue = detail::SPSCQueueBase<T, false>;
-
-template <typename T>
-using WaitableSPSCQueue = detail::SPSCQueueBase<T, true>;
-
-}  // namespace Common

Source/Core/Common/WorkQueueThread.h

@@ -4,49 +4,73 @@
 #pragma once
 
 #include <functional>
-#include <future>
-#include <mutex>
 #include <string>
-#include <thread>
+#include <utility>
 
-#include "Common/Event.h"
-#include "Common/SPSCQueue.h"
-#include "Common/Thread.h"
+#include <queues/mpsc_stack.h>
+#include <retro_atomic.h>
+#include <rthreads/retro_eventcount.h>
+#include <rthreads/rthreads.h>
 
 namespace Common
 {
 namespace detail
 {
+// A worker thread fed by a lock-free queue. Producers push onto an mpsc_stack, so any number of
+// threads may push without a lock; the worker takes the whole stack at once and runs it in push
+// order. Items are moved into their queue node on push and moved out of it once, by the worker.
+//
+// The control calls (Reset, Cancel, Shutdown, Stop, WaitForCompletion) are made from one thread.
 template <typename T, bool IsSingleProducer>
 class WorkQueueThreadBase final
 {
 public:
   using FunctionType = std::function<void(T)>;
 
-  WorkQueueThreadBase() = default;
-  WorkQueueThreadBase(std::string name, FunctionType function)
+  WorkQueueThreadBase()
+  {
+    mpsc_stack_init(&m_items);
+    retro_atomic_size_init(&m_pushed, 0);
+    retro_atomic_size_init(&m_done, 0);
+    retro_atomic_size_init(&m_cancel_to, 0);
+    retro_atomic_int_init(&m_stop, 0);
+    retro_eventcount_init(&m_wake);
+    retro_eventcount_init(&m_progress);
+  }
+  WorkQueueThreadBase(std::string name, FunctionType function) : WorkQueueThreadBase()
   {
     Reset(std::move(name), std::move(function));
   }
-  ~WorkQueueThreadBase() { Shutdown(); }
-
-  // Shuts the current work thread down (if any) and starts a new thread with the given function
-  // Note: Some consumers of this API push items to the queue before starting the thread.
-  void Reset(std::string name, FunctionType function)
+  ~WorkQueueThreadBase()
   {
-    auto lg = GetLockGuard();
     Shutdown();
-    m_thread = std::thread(std::bind_front(&WorkQueueThreadBase::ThreadLoop, this), std::move(name),
-                           std::move(function));
+    FreeChain(m_held);
+    FreeChain(reinterpret_cast<Node*>(mpsc_stack_reverse(mpsc_stack_drain(&m_items))));
+    retro_eventcount_free(&m_progress);
+    retro_eventcount_free(&m_wake);
   }
 
-  // Adds an item to the work queue
+  WorkQueueThreadBase(const WorkQueueThreadBase&) = delete;
+  WorkQueueThreadBase& operator=(const WorkQueueThreadBase&) = delete;
+
+  // Shuts the current work thread down (if any) and starts a new thread with the given function.
+  // Items pushed before the thread starts are kept and run by it.
+  void Reset(std::string name, FunctionType function)
+  {
+    Shutdown();
+    m_name = std::move(name);
+    m_function = std::move(function);
+    m_thread = sthread_create(ThreadEntry, this);
+  }
+
+  // Adds an item to the work queue. Never blocks.
   template <typename... Args>
   void EmplaceItem(Args&&... args)
   {
-    auto lg = GetLockGuard();
-    m_items.Emplace(std::forward<Args>(args)...);
-    m_event.Set();
+    Node* const node = new Node{{nullptr}, T(std::forward<Args>(args)...)};
+    mpsc_stack_push(&m_items, &node->link);
+    retro_atomic_fetch_add_size(&m_pushed, 1);
+    retro_eventcount_notify(&m_wake);
   }
   void Push(T&& item) { EmplaceItem(std::move(item)); }
   void Push(const T& item) { EmplaceItem(item); }
@@ -55,36 +79,46 @@ public:
   // Blocks until the current work is cancelled.
   void Cancel()
   {
-    auto lg = GetLockGuard();
-
-    // Fast path avoids round trip thread communication and saves ~20us.
-    if (m_items.Empty())
+    const std::size_t pushed = retro_atomic_load_acquire_size(&m_pushed);
+    if (retro_atomic_load_acquire_size(&m_done) == pushed)
       return;
 
-    RunCommand([&] { m_items.Clear(); });
+    if (!IsRunning())
+    {
+      // No worker: the queue is ours.
+      Node* chain = TakeAll();
+      std::size_t skipped = 0;
+      while (chain)
+      {
+        Node* const next = reinterpret_cast<Node*>(chain->link.next);
+        delete chain;
+        chain = next;
+        ++skipped;
+      }
+      retro_atomic_fetch_add_size(&m_done, skipped);
+      return;
+    }
+
+    retro_atomic_store_release_size(&m_cancel_to, pushed);
+    retro_eventcount_notify(&m_wake);
+    WaitForDone(pushed);
   }
 
   // Tells the worker thread to stop when its queue is empty.
   // Blocks until the worker thread exits. Does nothing if thread isn't running.
   void Shutdown()
   {
-    auto lg = GetLockGuard();
     WaitForCompletion();
     StopThread();
   }
 
   // Tells the worker thread to stop immediately, potentially leaving work in the queue.
   // Blocks until the worker thread exits. Does nothing if thread isn't running.
-  void Stop()
-  {
-    auto lg = GetLockGuard();
-    StopThread();
-  }
+  void Stop() { StopThread(); }
 
   // Stops the worker thread ASAP and empties the queue.
   void StopAndCancel()
   {
-    auto lg = GetLockGuard();
     Stop();
     Cancel();
   }
@@ -93,93 +127,126 @@ public:
   // Does nothing if thread isn't running.
   void WaitForCompletion()
   {
-    auto lg = GetLockGuard();
     if (IsRunning())
-      m_items.WaitForEmpty();
+      WaitForDone(retro_atomic_load_acquire_size(&m_pushed));
   }
 
-  bool IsRunning() { return m_thread.joinable(); }
+  bool IsRunning() const { return m_thread != nullptr; }
 
 private:
-  using CommandFunction = std::function<void()>;
-
-  // Blocking.
-  void RunCommand(CommandFunction cmd)
+  struct Node
   {
-    if (!IsRunning())
-    {
-      std::invoke(cmd);
-      return;
-    }
+    mpsc_stack_node_t link;
+    T value;
+  };
 
-    m_commands.Emplace(std::move(cmd));
-    m_event.Set();
-    m_commands.WaitForEmpty();
+  static void FreeChain(Node* chain)
+  {
+    while (chain)
+    {
+      Node* const next = reinterpret_cast<Node*>(chain->link.next);
+      delete chain;
+      chain = next;
+    }
+  }
+
+  // Everything queued, oldest first: what a stopped worker held back, then the stack.
+  Node* TakeAll()
+  {
+    Node* chain = reinterpret_cast<Node*>(mpsc_stack_reverse(mpsc_stack_drain(&m_items)));
+    if (!m_held)
+      return chain;
+    Node* tail = m_held;
+    while (tail->link.next)
+      tail = reinterpret_cast<Node*>(tail->link.next);
+    tail->link.next = chain ? &chain->link : nullptr;
+    chain = m_held;
+    m_held = nullptr;
+    return chain;
+  }
+
+  void WaitForDone(std::size_t target)
+  {
+    while (retro_atomic_load_acquire_size(&m_done) < target)
+    {
+      const int key = retro_eventcount_prepare_wait(&m_progress);
+      if (retro_atomic_load_acquire_size(&m_done) >= target)
+      {
+        retro_eventcount_cancel_wait(&m_progress);
+        break;
+      }
+      retro_eventcount_commit_wait(&m_progress, key);
+    }
   }
 
-  // Stop immediately.
   void StopThread()
   {
-    if (!m_thread.joinable())
+    if (!m_thread)
       return;
 
-    // empty-function shutdown signal.
-    m_commands.Emplace(CommandFunction{});
-    m_event.Set();
-    m_thread.join();
-    m_commands.Clear();
+    retro_atomic_store_release_int(&m_stop, 1);
+    retro_eventcount_notify(&m_wake);
+    sthread_join(m_thread);
+    m_thread = nullptr;
+    retro_atomic_store_release_int(&m_stop, 0);
   }
 
-  auto GetLockGuard()
+  static void ThreadEntry(void* data) { static_cast<WorkQueueThreadBase*>(data)->ThreadLoop(); }
+
+  void ThreadLoop()
   {
-    struct DummyLockGuard
+    sthread_setname(m_name.c_str());
+
+    // Items are numbered in the order they run; m_done is the last one finished.
+    std::size_t seq = retro_atomic_load_relaxed_size(&m_done);
+    Node* chain = m_held;
+    m_held = nullptr;
+
+    for (;;)
     {
-      // Silences unused variable warning.
-      ~DummyLockGuard() { void(); }
-    };
+      if (!chain)
+        chain = reinterpret_cast<Node*>(mpsc_stack_reverse(mpsc_stack_drain(&m_items)));
 
-    if constexpr (IsSingleProducer)
-      return DummyLockGuard{};
-    else
-      return std::lock_guard{m_mutex};
-  }
+      if (retro_atomic_load_acquire_int(&m_stop))
+        break;
 
-  void ThreadLoop(const std::string& thread_name, const FunctionType& function)
-  {
-    Common::SetCurrentThreadName(thread_name.c_str());
-
-    while (true)
-    {
-      while (!m_commands.Empty())
+      if (chain)
       {
-        CommandFunction& command = m_commands.Front();
-        // empty-function shutdown signal.
-        if (!command)
-          return;
-
-        std::invoke(command);
-        m_commands.Pop();
-      }
-
-      if (m_items.Empty())
-      {
-        m_event.Wait();
+        Node* const node = chain;
+        chain = reinterpret_cast<Node*>(node->link.next);
+        ++seq;
+        if (seq > retro_atomic_load_acquire_size(&m_cancel_to))
+          m_function(std::move(node->value));
+        delete node;
+        retro_atomic_store_release_size(&m_done, seq);
+        retro_eventcount_notify(&m_progress);
         continue;
       }
 
-      function(std::move(m_items.Front()));
-      m_items.Pop();
+      const int key = retro_eventcount_prepare_wait(&m_wake);
+      if (!mpsc_stack_empty(&m_items) || retro_atomic_load_acquire_int(&m_stop))
+      {
+        retro_eventcount_cancel_wait(&m_wake);
+        continue;
+      }
+      retro_eventcount_commit_wait(&m_wake, key);
     }
+
+    // Stopped with work left: keep it, in order, for the next thread.
+    m_held = chain;
   }
 
-  std::thread m_thread;
-  Common::WaitableSPSCQueue<T> m_items;
-  Common::WaitableSPSCQueue<CommandFunction> m_commands;
-  Common::Event m_event;
-
-  using DummyMutex = std::type_identity<void>;
-  using ProducerMutex = std::conditional_t<IsSingleProducer, DummyMutex, std::recursive_mutex>;
-  ProducerMutex m_mutex;
+  sthread_t* m_thread = nullptr;
+  Node* m_held = nullptr;  // worker-owned while running, ours while stopped
+  FunctionType m_function;
+  std::string m_name;
+  mpsc_stack_t m_items;
+  retro_eventcount_t m_wake;
+  retro_eventcount_t m_progress;
+  retro_atomic_size_t m_pushed;
+  retro_atomic_size_t m_done;
+  retro_atomic_size_t m_cancel_to;
+  retro_atomic_int_t m_stop;
 };
 
 // A WorkQueueThread-like class that takes functions to invoke.
@@ -189,8 +256,16 @@ class AsyncWorkThreadBase
 public:
   using FuncType = std::function<void()>;
 
-  AsyncWorkThreadBase() = default;
-  explicit AsyncWorkThreadBase(std::string thread_name) { Reset(std::move(thread_name)); }
+  AsyncWorkThreadBase() { retro_eventcount_init(&m_ran); }
+  explicit AsyncWorkThreadBase(std::string thread_name) : AsyncWorkThreadBase()
+  {
+    Reset(std::move(thread_name));
+  }
+  ~AsyncWorkThreadBase()
+  {
+    m_worker.Shutdown();
+    retro_eventcount_free(&m_ran);
+  }
 
   void Reset(std::string thread_name)
   {
@@ -199,11 +274,28 @@ public:
 
   void Push(FuncType func) { m_worker.Push(std::move(func)); }
 
-  auto PushBlocking(FuncType func)
+  // Runs func on the worker and returns once it has run.
+  void PushBlocking(FuncType func)
   {
-    std::packaged_task task{std::move(func)};
-    m_worker.EmplaceItem([&] { task(); });
-    return task.get_future().get();
+    // The flag lives on this stack; the eventcount outlives the call, since the worker may
+    // still be inside the notify when this returns.
+    retro_atomic_int_t done;
+    retro_atomic_int_init(&done, 0);
+    m_worker.EmplaceItem([&func, &done, this] {
+      func();
+      retro_atomic_store_release_int(&done, 1);
+      retro_eventcount_notify(&m_ran);
+    });
+    while (!retro_atomic_load_acquire_int(&done))
+    {
+      const int key = retro_eventcount_prepare_wait(&m_ran);
+      if (retro_atomic_load_acquire_int(&done))
+      {
+        retro_eventcount_cancel_wait(&m_ran);
+        break;
+      }
+      retro_eventcount_commit_wait(&m_ran, key);
+    }
   }
 
   void Cancel() { m_worker.Cancel(); }
@@ -212,15 +304,15 @@ public:
 
 private:
   WorkThread<FuncType> m_worker;
+  retro_eventcount_t m_ran;
 };
 }  // namespace detail
 
-// Multiple threads may use the public interface.
+// Multiple threads may push.
 template <typename T>
 using WorkQueueThread = detail::WorkQueueThreadBase<T, false>;
 
-// A "Single Producer" WorkQueueThread.
-// It uses no mutex but only one thread can safely manipulate the queue.
+// A "Single Producer" WorkQueueThread: only one thread pushes.
 template <typename T>
 using WorkQueueThreadSP = detail::WorkQueueThreadBase<T, true>;
 

Source/Core/Core/Core.cpp

@@ -967,21 +967,9 @@ void Callback_NewField(Core::System& system)
 {
   if (s_frame_step)
   {
-#ifndef __LIBRETRO__
-    // To ensure that s_stop_frame_step is up to date, wait for the GPU thread queue to empty,
-    // since it is may contain a swap event (which will call Callback_FramePresented). This hurts
-    // the performance a little, but luckily, performance matters less when using frame stepping.
-    AsyncRequests::GetInstance()->WaitForEmptyQueue();
-
-    // Only stop the frame stepping if a new frame was displayed
-    // (as opposed to the previous frame being displayed for another frame).
-    if (s_stop_frame_step.load())
-#endif
-    {
-      s_frame_step = false;
-      system.GetCPU().Break();
-      NotifyStateChanged(Core::GetState(system));
-    }
+    s_frame_step = false;
+    system.GetCPU().Break();
+    NotifyStateChanged(Core::GetState(system));
   }
 
   AchievementManager::GetInstance().DoFrame();

Source/Core/Core/CoreTiming.cpp

@@ -4,7 +4,6 @@
 #include "Core/CoreTiming.h"
 
 #include <algorithm>
-#include <mutex>
 #include <string>
 #include <unordered_map>
 #include <vector>
@@ -14,7 +13,6 @@
 #include "Common/Assert.h"
 #include "Common/ChunkFile.h"
 #include "Common/Logging/Log.h"
-#include "Common/SPSCQueue.h"
 #include "Common/ScopeGuard.h"
 
 #include "Core/AchievementManager.h"
@@ -42,6 +40,7 @@ static void EmptyTimedCallback(Core::System& system, u64 userdata, s64 cyclesLat
 
 CoreTimingManager::CoreTimingManager(Core::System& system) : m_system(system)
 {
+  mpsc_stack_init(&m_ts_queue);
 }
 
 // Changing the CPU speed in Dolphin isn't actually done by changing the physical clock rate,
@@ -122,11 +121,21 @@ void CoreTimingManager::Init()
   });
 }
 
+CoreTimingManager::~CoreTimingManager()
+{
+  mpsc_stack_node_t* node = mpsc_stack_drain(&m_ts_queue);
+  while (node)
+  {
+    mpsc_stack_node_t* const next = node->next;
+    delete reinterpret_cast<ThreadSafeEvent*>(node);
+    node = next;
+  }
+}
+
 void CoreTimingManager::Shutdown()
 {
   m_core_state_changed_hook.reset();
 
-  std::lock_guard lk(m_ts_write_lock);
   MoveEvents();
   ClearPendingEvents();
   UnregisterAllEvents();
@@ -171,7 +180,6 @@ void CoreTimingManager::RefreshConfig()
 
 void CoreTimingManager::DoState(PointerWrap& p)
 {
-  std::lock_guard lk(m_ts_write_lock);
   p.Do(m_globals.slice_length);
   p.Do(m_globals.global_timer);
   p.Do(m_idled_cycles);
@@ -294,8 +302,9 @@ void CoreTimingManager::ScheduleEvent(s64 cycles_into_future, EventType* event_t
                     *event_type->name);
     }
 
-    std::lock_guard lk(m_ts_write_lock);
-    m_ts_queue.Push(Event{cycles_into_future, 0, userdata, event_type});
+    ThreadSafeEvent* const node =
+        new ThreadSafeEvent{{nullptr}, Event{cycles_into_future, 0, userdata, event_type}};
+    mpsc_stack_push(&m_ts_queue, &node->link);
   }
 }
 
@@ -332,10 +341,17 @@ void CoreTimingManager::ForceExceptionCheck(s64 cycles)
 
 void CoreTimingManager::MoveEvents()
 {
-  while (!m_ts_queue.Empty())
+  if (mpsc_stack_empty(&m_ts_queue))
+    return;
+
+  // Oldest first, so events keep the order they were scheduled in.
+  mpsc_stack_node_t* node = mpsc_stack_reverse(mpsc_stack_drain(&m_ts_queue));
+  while (node)
   {
-    auto& ev = m_event_queue.emplace_back(m_ts_queue.Front());
-    m_ts_queue.Pop();
+    ThreadSafeEvent* const ts_event = reinterpret_cast<ThreadSafeEvent*>(node);
+    node = node->next;
+    auto& ev = m_event_queue.emplace_back(ts_event->event);
+    delete ts_event;
 
     ev.fifo_order = m_event_fifo_id++;
     ev.time += m_globals.global_timer;

Source/Core/Core/CoreTiming.h

@@ -16,7 +16,6 @@
 // inside callback:
 //   ScheduleEvent(periodInCycles - cyclesLate, callback, "whatever")
 
-#include <mutex>
 #include <string>
 #include <tuple>
 #include <unordered_map>
@@ -25,10 +24,11 @@
 #include "Common/CommonTypes.h"
 #include "Common/Functional.h"
 #include "Common/HookableEvent.h"
-#include "Common/SPSCQueue.h"
 #include "Common/Timer.h"
 #include "Core/CPUThreadConfigCallback.h"
 
+#include <queues/mpsc_stack.h>
+
 class PointerWrap;
 
 namespace Core
@@ -92,6 +92,7 @@ class CoreTimingManager
 {
 public:
   explicit CoreTimingManager(Core::System& system);
+  ~CoreTimingManager();
 
   // CoreTiming begins at the boundary of timing slice -1. An initial call to Advance() is
   // required to end slice -1 and start slice 0 before the first cycle of code is executed.
@@ -188,11 +189,15 @@ private:
   // by the standard adaptor class.
   std::vector<Event> m_event_queue;
   u64 m_event_fifo_id = 0;
-  std::mutex m_ts_write_lock;
 
-  // Event objects created from other threads.
-  // The time value of each Event here is a cycles_into_future value.
-  Common::SPSCQueue<Event> m_ts_queue;
+  // Event objects created from other threads, pushed lock-free by any of them and taken all at
+  // once by the CPU thread. The time value of each Event here is a cycles_into_future value.
+  struct ThreadSafeEvent
+  {
+    mpsc_stack_node_t link;
+    Event event;
+  };
+  mpsc_stack_t m_ts_queue;
 
   float m_last_oc_factor = 0.0f;
 

Source/Core/Core/HW/DVD/AMMediaboard.cpp

@@ -3,6 +3,8 @@
 
 #include "Core/HW/DVD/AMMediaboard.h"
 
+#include <cassert>
+
 #include <algorithm>
 #include <bit>
 #include <random>

Source/Core/Core/HW/DVD/DVDThread.cpp

@@ -13,7 +13,6 @@
 #include "Common/CommonTypes.h"
 #include "Common/Logging/Log.h"
 #include "Common/MsgHandler.h"
-#include "Common/SPSCQueue.h"
 #include "Common/Timer.h"
 
 #include "Core/ConfigManager.h"
@@ -33,9 +32,60 @@ namespace DVD
 {
 DVDThread::DVDThread(Core::System& system) : m_system(system)
 {
+  mpsc_stack_init(&m_result_queue);
+  retro_eventcount_init(&m_result_ready);
 }
 
-DVDThread::~DVDThread() = default;
+DVDThread::~DVDThread()
+{
+  m_dvd_thread.Shutdown();
+  FreeResults();
+  retro_eventcount_free(&m_result_ready);
+}
+
+void DVDThread::FreeResults()
+{
+  mpsc_stack_node_t* node = m_result_chain;
+  m_result_chain = nullptr;
+  while (node)
+  {
+    mpsc_stack_node_t* const next = node->next;
+    delete reinterpret_cast<ResultNode*>(node);
+    node = next;
+  }
+  mpsc_stack_node_t* queued = mpsc_stack_drain(&m_result_queue);
+  while (queued)
+  {
+    mpsc_stack_node_t* const next = queued->next;
+    delete reinterpret_cast<ResultNode*>(queued);
+    queued = next;
+  }
+}
+
+// CPU thread. Unlinks and returns the result with this id, or every result taken so far when
+// id is ~0, without waiting; nullptr when none has arrived.
+DVDThread::ResultNode* DVDThread::TakeResult(u64 id)
+{
+  if (!mpsc_stack_empty(&m_result_queue))
+  {
+    mpsc_stack_node_t* const arrived = mpsc_stack_reverse(mpsc_stack_drain(&m_result_queue));
+    mpsc_stack_node_t** tail = &m_result_chain;
+    while (*tail)
+      tail = &(*tail)->next;
+    *tail = arrived;
+  }
+
+  for (mpsc_stack_node_t** link = &m_result_chain; *link; link = &(*link)->next)
+  {
+    ResultNode* const node = reinterpret_cast<ResultNode*>(*link);
+    if (id == ~u64{0} || node->result.first.id == id)
+    {
+      *link = node->link.next;
+      return node;
+    }
+  }
+  return nullptr;
+}
 
 void DVDThread::Start()
 {
@@ -53,7 +103,7 @@ void DVDThread::Stop()
   m_dvd_thread.Cancel();
   m_dvd_thread.Shutdown();
 
-  m_result_queue.Clear();
+  FreeResults();
   m_result_map.clear();
 
   m_disc.reset();
@@ -65,11 +115,13 @@ void DVDThread::DoState(PointerWrap& p)
   WaitUntilIdle();
 
   // Move all results from result_queue to result_map because
-  // PointerWrap::Do supports std::map but not Common::WaitableSPSCQueue.
+  // PointerWrap::Do supports std::map but not the queue.
   // This won't affect the behavior of FinishRead.
-  ReadResult result;
-  while (m_result_queue.Pop(result))
-    m_result_map.emplace(result.first.id, std::move(result));
+  while (ResultNode* const node = TakeResult(~u64{0}))
+  {
+    m_result_map.emplace(node->result.first.id, std::move(node->result));
+    delete node;
+  }
 
   p.Do(m_result_map);
   p.Do(m_next_id);
@@ -225,16 +277,8 @@ void DVDThread::GlobalFinishRead(Core::System& system, u64 id, s64 cycles_late)
 
 void DVDThread::FinishRead(u64 id, s64 cycles_late)
 {
-  // We can't simply pop result_queue and always get the ReadResult
-  // we want, because the DVD thread may add ReadResults to the queue
-  // in a different order than we want to get them. What we do instead
-  // is to pop the queue until we find the ReadResult we want (the one
-  // whose ID matches userdata), which means we may end up popping
-  // ReadResults that we don't want. We can't add those unwanted results
-  // back to the queue, because the queue can only have one writer.
-  // Instead, we add them to a map that only is used by the CPU thread.
-  // When this function is called again later, it will check the map for
-  // the wanted ReadResult before it starts searching through the queue.
+  // The DVD thread may finish reads in a different order than they are wanted, so results
+  // are looked up by id: in the map a savestate load filled, else among those that arrived.
   ReadResult result;
   auto it = m_result_map.find(id);
   if (it != m_result_map.end())
@@ -244,15 +288,19 @@ void DVDThread::FinishRead(u64 id, s64 cycles_late)
   }
   else
   {
-    while (true)
+    ResultNode* node;
+    while (!(node = TakeResult(id)))
     {
-      m_result_queue.WaitForData();
-      m_result_queue.Pop(result);
-      if (result.first.id == id)
-        break;
-
-      m_result_map.emplace(result.first.id, std::move(result));
+      const int key = retro_eventcount_prepare_wait(&m_result_ready);
+      if (!mpsc_stack_empty(&m_result_queue))
+      {
+        retro_eventcount_cancel_wait(&m_result_ready);
+        continue;
+      }
+      retro_eventcount_commit_wait(&m_result_ready, key);
     }
+    result = std::move(node->result);
+    delete node;
   }
   // We have now obtained the right ReadResult.
 
@@ -303,6 +351,9 @@ void DVDThread::ProcessReadRequest(ReadRequest&& request)
 
   request.realtime_done_us = Common::Timer::NowUs();
 
-  m_result_queue.Push(ReadResult(std::move(request), std::move(buffer)));
+  ResultNode* const node =
+      new ResultNode{{nullptr}, ReadResult(std::move(request), std::move(buffer))};
+  mpsc_stack_push(&m_result_queue, &node->link);
+  retro_eventcount_notify(&m_result_ready);
 }
 }  // namespace DVD

Source/Core/Core/HW/DVD/DVDThread.h

@@ -10,7 +10,6 @@
 #include <vector>
 
 #include "Common/CommonTypes.h"
-#include "Common/SPSCQueue.h"
 
 #include "Common/WorkQueueThread.h"
 #include "Core/HW/DVD/DVDInterface.h"
@@ -18,6 +17,9 @@
 
 #include "DiscIO/Volume.h"
 
+#include <queues/mpsc_stack.h>
+#include <rthreads/retro_eventcount.h>
+
 class PointerWrap;
 namespace Core
 {
@@ -128,7 +130,19 @@ private:
 
   Common::WorkQueueThreadSP<ReadRequest> m_dvd_thread;
 
-  Common::WaitableSPSCQueue<ReadResult> m_result_queue;
+  // Results travel from the DVD thread to the CPU thread through a lock-free stack; the CPU thread
+  // takes them all at once into m_result_chain, oldest first, and wakes on m_result_ready.
+  struct ResultNode
+  {
+    mpsc_stack_node_t link;
+    ReadResult result;
+  };
+  ResultNode* TakeResult(u64 id);
+  void FreeResults();
+
+  mpsc_stack_t m_result_queue;
+  retro_eventcount_t m_result_ready;
+  mpsc_stack_node_t* m_result_chain = nullptr;  // CPU thread only
   std::map<u64, ReadResult> m_result_map;
 
   std::unique_ptr<DiscIO::Volume> m_disc;

Source/Core/Core/HW/WiimoteReal/IOWin.cpp

@@ -3,6 +3,8 @@
 
 #include "Core/HW/WiimoteReal/IOWin.h"
 
+#include <cassert>
+
 #include <algorithm>
 #include <array>
 #include <bit>

Source/Core/Core/HW/WiimoteReal/IOhidapi.cpp

@@ -3,6 +3,8 @@
 
 #include "Core/HW/WiimoteReal/IOhidapi.h"
 
+#include <cassert>
+
 #if defined(__linux__)
 #include <filesystem>
 #endif

Source/Core/Core/HW/WiimoteReal/WiimoteReal.cpp

@@ -9,6 +9,7 @@
 
 #include <SFML/Network/UdpSocket.hpp>
 
+#include "Common/Assert.h"
 #include "Common/CommonTypes.h"
 #include "Common/Config/Config.h"
 #include "Common/Swap.h"
@@ -145,8 +146,14 @@ void AddWiimoteToPool(std::unique_ptr<Wiimote> wiimote)
   s_wiimote_pool.emplace_back(WiimotePoolEntry{std::move(wiimote)});
 }
 
+// Records each ring holds: 256 reports is over a second of input at the Wiimote's 200 Hz.
+static constexpr size_t REPORT_RING_RECORDS = 256;
+
 Wiimote::Wiimote()
 {
+  retro_spsc_init(&m_read_reports, REPORT_RING_RECORDS * sizeof(ReadRecord));
+  retro_spsc_init(&m_write_reports, REPORT_RING_RECORDS * sizeof(WriteRecord));
+  retro_eventcount_init(&m_write_event);
   m_config_changed_callback_id = Config::AddConfigChangedCallback([this] { RefreshConfig(); });
   RefreshConfig();
 }
@@ -154,6 +161,22 @@ Wiimote::Wiimote()
 Wiimote::~Wiimote()
 {
   Config::RemoveConfigChangedCallback(m_config_changed_callback_id);
+  retro_eventcount_free(&m_write_event);
+  retro_spsc_free(&m_write_reports);
+  retro_spsc_free(&m_read_reports);
+}
+
+// Producer side of m_write_reports. A ring this deep is only full when the device has stopped
+// taking reports, and the write thread is about to drop the connection then.
+void Wiimote::QueueWrite(TimePoint time, const u8* data, size_t size)
+{
+  ASSERT(size <= MAX_PAYLOAD);
+  WriteRecord record;
+  record.time = time;
+  record.size = static_cast<u8>(size);
+  std::copy_n(data, size, record.data);
+  retro_spsc_write_frames(&m_write_reports, &record, 1, sizeof(record));
+  retro_eventcount_notify(&m_write_event);
 }
 
 void Wiimote::Shutdown()
@@ -202,8 +225,7 @@ void Wiimote::WriteReport(Report rpt)
   const auto report_time =
       Core::IsCPUThread() ? core_timing.GetTargetHostTime(core_timing.GetTicks()) : Clock::now();
 
-  m_write_reports.Emplace(report_time, std::move(rpt));
-  m_write_event.Set();
+  QueueWrite(report_time, rpt.data(), rpt.size());
 }
 
 // to be called from CPU thread
@@ -232,12 +254,8 @@ void Wiimote::ResetDataReporting()
 
 void Wiimote::ClearReadQueue()
 {
-  Report rpt;
-
-  // The "Clear" function isn't thread-safe :/
-  while (m_read_reports.Pop(rpt))
-  {
-  }
+  // Consumer side: discards what the read thread has delivered so far.
+  retro_spsc_skip(&m_read_reports, retro_spsc_read_avail(&m_read_reports));
 }
 
 void Wiimote::EventLinked()
@@ -293,41 +311,47 @@ void Wiimote::InterruptDataOutput(const u8* data, const u32 size)
 
 bool Wiimote::Read()
 {
-  Report rpt(MAX_PAYLOAD);
-  auto const result = IORead(rpt.data());
+  // Receive straight into the next ring record. With the ring full the core has stopped taking
+  // reports; receive into a scratch record and drop it.
+  ReadRecord scratch;
+  ReadRecord* record = &scratch;
+  void* span;
+  const bool reserved = retro_spsc_write_begin(&m_read_reports, &span) >= sizeof(ReadRecord);
+  if (reserved)
+    record = static_cast<ReadRecord*>(span);
+
+  auto const result = IORead(record->data);
 
   // Drop the report if not connected.
-  if (m_is_linked && result > 0)
+  const bool keep = m_is_linked && result > 0;
+  if (keep && m_balance_board_dump_port > 0 && m_index == WIIMOTE_BALANCE_BOARD)
   {
-    if (m_balance_board_dump_port > 0 && m_index == WIIMOTE_BALANCE_BOARD)
-    {
-      static sf::UdpSocket Socket;
-      (void)Socket.send(rpt.data(), rpt.size(), sf::IpAddress::LocalHost,
-                        m_balance_board_dump_port);
-    }
+    static sf::UdpSocket Socket;
+    (void)Socket.send(record->data, MAX_PAYLOAD, sf::IpAddress::LocalHost,
+                      m_balance_board_dump_port);
+  }
 
-    // Add it to queue
-    rpt.resize(result);
-    m_read_reports.Push(std::move(rpt));
+  if (reserved)
+  {
+    record->size = static_cast<u8>(result);
+    retro_spsc_write_end(&m_read_reports, keep ? sizeof(ReadRecord) : 0);
   }
 
   return result != 0;
 }
 
-bool Wiimote::Write(const TimedReport& timed_report)
+bool Wiimote::Write(const WriteRecord& record)
 {
-  auto const& rpt = timed_report.report;
-
   if (m_balance_board_dump_port > 0 && m_index == WIIMOTE_BALANCE_BOARD)
   {
     static sf::UdpSocket Socket;
-    (void)Socket.send((char*)rpt.data(), rpt.size(), sf::IpAddress::LocalHost,
+    (void)Socket.send(record.data, record.size, sf::IpAddress::LocalHost,
                       m_balance_board_dump_port);
   }
 
   // Write the report at the proper time, mainly for speaker data, not that it will help much.
-  std::this_thread::sleep_until(timed_report.time);
-  int ret = IOWrite(rpt.data(), rpt.size());
+  std::this_thread::sleep_until(record.time);
+  int ret = IOWrite(record.data, record.size);
 
   return ret != 0;
 }
@@ -416,7 +440,13 @@ static bool IsDataReport(const Report& rpt)
 
 bool Wiimote::GetNextReport(Report* report)
 {
-  return m_read_reports.Pop(*report);
+  const void* span;
+  if (retro_spsc_read_begin(&m_read_reports, &span) < sizeof(ReadRecord))
+    return false;
+  const ReadRecord* const record = static_cast<const ReadRecord*>(span);
+  report->assign(record->data, record->data + record->size);
+  retro_spsc_read_end(&m_read_reports, sizeof(ReadRecord));
+  return true;
 }
 
 // Returns the next report that should be sent
@@ -503,15 +533,13 @@ void Wiimote::Prepare()
   const auto now = Clock::now();
 
   // Set reporting mode to non-continuous core buttons and turn on rumble.
-  Report mode_report = {WR_SET_REPORT | BT_OUTPUT, u8(OutputReportID::ReportMode), 1,
-                        u8(InputReportID::ReportCore)};
-  m_write_reports.Emplace(now, std::move(mode_report));
+  const u8 mode_report[] = {WR_SET_REPORT | BT_OUTPUT, u8(OutputReportID::ReportMode), 1,
+                            u8(InputReportID::ReportCore)};
+  QueueWrite(now, mode_report, sizeof(mode_report));
 
   // Request status and turn off rumble.
-  Report req_status_report = {WR_SET_REPORT | BT_OUTPUT, u8(OutputReportID::RequestStatus), 0};
-  m_write_reports.Emplace(now + std::chrono::milliseconds{200}, std::move(req_status_report));
-
-  m_write_event.Set();
+  const u8 req_status_report[] = {WR_SET_REPORT | BT_OUTPUT, u8(OutputReportID::RequestStatus), 0};
+  QueueWrite(now + std::chrono::milliseconds{200}, req_status_report, sizeof(req_status_report));
 }
 
 void Wiimote::EmuStop()
@@ -778,7 +806,7 @@ void Wiimote::StopThread()
     return;
 
   m_run_thread.Clear();
-  m_write_event.Set();
+  retro_eventcount_notify(&m_write_event);
 
   // Note that the write thread stops the read thread.
   m_write_thread.join();
@@ -794,7 +822,7 @@ void Wiimote::ReadThreadFunc()
     {
       WARN_LOG_FMT(WIIMOTE, "Wiimote::Read failed on Wiimote {}.", m_index + 1);
       m_run_thread.Clear();
-      m_write_event.Set();
+      retro_eventcount_notify(&m_write_event);
       break;
     }
   }
@@ -834,11 +862,12 @@ void Wiimote::WriteThreadFunc()
   {
     bool write_success = false;
 
-    if (!m_write_reports.Empty())
+    if (retro_spsc_read_avail(&m_write_reports) >= sizeof(WriteRecord))
     {
       // Send a normal report.
-      write_success = Write(m_write_reports.Front());
-      m_write_reports.Pop();
+      WriteRecord record;
+      retro_spsc_read(&m_write_reports, &record, sizeof(record));
+      write_success = Write(record);
     }
     else if (Clock::now() - last_write_time >= WRITE_TEST_INTERVAL)
     {
@@ -851,7 +880,13 @@ void Wiimote::WriteThreadFunc()
     else
     {
       // Nothing to do. Wait a while for a kick.
-      m_write_event.WaitFor(WRITE_TEST_INTERVAL);
+      const int key = retro_eventcount_prepare_wait(&m_write_event);
+      if (retro_spsc_read_avail(&m_write_reports) >= sizeof(WriteRecord) || !m_run_thread.IsSet())
+        retro_eventcount_cancel_wait(&m_write_event);
+      else
+        retro_eventcount_commit_wait_timeout(
+            &m_write_event, key,
+            std::chrono::duration_cast<std::chrono::microseconds>(WRITE_TEST_INTERVAL).count());
       continue;
     }
 

Source/Core/Core/HW/WiimoteReal/WiimoteReal.h

@@ -14,7 +14,6 @@
 #include "Common/Config/Config.h"
 #include "Common/Event.h"
 #include "Common/Flag.h"
-#include "Common/SPSCQueue.h"
 #include "Common/WorkQueueThread.h"
 
 #include "Core/HW/Wiimote.h"
@@ -23,6 +22,9 @@
 #include "Core/HW/WiimoteCommon/WiimoteReport.h"
 #include "Core/USBUtils.h"
 
+#include <retro_spsc.h>
+#include <rthreads/retro_eventcount.h>
+
 class PointerWrap;
 
 namespace WiimoteReal
@@ -115,14 +117,25 @@ protected:
   u8 m_bt_device_index = 0;
 
 private:
-  struct TimedReport
+  // Reports cross threads as fixed 32-byte records in lock-free rings: a record never straddles
+  // the end of a ring, so the read thread receives straight into one.
+  struct ReadRecord
+  {
+    u8 size;
+    u8 data[MAX_PAYLOAD];
+    u8 pad[32 - 1 - MAX_PAYLOAD];
+  };
+  struct WriteRecord
   {
     TimePoint time;
-    Report report;
+    u8 size;
+    u8 data[MAX_PAYLOAD];
   };
+  static_assert(sizeof(ReadRecord) == 32 && sizeof(WriteRecord) == 32);
 
+  void QueueWrite(TimePoint time, const u8* data, size_t size);
   bool Read();
-  bool Write(const TimedReport& timed_report);
+  bool Write(const WriteRecord& record);
 
   void StartThread();
   void StopThread();
@@ -165,10 +178,10 @@ private:
   // Triggered when the thread has finished ConnectInternal.
   Common::Event m_thread_ready_event;
 
-  Common::SPSCQueue<Report> m_read_reports;
-  Common::SPSCQueue<TimedReport> m_write_reports;
+  retro_spsc_t m_read_reports;
+  retro_spsc_t m_write_reports;
   // Kick the write thread.
-  Common::Event m_write_event;
+  retro_eventcount_t m_write_event;
 
   bool m_speaker_enabled_in_dolphin_config = false;
   int m_balance_board_dump_port = 0;

Source/Core/Core/IOS/Network/KD/NetKDRequest.cpp

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: GPL-2.0-or-later
 
 #include "Core/IOS/Network/KD/NetKDRequest.h"
+#include "Common/Thread.h"
 
 #include <algorithm>
 #include <array>

Source/Core/Core/IOS/Network/KD/NetKDRequest.h

@@ -7,6 +7,7 @@
 #include <memory>
 #include <queue>
 #include <string>
+#include <thread>
 
 #include "Common/CommonTypes.h"
 #include "Common/Event.h"

Source/Core/Core/IOS/USB/Bluetooth/BTReal.cpp

@@ -3,6 +3,8 @@
 
 #include "Core/IOS/USB/Bluetooth/BTReal.h"
 
+#include <cassert>
+
 #include <algorithm>
 #include <cstdlib>
 #include <cstring>
@@ -302,7 +304,7 @@ void BluetoothRealDevice::TryToFillACLEndpoint()
   auto& core_timing = GetSystem().GetCoreTiming();
   core_timing.Throttle(core_timing.GetTicks());
 
-  const auto buffer = m_lib_usb_bt_adapter->ReceiveACLData();
+  const auto buffer = m_lib_usb_bt_adapter->PeekACLData();
   if (buffer.empty())
     return;
 
@@ -310,6 +312,7 @@ void BluetoothRealDevice::TryToFillACLEndpoint()
 
   m_acl_endpoint->FillBuffer(buffer.data(), buffer.size());
   GetEmulationKernel().EnqueueIPCReply(m_acl_endpoint->ios_request, s32(buffer.size()));
+  m_lib_usb_bt_adapter->PopACLData();
 
   m_acl_endpoint.reset();
 }

Source/Core/Core/IOS/USB/Bluetooth/LibUSBBluetoothAdapter.cpp

@@ -2,6 +2,9 @@
 // SPDX-License-Identifier: GPL-2.0-or-later
 
 #include "Core/IOS/USB/Bluetooth/LibUSBBluetoothAdapter.h"
+#include "Common/Thread.h"
+
+#include <cassert>
 
 #include <algorithm>
 #include <chrono>
@@ -86,6 +89,11 @@ bool LibUSBBluetoothAdapter::IsConfiguredBluetoothDevice(u16 vid, u16 pid)
 
 LibUSBBluetoothAdapter::LibUSBBluetoothAdapter()
 {
+  retro_atomic_int_init(&m_run_input_transfers, 1);
+  mpsc_stack_init(&m_transfers_to_free);
+  retro_spsc_init_mirrored(&m_hci_event_queue, INPUT_RING_SIZE);
+  retro_spsc_init_mirrored(&m_acl_data_queue, INPUT_RING_SIZE);
+
   if (!m_context.IsValid())
     return;
 
@@ -175,9 +183,15 @@ LibUSBBluetoothAdapter::LibUSBBluetoothAdapter()
 
 LibUSBBluetoothAdapter::~LibUSBBluetoothAdapter()
 {
-  if (m_handle == nullptr)
-    return;
+  if (m_handle != nullptr)
+    Close();
 
+  retro_spsc_free(&m_acl_data_queue);
+  retro_spsc_free(&m_hci_event_queue);
+}
+
+void LibUSBBluetoothAdapter::Close()
+{
   // Wait for completion (or time out) of all HCI commands.
   while (AreCommandsPendingResponse())
   {
@@ -188,16 +202,19 @@ LibUSBBluetoothAdapter::~LibUSBBluetoothAdapter()
   m_output_worker.Shutdown();
 
   // Stop the repeating input transfers.
-  if (std::lock_guard lg{m_transfers_mutex}; true)
-    m_run_input_transfers = false;
+  retro_atomic_store_release_int(&m_run_input_transfers, 0);
 
-  // Cancel all LibUSB transfers.
-  std::ranges::for_each(m_transfer_buffers | std::ranges::views::keys, libusb_cancel_transfer);
-
-  // Wait for transfer callbacks and clean up all the buffers.
+  // Cancel all LibUSB transfers, wait for their callbacks and clean up all the buffers. A
+  // callback that read the run flag before it was cleared may resubmit its transfer after the
+  // cancel, so whatever is still in flight is cancelled again on every pass.
   while (!m_transfer_buffers.empty())
   {
-    m_transfers_to_free.WaitForData();
+    std::ranges::for_each(m_transfer_buffers | std::ranges::views::keys, libusb_cancel_transfer);
+    const int key = retro_eventcount_prepare_wait(&m_transfer_freed.ec);
+    if (!mpsc_stack_empty(&m_transfers_to_free))
+      retro_eventcount_cancel_wait(&m_transfer_freed.ec);
+    else
+      retro_eventcount_commit_wait_timeout(&m_transfer_freed.ec, key, 10000);
     CleanCompletedTransfers();
   }
 
@@ -236,16 +253,29 @@ void LibUSBBluetoothAdapter::Update()
   CleanCompletedTransfers();
 }
 
+// Consumer side of an input ring: the length of the whole packet at its front, or 0.
+static u16 PeekPacketLength(const retro_spsc_t* queue)
+{
+  const std::size_t avail = retro_spsc_read_avail(queue);
+  u16 length;
+  if (avail < sizeof(length))
+    return 0;
+  retro_spsc_peek(queue, &length, sizeof(length));
+  return avail >= sizeof(length) + length ? length : 0;
+}
+
 auto LibUSBBluetoothAdapter::ReceiveHCIEvent() -> BufferType
 {
-  if (m_hci_event_queue.Empty())
+  const u16 length = PeekPacketLength(&m_hci_event_queue);
+  if (length == 0)
   {
     Update();
     return {};
   }
 
-  auto buffer = std::move(m_hci_event_queue.Front());
-  m_hci_event_queue.Pop();
+  BufferType buffer(length);
+  retro_spsc_skip(&m_hci_event_queue, sizeof(length));
+  retro_spsc_read(&m_hci_event_queue, buffer.data(), length);
 
   // We only care about "COMMAND_COMPL" and "COMMAND_STATUS" events here.
   // The controller will reply with one of those for every command.
@@ -264,11 +294,26 @@ auto LibUSBBluetoothAdapter::ReceiveHCIEvent() -> BufferType
   return buffer;
 }
 
-auto LibUSBBluetoothAdapter::ReceiveACLData() -> BufferType
+std::span<const u8> LibUSBBluetoothAdapter::PeekACLData()
 {
-  BufferType buffer;
-  m_acl_data_queue.Pop(buffer);
-  return buffer;
+  const u16 length = PeekPacketLength(&m_acl_data_queue);
+  if (length == 0)
+    return {};
+
+  m_acl_peeked = sizeof(length) + length;
+  const void* span;
+  if (retro_spsc_read_begin(&m_acl_data_queue, &span) >= m_acl_peeked)
+    return {static_cast<const u8*>(span) + sizeof(length), length};
+
+  // Only a ring without a mirror splits a packet at its end.
+  m_acl_bounce.resize(m_acl_peeked);
+  retro_spsc_peek(&m_acl_data_queue, m_acl_bounce.data(), m_acl_peeked);
+  return {m_acl_bounce.data() + sizeof(length), length};
+}
+
+void LibUSBBluetoothAdapter::PopACLData()
+{
+  retro_spsc_skip(&m_acl_data_queue, std::exchange(m_acl_peeked, 0));
 }
 
 bool LibUSBBluetoothAdapter::IsControllerReadyForCommand() const
@@ -305,12 +350,13 @@ void LibUSBBluetoothAdapter::AcknowledgeCommand(std::span<const u8> buffer)
 
 libusb_transfer* LibUSBBluetoothAdapter::AllocateTransfer(std::size_t buffer_size)
 {
-  auto& item =
-      *m_transfer_buffers.emplace(libusb_alloc_transfer(0), Common::UniqueBuffer<u8>{buffer_size})
-           .first;
+  auto& item = *m_transfer_buffers
+                    .emplace(libusb_alloc_transfer(0),
+                             Common::UniqueBuffer<u8>{BUFFER_HEADER_SIZE + buffer_size})
+                    .first;
 
   auto* const transfer = item.first;
-  transfer->buffer = item.second.get();
+  transfer->buffer = item.second.get() + BUFFER_HEADER_SIZE;
 
   return transfer;
 }
@@ -321,11 +367,25 @@ void LibUSBBluetoothAdapter::FreeTransfer(libusb_transfer* transfer)
   m_transfer_buffers.erase(transfer);
 }
 
+// Any thread.
+void LibUSBBluetoothAdapter::QueueFree(libusb_transfer* transfer)
+{
+  FreeNode* const node = reinterpret_cast<FreeNode*>(transfer->buffer - BUFFER_HEADER_SIZE);
+  node->transfer = transfer;
+  mpsc_stack_push(&m_transfers_to_free, &node->link);
+  retro_eventcount_notify(&m_transfer_freed.ec);
+}
+
 void LibUSBBluetoothAdapter::CleanCompletedTransfers()
 {
-  libusb_transfer* transfer{};
-  while (m_transfers_to_free.Pop(transfer))
-    FreeTransfer(transfer);
+  mpsc_stack_node_t* node = mpsc_stack_drain(&m_transfers_to_free);
+  while (node)
+  {
+    // The node is freed with its transfer's buffer.
+    mpsc_stack_node_t* const next = node->next;
+    FreeTransfer(reinterpret_cast<FreeNode*>(node)->transfer);
+    node = next;
+  }
 }
 
 void LibUSBBluetoothAdapter::PushHCICommand(TimedTransfer transfer)
@@ -365,8 +425,7 @@ void LibUSBBluetoothAdapter::SubmitTransfer(libusb_transfer* transfer)
   ERROR_LOG_FMT(IOS_WIIMOTE, "libusb_submit_transfer: {}", LibusbUtils::ErrorWrap(ret));
 
   // Failed transers will not invoke callbacks so we must mark the buffer to be free'd here.
-  std::lock_guard lk{m_transfers_mutex};
-  m_transfers_to_free.Emplace(transfer);
+  QueueFree(transfer);
 }
 
 void LibUSBBluetoothAdapter::ScheduleBulkTransfer(u8 endpoint, std::span<const u8> data,
@@ -520,9 +579,7 @@ bool LibUSBBluetoothAdapter::OpenDevice(const libusb_device_descriptor& device_d
 void LibUSBBluetoothAdapter::HandleOutputTransfer(libusb_transfer* tr)
 {
   HandleTransferError(tr);
-
-  std::lock_guard lg{m_transfers_mutex};
-  m_transfers_to_free.Emplace(tr);
+  QueueFree(tr);
 }
 
 void LibUSBBluetoothAdapter::HandleInputTransfer(libusb_transfer* tr)
@@ -530,15 +587,16 @@ void LibUSBBluetoothAdapter::HandleInputTransfer(libusb_transfer* tr)
   if (tr->status == LIBUSB_TRANSFER_COMPLETED)
   {
     const bool is_hci_event = tr->endpoint == HCI_EVENT;
-    auto& queue = is_hci_event ? m_hci_event_queue : m_acl_data_queue;
+    retro_spsc_t* const queue = is_hci_event ? &m_hci_event_queue : &m_acl_data_queue;
+    const u16 length = static_cast<u16>(tr->actual_length);
 
-    if (queue.Size() < MAX_INPUT_QUEUE_SIZE)
+    if (retro_spsc_write_avail(queue) >= sizeof(length) + length)
     {
-      if (tr->actual_length != 0)
+      if (length != 0)
       {
-        BufferType buffer(tr->actual_length);
-        std::copy_n(tr->buffer, tr->actual_length, buffer.data());
-        queue.Emplace(std::move(buffer));
+        // The consumer only takes a packet once all of it is there.
+        retro_spsc_write(queue, &length, sizeof(length));
+        retro_spsc_write(queue, tr->buffer, length);
       }
 
       m_showed_long_queue_drop = false;
@@ -553,9 +611,7 @@ void LibUSBBluetoothAdapter::HandleInputTransfer(libusb_transfer* tr)
 
   HandleTransferError(tr);
 
-  std::lock_guard lg{m_transfers_mutex};
-
-  if (m_run_input_transfers)
+  if (retro_atomic_load_acquire_int(&m_run_input_transfers))
   {
     // Continuously repeat this input transfer so long as we're still running.
     const auto ret = libusb_submit_transfer(tr);
@@ -566,7 +622,7 @@ void LibUSBBluetoothAdapter::HandleInputTransfer(libusb_transfer* tr)
                   LibusbUtils::ErrorWrap(ret));
   }
 
-  m_transfers_to_free.Emplace(tr);
+  QueueFree(tr);
 }
 
 void LibUSBBluetoothAdapter::HandleTransferError(libusb_transfer* transfer)

Source/Core/Core/IOS/USB/Bluetooth/LibUSBBluetoothAdapter.h

@@ -7,6 +7,7 @@
 #include <queue>
 #include <span>
 #include <string>
+#include <vector>
 
 #include "Common/Buffer.h"
 #include "Common/Timer.h"
@@ -15,6 +16,11 @@
 #include "Core/IOS/USB/Bluetooth/hci.h"
 #include "Core/LibusbUtils.h"
 
+#include <queues/mpsc_stack.h>
+#include <retro_atomic.h>
+#include <retro_spsc.h>
+#include <rthreads/retro_eventcount.h>
+
 struct libusb_device_handle;
 struct libusb_device_descriptor;
 struct libusb_transfer;
@@ -43,7 +49,11 @@ public:
   // Return a packet at the front of the queue, or an empty buffer when nothing is queued.
   // ReceiveHCIEvent is expected to be called regularly to do some internal processing.
   [[nodiscard]] BufferType ReceiveHCIEvent();
-  [[nodiscard]] BufferType ReceiveACLData();
+
+  // The ACL packet at the front of the queue, read in place; empty when nothing is queued.
+  // PopACLData releases it.
+  [[nodiscard]] std::span<const u8> PeekACLData();
+  void PopACLData();
 
   // Schedule a transfer to submit at a specific time.
   void ScheduleBulkTransfer(u8 endpoint, std::span<const u8> data, TimePoint target_time);
@@ -61,28 +71,43 @@ public:
   bool IsWiiBTModule() const;
 
 private:
-  // Inputs will be dropped when queue is full.
-  static constexpr std::size_t MAX_INPUT_QUEUE_SIZE = 100;
+  // Bytes each input ring holds; inputs are dropped when it is full.
+  static constexpr std::size_t INPUT_RING_SIZE = 64 * 1024;
 
   static constexpr u8 INTERFACE = 0x00;
 
   std::string m_last_open_error;
 
+  // Callbacks on the libusb event thread notify this until that thread is joined, which happens
+  // when m_context goes, so it is declared first to be released last.
+  struct TransferFreedEvent
+  {
+    TransferFreedEvent() { retro_eventcount_init(&ec); }
+    ~TransferFreedEvent() { retro_eventcount_free(&ec); }
+    retro_eventcount_t ec;
+  } m_transfer_freed;
+
   LibusbUtils::Context m_context;
   libusb_device_handle* m_handle = nullptr;
 
-  // Protects run-flag and Push'ing to the completed transfer queue.
-  std::mutex m_transfers_mutex;
+  retro_atomic_int_t m_run_input_transfers;
 
-  bool m_run_input_transfers = true;
+  // Callback and worker threads push transfers here, lock-free, when they are ready for clean up.
+  // The node lives in the front of the transfer's own buffer, so this allocates nothing.
+  struct FreeNode
+  {
+    mpsc_stack_node_t link;
+    libusb_transfer* transfer;
+  };
+  static constexpr std::size_t BUFFER_HEADER_SIZE = (sizeof(FreeNode) + 15) & ~std::size_t{15};
+  void QueueFree(libusb_transfer* transfer);
+  mpsc_stack_t m_transfers_to_free;
 
-  // callback/worker threads push transfers here that are ready for clean up.
-  // This avoids locking around m_transfer_buffers.
-  Common::WaitableSPSCQueue<libusb_transfer*> m_transfers_to_free;
-
-  // Incoming packets Push'd from libusb callback thread.
-  Common::SPSCQueue<BufferType> m_hci_event_queue;
-  Common::SPSCQueue<BufferType> m_acl_data_queue;
+  // Incoming packets from the libusb callback thread, each a 16-bit length and the payload.
+  retro_spsc_t m_hci_event_queue;
+  retro_spsc_t m_acl_data_queue;
+  std::size_t m_acl_peeked = 0;
+  std::vector<u8> m_acl_bounce;
 
   // All transfers are alloc'd and kept here until complete.
   std::map<libusb_transfer*, Common::UniqueBuffer<u8>> m_transfer_buffers;
@@ -137,6 +162,7 @@ private:
   void AcknowledgeCommand(std::span<const u8> buffer);
 
   libusb_transfer* AllocateTransfer(std::size_t buffer_size);
+  void Close();
   void FreeTransfer(libusb_transfer*);
   void CleanCompletedTransfers();
 

Source/Core/Core/IOS/USB/Emulated/LogitechMic.cpp

@@ -3,6 +3,8 @@
 
 #include "Core/IOS/USB/Emulated/LogitechMic.h"
 
+#include <cassert>
+
 #include <algorithm>
 #include <utility>
 

Source/Core/DolphinLibretro/Boot.cpp

@@ -6,6 +6,7 @@
 #include <fstream>
 
 #include "Common/CommonPaths.h"
+#include "Common/Thread.h"
 #include "Core/CommonTitles.h"
 #include "Common/FileUtil.h"
 #include "Common/StringUtil.h"

Source/Core/VideoCommon/AsyncRequests.cpp

@@ -4,43 +4,64 @@
 #include "VideoCommon/AsyncRequests.h"
 
 #include "Core/System.h"
-
 #include "VideoCommon/Fifo.h"
 #include "VideoCommon/VertexManagerBase.h"
-#include "VideoCommon/VideoBackendBase.h"
-#include "VideoCommon/VideoEvents.h"
 
 AsyncRequests AsyncRequests::s_singleton;
 
-AsyncRequests::AsyncRequests() = default;
+// Pending requests the ring holds before the CPU thread has to wait for the video thread.
+static constexpr size_t QUEUE_REQUESTS = 1024;
+
+AsyncRequests::AsyncRequests()
+{
+  retro_waitable_spsc_init(&m_queue, QUEUE_REQUESTS * sizeof(Request*));
+  retro_eventcount_init(&m_completed);
+}
+
+AsyncRequests::~AsyncRequests()
+{
+  retro_eventcount_free(&m_completed);
+  retro_waitable_spsc_free(&m_queue);
+}
 
 void AsyncRequests::PullEvents()
 {
-  if (m_queue.Empty())
+  if (retro_spsc_read_avail(&m_queue.queue) < sizeof(Request*))
     return;
 
   // This is only called if the queue isn't empty.
   // So just flush the pipeline to get accurate results.
   g_vertex_manager->Flush();
 
-  while (!m_queue.Empty())
-  {
-    std::invoke(std::move(m_queue.Front()));
-    m_queue.Pop();
-  }
+  Request* request;
+  while (retro_waitable_spsc_read(&m_queue, &request, sizeof(request)) == sizeof(request))
+    request->run(request);
 }
 
-void AsyncRequests::QueueEvent(Event&& event)
+void AsyncRequests::Submit(Request* request)
 {
-  m_queue.Push(std::move(event));
+  // Only waits when the video thread is QUEUE_REQUESTS requests behind.
+  while (!retro_waitable_spsc_wait_writable(&m_queue, sizeof(request), 1000000))
+  {
+  }
+  retro_waitable_spsc_write(&m_queue, &request, sizeof(request));
 
   auto& system = Core::System::GetInstance();
   system.GetFifo().RunGpu();
 }
 
-void AsyncRequests::WaitForEmptyQueue()
+void AsyncRequests::WaitDone(retro_atomic_int_t* done)
 {
-  m_queue.WaitForEmpty();
+  while (!retro_atomic_load_acquire_int(done))
+  {
+    const int key = retro_eventcount_prepare_wait(&m_completed);
+    if (retro_atomic_load_acquire_int(done))
+    {
+      retro_eventcount_cancel_wait(&m_completed);
+      break;
+    }
+    retro_eventcount_commit_wait(&m_completed, key);
+  }
 }
 
 void AsyncRequests::SetPassthrough(bool enable)

Source/Core/VideoCommon/AsyncRequests.h

@@ -5,25 +5,30 @@
 
 #include <concepts>
 #include <functional>
-#include <future>
+#include <optional>
+#include <type_traits>
+#include <utility>
 
-#include "Common/Functional.h"
-#include "Common/SPSCQueue.h"
+#include <retro_atomic.h>
+#include <retro_waitable_spsc.h>
+#include <rthreads/retro_eventcount.h>
 
 struct EfbPokeData;
 class PointerWrap;
 
+// Requests from the CPU thread to the video thread. The CPU thread is the only producer and the
+// video thread the only consumer, so the requests travel through a lock-free SPSC ring as
+// pointers: a blocking request lives on the caller's stack and allocates nothing.
 class AsyncRequests
 {
 public:
   AsyncRequests();
+  ~AsyncRequests();
 
   // Called from the Video thread.
   void PullEvents();
 
   // The following are called from the CPU thread.
-  void WaitForEmptyQueue();
-
   template <std::invocable<> F>
   void PushEvent(F&& callback)
   {
@@ -33,7 +38,18 @@ public:
       return;
     }
 
-    QueueEvent(Event{std::forward<F>(callback)});
+    using Fn = std::decay_t<F>;
+    struct Owned : Request
+    {
+      Fn fn;
+      static void Run(Request* r)
+      {
+        Owned* const self = static_cast<Owned*>(r);
+        std::invoke(self->fn);
+        delete self;
+      }
+    };
+    Submit(new Owned{{&Owned::Run}, Fn(std::forward<F>(callback))});
   }
 
   template <std::invocable<> F>
@@ -42,10 +58,33 @@ public:
     if (m_passthrough)
       return std::invoke(std::forward<F>(callback));
 
-    std::packaged_task task{std::forward<F>(callback)};
-    QueueEvent(Event{[&] { task(); }});
-
-    return task.get_future().get();
+    using R = std::invoke_result_t<F>;
+    using Fn = std::remove_reference_t<F>;
+    struct Blocking : Request
+    {
+      Fn* fn;
+      retro_atomic_int_t done;
+      std::conditional_t<std::is_void_v<R>, char, std::optional<R>> result;
+      static void Run(Request* r)
+      {
+        Blocking* const self = static_cast<Blocking*>(r);
+        if constexpr (std::is_void_v<R>)
+          std::invoke(*self->fn);
+        else
+          self->result.emplace(std::invoke(*self->fn));
+        // The caller owns this request again from here on.
+        retro_atomic_store_release_int(&self->done, 1);
+        retro_eventcount_notify(&s_singleton.m_completed);
+      }
+    };
+    Blocking request;
+    request.run = &Blocking::Run;
+    request.fn = &callback;
+    retro_atomic_int_init(&request.done, 0);
+    Submit(&request);
+    WaitDone(&request.done);
+    if constexpr (!std::is_void_v<R>)
+      return std::move(*request.result);
   }
 
   // Not thread-safe. Only set during initialization.
@@ -54,13 +93,17 @@ public:
   static AsyncRequests* GetInstance() { return &s_singleton; }
 
 private:
-  using Event = Common::MoveOnlyFunction<void()>;
+  struct Request
+  {
+    void (*run)(Request*);
+  };
 
-  void QueueEvent(Event&& event);
+  void Submit(Request* request);
+  void WaitDone(retro_atomic_int_t* done);
 
   static AsyncRequests s_singleton;
 
-  Common::WaitableSPSCQueue<Event> m_queue;
-
+  retro_waitable_spsc_t m_queue;
+  retro_eventcount_t m_completed;
   bool m_passthrough = true;
 };

Source/Core/VideoCommon/PerformanceTracker.cpp

@@ -23,12 +23,18 @@ PerformanceTracker::PerformanceTracker(std::optional<std::string> log_name,
                                        const std::optional<DT> sample_window_duration)
     : m_log_name{std::move(log_name)}, m_sample_window_duration{sample_window_duration}
 {
+  retro_spsc_init(&m_raw_dts, MAX_DT_QUEUE_SIZE * sizeof(DT));
   Reset();
 }
 
+PerformanceTracker::~PerformanceTracker()
+{
+  retro_spsc_free(&m_raw_dts);
+}
+
 void PerformanceTracker::Reset()
 {
-  m_raw_dts.Clear();
+  retro_spsc_skip(&m_raw_dts, retro_spsc_read_avail(&m_raw_dts));
   m_dt_queue.clear();
 
   m_dt_total = DT::zero();
@@ -54,14 +60,18 @@ void PerformanceTracker::Count()
   }
 
   m_last_raw_dt = diff;
-  m_raw_dts.Push(diff);
+  // A full ring already holds the MAX_DT_QUEUE_SIZE samples UpdateStats keeps.
+  retro_spsc_write_frames(&m_raw_dts, &diff, 1, sizeof(diff));
 }
 
 void PerformanceTracker::UpdateStats()
 {
   DT diff{};
-  while (m_raw_dts.Pop(diff))
+  while (retro_spsc_read_avail(&m_raw_dts) >= sizeof(diff))
+  {
+    retro_spsc_read(&m_raw_dts, &diff, sizeof(diff));
     HandleRawDt(diff);
+  }
 
   // Update Std Dev
   MathUtil::RunningVariance<double> variance;

Source/Core/VideoCommon/PerformanceTracker.h

@@ -9,14 +9,15 @@
 #include <optional>
 
 #include "Common/CommonTypes.h"
-#include "Common/SPSCQueue.h"
+
+#include <retro_spsc.h>
 
 class PerformanceTracker
 {
 public:
   PerformanceTracker(std::optional<std::string> log_name = std::nullopt,
                      const std::optional<DT> sample_window_duration = std::nullopt);
-  ~PerformanceTracker() = default;
+  ~PerformanceTracker();
 
   PerformanceTracker(const PerformanceTracker&) = delete;
   PerformanceTracker& operator=(const PerformanceTracker&) = delete;
@@ -58,7 +59,8 @@ private:
 
   // Push'd from Count()
   //  and Pop'd from UpdateStats()
-  Common::SPSCQueue<DT> m_raw_dts;
+  // Frame times from Count() to UpdateStats(), lock-free.
+  retro_spsc_t m_raw_dts;
   std::atomic<DT> m_last_raw_dt = DT::zero();
 
   // Amount of time to sample dt's over (defaults to config)

Source/UnitTests/Common/SPSCQueueTest.cpp

@@ -1,91 +0,0 @@
-// Copyright 2014 Dolphin Emulator Project
-// SPDX-License-Identifier: GPL-2.0-or-later
-
-#include <gtest/gtest.h>
-
-#include <memory>
-#include <thread>
-
-#include "Common/CommonTypes.h"
-#include "Common/SPSCQueue.h"
-
-TEST(SPSCQueue, Simple)
-{
-  Common::SPSCQueue<u32> q;
-
-  EXPECT_EQ(0u, q.Size());
-  EXPECT_TRUE(q.Empty());
-
-  q.Push(1);
-  EXPECT_EQ(1u, q.Size());
-  EXPECT_FALSE(q.Empty());
-
-  u32 v;
-  q.Pop(v);
-  EXPECT_EQ(1u, v);
-  EXPECT_EQ(0u, q.Size());
-  EXPECT_TRUE(q.Empty());
-
-  // Test the FIFO order.
-  for (u32 i = 0; i < 1000; ++i)
-    q.Push(i);
-  EXPECT_EQ(1000u, q.Size());
-  for (u32 i = 0; i < 1000; ++i)
-  {
-    u32 v2 = 0;
-    q.Pop(v2);
-    EXPECT_EQ(i, v2);
-  }
-  EXPECT_TRUE(q.Empty());
-
-  for (u32 i = 0; i < 1000; ++i)
-    q.Push(i);
-  EXPECT_FALSE(q.Empty());
-  q.Clear();
-  EXPECT_TRUE(q.Empty());
-}
-
-TEST(SPSCQueue, MultiThreaded)
-{
-  struct Foo
-  {
-    std::shared_ptr<int> ptr;
-    u32 i;
-  };
-
-  // A shared_ptr held by every element in the queue.
-  auto sptr = std::make_shared<int>(0);
-
-  auto queue_ptr = std::make_unique<Common::WaitableSPSCQueue<Foo>>();
-  auto& q = *queue_ptr;
-
-  constexpr u32 reps = 100000;
-
-  auto inserter = [&] {
-    for (u32 i = 0; i != reps; ++i)
-      q.Push({sptr, i});
-
-    q.WaitForEmpty();
-    EXPECT_EQ(sptr.use_count(), 1);
-    q.Push({sptr, 0});
-    EXPECT_EQ(sptr.use_count(), 2);
-  };
-
-  auto popper = [&] {
-    for (u32 i = 0; i != reps; ++i)
-    {
-      q.WaitForData();
-      EXPECT_EQ(i, q.Front().i);
-      q.Pop();
-    }
-  };
-
-  std::thread popper_thread(popper);
-  std::thread inserter_thread(inserter);
-
-  popper_thread.join();
-  inserter_thread.join();
-
-  queue_ptr.reset();
-  EXPECT_EQ(sptr.use_count(), 1);
-}