Common: SPSCQueue cleanups and improvements.

Source/Core/Common/SPSCQueue.h

@@ -6,104 +6,185 @@
 // a simple lockless thread-safe,
 // single producer, single consumer queue
 
-#include <algorithm>
 #include <atomic>
-#include <cstddef>
+#include <cassert>
+#include <memory>
 
-#include "Common/CommonTypes.h"
+#if defined(__APPLE__)
+#include <condition_variable>
+// std::atomic<T>::notify_one is "unavailable: introduced in macOS 11.0"
+// TODO: Remove this when macOS required version is bumped to 11.0+
+#define ATOMIC_NOTIFY_ONE_IS_UNAVAILABLE
+#endif
 
 namespace Common
 {
-template <typename T, bool NeedSize = true>
-class SPSCQueue
+namespace detail
+{
+class AtomicSize
 {
 public:
-  SPSCQueue() : m_size(0) { m_write_ptr = m_read_ptr = new ElementPtr(); }
+  std::size_t Size() const { return m_size.load(std::memory_order_acquire); }
+
+protected:
+  void IncSize() { m_size.fetch_add(1, std::memory_order_release); }
+  void DecSize() { m_size.fetch_sub(1, std::memory_order_release); }
+
+  std::atomic<std::size_t> m_size = 0;
+};
+
+#if defined(ATOMIC_NOTIFY_ONE_IS_UNAVAILABLE)
+class WaitableAtomicSize : public AtomicSize
+{
+public:
+  // To be used by "producer" thread.
+  void WaitForEmpty()
+  {
+    std::unique_lock lg{m_lock};
+    m_cond.wait(lg, [this] { return Size() == 0; });
+  }
+
+  // To be used by "consumer" thread.
+  void WaitForData()
+  {
+    std::unique_lock lg{m_lock};
+    m_cond.wait(lg, [this] { return Size() != 0; });
+  }
+
+protected:
+  void IncSize()
+  {
+    std::unique_lock lg{m_lock};
+    AtomicSize::IncSize();
+    m_cond.notify_one();
+  }
+
+  void DecSize()
+  {
+    std::unique_lock lg{m_lock};
+    AtomicSize::DecSize();
+    m_cond.notify_one();
+  }
+
+private:
+  std::condition_variable m_cond;
+  std::mutex m_lock;
+};
+#else
+class WaitableAtomicSize : public AtomicSize
+{
+public:
+  // To be used by "producer" thread.
+  void WaitForEmpty()
+  {
+    while (auto const old_size = Size())
+      m_size.wait(old_size, std::memory_order_acquire);
+  }
+
+  // To be used by "consumer" thread.
+  void WaitForData() { m_size.wait(0, std::memory_order_acquire); }
+
+protected:
+  void IncSize()
+  {
+    AtomicSize::IncSize();
+    m_size.notify_one();
+  }
+
+  void DecSize()
+  {
+    AtomicSize::DecSize();
+    m_size.notify_one();
+  }
+};
+#endif
+};  // namespace detail
+
+template <typename T, typename SizeBase = detail::AtomicSize>
+class SPSCQueue final : public SizeBase
+{
+public:
+  SPSCQueue() = default;
   ~SPSCQueue()
   {
-    // this will empty out the whole queue
+    Clear();
     delete m_read_ptr;
   }
 
-  u32 Size() const
+  SPSCQueue(const SPSCQueue&) = delete;
+  SPSCQueue& operator=(const SPSCQueue&) = delete;
+
+  bool Empty() const { return SizeBase::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)
   {
-    static_assert(NeedSize, "using Size() on SPSCQueue without NeedSize");
-    return m_size.load();
+    std::construct_at(&m_write_ptr->value.data, std::forward<Args>(args)...);
+
+    Node* const new_ptr = new Node;
+    m_write_ptr->next = new_ptr;
+    m_write_ptr = new_ptr;
+
+    SizeBase::IncSize();
   }
 
-  bool Empty() const { return !m_read_ptr->next.load(); }
-  T& Front() const { return m_read_ptr->current; }
-  template <typename Arg>
-  void Push(Arg&& t)
-  {
-    // create the element, add it to the queue
-    m_write_ptr->current = std::forward<Arg>(t);
-    // set the next pointer to a new element ptr
-    // then advance the write pointer
-    ElementPtr* new_ptr = new ElementPtr();
-    m_write_ptr->next.store(new_ptr, std::memory_order_release);
-    m_write_ptr = new_ptr;
-    if (NeedSize)
-      m_size++;
-  }
+  // The following are only safe from the "consumer thread":
+  T& Front() { return m_read_ptr->value.data; }
+  const T& Front() const { return m_read_ptr->value.data; }
 
   void Pop()
   {
-    if (NeedSize)
-      m_size--;
-    ElementPtr* tmpptr = m_read_ptr;
-    // advance the read pointer
-    m_read_ptr = tmpptr->next.load();
-    // set the next element to nullptr to stop the recursive deletion
-    tmpptr->next.store(nullptr);
-    delete tmpptr;  // this also deletes the element
+    assert(!Empty());
+
+    std::destroy_at(&Front());
+
+    Node* const old_node = m_read_ptr;
+    m_read_ptr = old_node->next;
+    delete old_node;
+
+    SizeBase::DecSize();
   }
 
-  bool Pop(T& t)
+  bool Pop(T& result)
   {
     if (Empty())
       return false;
 
-    if (NeedSize)
-      m_size--;
-
-    ElementPtr* tmpptr = m_read_ptr;
-    m_read_ptr = tmpptr->next.load(std::memory_order_acquire);
-    t = std::move(tmpptr->current);
-    tmpptr->next.store(nullptr);
-    delete tmpptr;
+    result = std::move(Front());
+    Pop();
     return true;
   }
 
-  // not thread-safe
   void Clear()
   {
-    m_size.store(0);
-    delete m_read_ptr;
-    m_write_ptr = m_read_ptr = new ElementPtr();
+    while (!Empty())
+      Pop();
   }
 
 private:
-  // stores a pointer to element
-  // and a pointer to the next ElementPtr
-  class ElementPtr
+  struct Node
   {
-  public:
-    ElementPtr() : next(nullptr) {}
-    ~ElementPtr()
+    // union allows value construction to be deferred until Push.
+    union Value
     {
-      ElementPtr* next_ptr = next.load();
+      T data;
+      Value() {}
+      ~Value() {}
+    } value;
 
-      if (next_ptr)
-        delete next_ptr;
-    }
-
-    T current{};
-    std::atomic<ElementPtr*> next;
+    Node* next;
   };
 
-  ElementPtr* m_write_ptr;
-  ElementPtr* m_read_ptr;
-  std::atomic<u32> m_size;
+  Node* m_write_ptr = new Node;
+  Node* m_read_ptr = m_write_ptr;
 };
+
+template <typename T>
+using WaitableSPSCQueue = SPSCQueue<T, detail::WaitableAtomicSize>;
+
 }  // namespace Common
+
+#undef ATOMIC_NOTIFY_ONE_IS_UNAVAILABLE

Source/Core/Core/CoreTiming.h

@@ -184,7 +184,7 @@ private:
   std::vector<Event> m_event_queue;
   u64 m_event_fifo_id = 0;
   std::mutex m_ts_write_lock;
-  Common::SPSCQueue<Event, false> m_ts_queue;
+  Common::SPSCQueue<Event> m_ts_queue;
 
   float m_last_oc_factor = 0.0f;
 

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

@@ -135,8 +135,8 @@ private:
   Common::Event m_result_queue_expanded;                    // Is set by DVD thread
   Common::Flag m_dvd_thread_exiting = Common::Flag(false);  // Is set by CPU thread
 
-  Common::SPSCQueue<ReadRequest, false> m_request_queue;
-  Common::SPSCQueue<ReadResult, false> m_result_queue;
+  Common::SPSCQueue<ReadRequest> m_request_queue;
+  Common::SPSCQueue<ReadResult> m_result_queue;
   std::map<u64, ReadResult> m_result_map;
 
   std::unique_ptr<DiscIO::Volume> m_disc;

Source/Core/Core/NetPlayClient.h

@@ -196,7 +196,7 @@ protected:
     std::recursive_mutex async_queue_write;
   } m_crit;
 
-  Common::SPSCQueue<AsyncQueueEntry, false> m_async_queue;
+  Common::SPSCQueue<AsyncQueueEntry> m_async_queue;
 
   std::array<Common::SPSCQueue<GCPadStatus>, 4> m_pad_buffer;
   std::array<Common::SPSCQueue<WiimoteEmu::SerializedWiimoteState>, 4> m_wiimote_buffer;

Source/Core/Core/NetPlayServer.h

@@ -197,8 +197,8 @@ private:
     std::recursive_mutex chunked_data_queue_write;
   } m_crit;
 
-  Common::SPSCQueue<AsyncQueueEntry, false> m_async_queue;
-  Common::SPSCQueue<ChunkedDataQueueEntry, false> m_chunked_data_queue;
+  Common::SPSCQueue<AsyncQueueEntry> m_async_queue;
+  Common::SPSCQueue<ChunkedDataQueueEntry> m_chunked_data_queue;
 
   SyncIdentifier m_selected_game_identifier;
   std::string m_selected_game_name;

Source/Core/VideoCommon/PerformanceTracker.h

@@ -59,7 +59,7 @@ private:
 
   // Push'd from Count()
   //  and Pop'd from UpdateStats()
-  Common::SPSCQueue<DT, false> m_raw_dts;
+  Common::SPSCQueue<DT> 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

@@ -2,8 +2,11 @@
 // 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)
@@ -44,21 +47,36 @@ TEST(SPSCQueue, Simple)
 
 TEST(SPSCQueue, MultiThreaded)
 {
-  Common::SPSCQueue<u32> q;
-
-  auto inserter = [&q]() {
-    for (u32 i = 0; i < 100000; ++i)
-      q.Push(i);
+  struct Foo
+  {
+    std::shared_ptr<int> ptr;
+    u32 i;
   };
 
-  auto popper = [&q]() {
-    for (u32 i = 0; i < 100000; ++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)
     {
-      while (q.Empty())
-        ;
-      u32 v;
-      q.Pop(v);
-      EXPECT_EQ(i, v);
+      q.WaitForData();
+      EXPECT_EQ(i, q.Front().i);
+      q.Pop();
     }
   };
 
@@ -67,4 +85,7 @@ TEST(SPSCQueue, MultiThreaded)
 
   popper_thread.join();
   inserter_thread.join();
+
+  queue_ptr.reset();
+  EXPECT_EQ(sptr.use_count(), 1);
 }