Atomic queue explicit atomic operations

src/struct/atomic_queue.h

@@ -1,6 +1,7 @@
 #ifndef SRC_STRUCT_ATOMIC_QUEUE
 #define SRC_STRUCT_ATOMIC_QUEUE
 
+#include <cstddef>
 #include <cstdint>
 #include <atomic>
 #include <memory>
@@ -13,7 +14,7 @@ template <typename T>
 class AtomicQueue
 {
 public:
-    AtomicQueue(uint16_t size)
+    AtomicQueue(size_t size)
         : _size(size), _data(new unique_ptr<T>[size]), _first(0), _last(0), _count(0)
     {
     }
@@ -25,19 +26,19 @@ public:
 
     bool pushDiscard(unique_ptr<T> obj)
     {
-        if (_count == _size)
+        if (_count.load(std::memory_order_acquire) == _size)
             return false;
 
         _first = (_first + 1) % _size;
         _data[_first] = std::move(obj);
-        ++_count;
+        _count.fetch_add(1, std::memory_order_release);
         _lock.notify_one();
         return true;
     }
 
     bool pushReplace(unique_ptr<T> obj)
     {
-        if (_count == _size)
+        if (_count.load(std::memory_order_acquire) == _size)
         {
             _data[_first] = std::move(obj);
             return false;
@@ -45,49 +46,49 @@ public:
 
         _first = (_first + 1) % _size;
         _data[_first] = std::move(obj);
-        ++_count;
+        _count.fetch_add(1, std::memory_order_release);
         _lock.notify_one();
         return true;
     }
 
     const T *peek()
     {
-        if (_count == 0)
+        if (_count.load(std::memory_order_acquire) == 0)
             return nullptr;
         return _data[(_last + 1) % _size].get();
     }
 
     unique_ptr<T> pop()
     {
-        if (_count == 0)
+        if (_count.load(std::memory_order_acquire) == 0)
             return nullptr;
 
         _last = (_last + 1) % _size;
         auto item = std::move(_data[_last]);
-        --_count;
+        _count.fetch_sub(1, std::memory_order_release);
         return item;
     }
 
-    bool has(uint8_t count)
+    bool has(size_t count)
     {
-        return _count >= count;
+        return _count.load(std::memory_order_acquire) >= count;
     }
 
-    bool wait(atomic<bool> &waitFlag, uint8_t count = 0)
+    bool wait(atomic<bool> &waitFlag, size_t count = 0)
     {
         unique_lock<std::mutex> lock(_mtx);
 
         _lock.wait(lock, [&]
-                   { return _count > count || !waitFlag.load(); });
-        return waitFlag.load();
+                   { return _count.load(std::memory_order_acquire) > count || !waitFlag.load(std::memory_order_acquire); });
+        return waitFlag.load(std::memory_order_acquire);
     }
 
-    bool waitFor(atomic<bool> &waitFlag, uint32_t timeoutMs, uint8_t count = 0)
+    bool waitFor(atomic<bool> &waitFlag, uint32_t timeoutMs, size_t count = 0)
     {
         unique_lock<std::mutex> lock(_mtx);
         _lock.wait_for(lock, std::chrono::milliseconds(timeoutMs), [&]
-                       { return _count > count || !waitFlag.load(); });
-        return waitFlag.load();
+                       { return _count.load(std::memory_order_acquire) > count || !waitFlag.load(std::memory_order_acquire); });
+        return waitFlag.load(std::memory_order_acquire);
     }
 
     void clear()
@@ -95,7 +96,7 @@ public:
         _data = std::make_unique<std::unique_ptr<T>[]>(_size);
         _first = 0;
         _last = 0;
-        _count = 0;
+        _count.store(0, std::memory_order_release);
     }
 
     void notify()
@@ -103,14 +104,14 @@ public:
         _lock.notify_all();
     }
 
-    uint16_t count() { return _count; }
+    size_t count() { return _count.load(std::memory_order_acquire); }
 
 private:
-    uint16_t _size;
+    size_t _size;
     unique_ptr<unique_ptr<T>[]> _data;
-    uint16_t _first;
-    uint16_t _last;
-    atomic<uint16_t> _count;
+    size_t _first;
+    size_t _last;
+    atomic<size_t> _count;
     mutex _mtx;
     condition_variable _lock;
 };