MPSCQueue: add queue size

Author: TheMostDiligent

Common/interface/MPSCQueue.hpp

@@ -87,6 +87,10 @@ class MPSCQueue
         Node* node = AllocateNode(std::move(value));
         node->pNext.store(nullptr, std::memory_order_relaxed);
 
+        // Increment size before pushing to ensure that the element is not
+        // dequeued before m_Size reflects the new element.
+        m_Size.fetch_add(1, std::memory_order_relaxed);
+
         // Standard Lock-Free Enqueue
         Node* prev = m_Tail.exchange(node, std::memory_order_acq_rel);
         prev->pNext.store(node, std::memory_order_release);
@@ -110,18 +114,39 @@ class MPSCQueue
         m_Head = next;
         result = std::move(next->Value);
 
+        m_Size.fetch_sub(1, std::memory_order_relaxed);
+
         // Current head became the old dummy node, recycle it
         RecycleNode(head);
 
         return true;
     }
 
     /// Checks if the queue is empty.
+
+    /// The method is safe to call concurrently with producers, but must only be
+    /// called by the consumer thread.
+    ///
+    /// \note In the presence of concurrent producers, this method may temporarily
+    ///       report the queue as empty while an enqueue is still in progress and
+    ///       the new node has not yet been linked into the queue.
     bool IsEmpty() const
     {
         return m_Head->pNext.load(std::memory_order_acquire) == nullptr;
     }
 
+    /// Returns the approximate number of items in the queue.
+
+    /// \note This value is not exact in the presence of concurrent producers and
+    ///       consumer. It may temporarily overestimate the number of items due to
+    ///       in-flight enqueue operations, and may disagree with IsEmpty() or a
+    ///       subsequent Dequeue() call. It must not be used to predict whether
+    ///       Dequeue() will succeed.
+    size_t Size() const
+    {
+        return m_Size.load(std::memory_order_relaxed);
+    }
+
 private:
     struct Node
     {
@@ -196,6 +221,8 @@ class MPSCQueue
     // Consumer Data (Hot)
     alignas(CacheLineSize) Node* m_Head = nullptr;
 
+    std::atomic<size_t> m_Size{0};
+
     // Free List (Shared - Moderate Contention)
     // The lock protects the "Pop" side from other Producers
     std::mutex         m_FreeHeadMtx;

Graphics/GraphicsTools/include/GPUUploadManagerImpl.hpp

@@ -138,7 +138,7 @@ class GPUUploadManagerImpl final : public ObjectBase<IGPUUploadManager>
         Uint32 GetSize() const { return m_Size; }
 
         // Returns the number of pending operations.
-        size_t GetNumPendingOps() const { return m_NumPendingOps.load(std::memory_order_acquire); }
+        size_t GetNumPendingOps() const { return m_PendingOps.Size(); }
 
 #ifdef DILIGENT_DEBUG
         // Returns the number of active writers. This is used for testing and debugging purposes.
@@ -171,8 +171,7 @@ class GPUUploadManagerImpl final : public ObjectBase<IGPUUploadManager>
         static constexpr Uint32 WRITER_MASK = ~SEALED_BIT; // low 31 bits
         std::atomic<Uint32>     m_State{0};
 
-        std::atomic<size_t> m_NumPendingOps{0};
-        std::atomic<bool>   m_Enqueued{false};
+        std::atomic<bool> m_Enqueued{false};
 
         Uint64 m_FenceValue = 0;
 

Graphics/GraphicsTools/src/GPUUploadManagerImpl.cpp

@@ -251,7 +251,6 @@ bool GPUUploadManagerImpl::Page::ScheduleBufferUpdate(const ScheduleBufferUpdate
     Op.DstOffset = UpdateInfo.DstOffset;
     Op.NumBytes  = UpdateInfo.NumBytes;
     m_PendingOps.Enqueue(std::move(Op));
-    m_NumPendingOps.fetch_add(1, std::memory_order_acq_rel);
 
     return true;
 }
@@ -288,7 +287,6 @@ void GPUUploadManagerImpl::Page::ExecutePendingOps(IDeviceContext* pContext, Uin
             }
         }
     }
-    m_NumPendingOps.store(0);
     m_FenceValue = FenceValue;
 }
 
@@ -300,7 +298,6 @@ void GPUUploadManagerImpl::Page::Reset(IDeviceContext* pContext)
     m_Offset.store(0);
     // Keep the page sealed until we make it current to prevent any accidental writes.
     m_State.store(SEALED_BIT);
-    m_NumPendingOps.store(0);
     m_Enqueued.store(false);
     m_FenceValue = 0;
 

Tests/DiligentCoreTest/src/Common/MPSCQueueTest.cpp

@@ -46,30 +46,39 @@ TEST(Common_MPSCQueue, EnqueueDequeue)
     Queue.Enqueue(84);
     Queue.Enqueue(126);
     EXPECT_FALSE(Queue.IsEmpty());
+    EXPECT_EQ(Queue.Size(), size_t{3});
 
     int Value = 0;
     EXPECT_TRUE(Queue.Dequeue(Value));
     EXPECT_EQ(Value, 42);
+    EXPECT_EQ(Queue.Size(), size_t{2});
     EXPECT_TRUE(Queue.Dequeue(Value));
     EXPECT_EQ(Value, 84);
+    EXPECT_EQ(Queue.Size(), size_t{1});
     EXPECT_TRUE(Queue.Dequeue(Value));
     EXPECT_EQ(Value, 126);
+    EXPECT_EQ(Queue.Size(), size_t{0});
 
     EXPECT_TRUE(Queue.IsEmpty());
     EXPECT_FALSE(Queue.Dequeue(Value));
+    EXPECT_EQ(Queue.Size(), size_t{0});
 
     Queue.Enqueue(168);
     Queue.Enqueue(210);
+    EXPECT_EQ(Queue.Size(), size_t{2});
     EXPECT_TRUE(Queue.Dequeue(Value));
     EXPECT_EQ(Value, 168);
     EXPECT_TRUE(Queue.Dequeue(Value));
     EXPECT_EQ(Value, 210);
 
+    EXPECT_EQ(Queue.Size(), size_t{0});
     EXPECT_TRUE(Queue.IsEmpty());
     EXPECT_FALSE(Queue.Dequeue(Value));
+    EXPECT_EQ(Queue.Size(), size_t{0});
 
     Queue.Enqueue(252);
     Queue.Enqueue(294);
+    EXPECT_EQ(Queue.Size(), size_t{2});
 }
 
 TEST(Common_MPSCQueue, EnqueueDequeueMoveOnly)
@@ -79,6 +88,7 @@ TEST(Common_MPSCQueue, EnqueueDequeueMoveOnly)
     Queue.Enqueue(std::make_unique<int>(42));
     Queue.Enqueue(std::make_unique<int>(84));
     Queue.Enqueue(std::make_unique<int>(126));
+    EXPECT_EQ(Queue.Size(), size_t{3});
 
     std::unique_ptr<int> Value;
     EXPECT_TRUE(Queue.Dequeue(Value));
@@ -87,20 +97,26 @@ TEST(Common_MPSCQueue, EnqueueDequeueMoveOnly)
     EXPECT_EQ(*Value, 84);
     EXPECT_TRUE(Queue.Dequeue(Value));
     EXPECT_EQ(*Value, 126);
+    EXPECT_EQ(Queue.Size(), size_t{0});
 
     EXPECT_FALSE(Queue.Dequeue(Value));
+    EXPECT_EQ(Queue.Size(), size_t{0});
 
     Queue.Enqueue(std::make_unique<int>(168));
     Queue.Enqueue(std::make_unique<int>(210));
+    EXPECT_EQ(Queue.Size(), size_t{2});
     EXPECT_TRUE(Queue.Dequeue(Value));
     EXPECT_EQ(*Value, 168);
     EXPECT_TRUE(Queue.Dequeue(Value));
     EXPECT_EQ(*Value, 210);
+    EXPECT_EQ(Queue.Size(), size_t{0});
 
     EXPECT_FALSE(Queue.Dequeue(Value));
+    EXPECT_EQ(Queue.Size(), size_t{0});
 
     Queue.Enqueue(std::make_unique<int>(252));
     Queue.Enqueue(std::make_unique<int>(294));
+    EXPECT_EQ(Queue.Size(), size_t{2});
 }
 
 
@@ -157,6 +173,7 @@ TEST(Common_MPSCQueue, EnqueueDequeueParallel)
     Consumer.join();
 
     EXPECT_TRUE(Queue.IsEmpty());
+    EXPECT_EQ(Queue.Size(), size_t{0});
 
     EXPECT_EQ(ProducedData.size(), NumProducers * NumItemsPerProducer);
     std::vector<Uint32> CurrProducedValue(NumProducers, 0);