Adjust MPMCRingQueue memory layout for better performance

Author: Coldwings

common/lockfree_queue.h

@@ -169,8 +169,12 @@ class LockfreeMPMCRingQueue : public LockfreeRingQueueBase<T, N> {
     using Base::idx;
     using Base::tail;
 
-    std::atomic<uint64_t> marks[Base::capacity]{};
-    T slots[Base::capacity];
+    struct alignas(Base::CACHELINE_SIZE) packedslot {
+        T data;
+        std::atomic<uint64_t> mark{0};
+    };
+    
+    packedslot slots[Base::capacity];
 
     uint64_t this_turn_write(const uint64_t x) const {
         return (Base::turn(x) << 1) + 1;
@@ -191,8 +195,9 @@ class LockfreeMPMCRingQueue : public LockfreeRingQueueBase<T, N> {
     bool push(const T& x) {
         auto t = tail.load(std::memory_order_acquire);
         for (;;) {
-            auto& slot = slots[idx(t)];
-            auto& mark = marks[idx(t)];
+            auto& ps = slots[idx(t)];
+            auto& slot = ps.data;
+            auto& mark = ps.mark;
             if (mark.load(std::memory_order_acquire) == last_turn_read(t)) {
                 if (tail.compare_exchange_strong(t, t + 1)) {
                     slot = x;
@@ -213,8 +218,9 @@ class LockfreeMPMCRingQueue : public LockfreeRingQueueBase<T, N> {
     bool pop(T& x) {
         auto h = head.load(std::memory_order_acquire);
         for (;;) {
-            auto& slot = slots[idx(h)];
-            auto& mark = marks[idx(h)];
+            auto& ps = slots[idx(h)];
+            auto& slot = ps.data;
+            auto& mark = ps.mark;
             if (mark.load(std::memory_order_acquire) == this_turn_write(h)) {
                 if (head.compare_exchange_strong(h, h + 1)) {
                     x = slot;
@@ -237,8 +243,9 @@ class LockfreeMPMCRingQueue : public LockfreeRingQueueBase<T, N> {
         static_assert(std::is_base_of<PauseBase, Pause>::value,
                       "Pause should be derived by PauseBase");
         auto const t = tail.fetch_add(1);
-        auto& slot = slots[idx(t)];
-        auto& mark = marks[idx(t)];
+        auto& ps = slots[idx(t)];
+        auto& slot = ps.data;
+        auto& mark = ps.mark;
         while (mark.load(std::memory_order_acquire) != last_turn_read(t))
             Pause::pause();
         slot = x;
@@ -250,8 +257,9 @@ class LockfreeMPMCRingQueue : public LockfreeRingQueueBase<T, N> {
         static_assert(std::is_base_of<PauseBase, Pause>::value,
                       "Pause should be derived by PauseBase");
         auto const h = head.fetch_add(1);
-        auto& slot = slots[idx(h)];
-        auto& mark = marks[idx(h)];
+        auto& ps = slots[idx(h)];
+        auto& slot = ps.data;
+        auto& mark = ps.mark;
         while (mark.load(std::memory_order_acquire) != this_turn_write(h))
             Pause::pause();
         T ret = slot;

common/test/test_lockfree.cpp

@@ -34,6 +34,8 @@ limitations under the License.
 #include <thread>
 #include <vector>
 #include <photon/common/alog.h>
+#include <gtest/gtest.h>
+#include <gflags/gflags.h>
 #include "../../test/ci-tools.h"
 
 static constexpr size_t sender_num = 4;
@@ -86,7 +88,7 @@ int test_queue(const char *name, QType &queue) {
     auto begin = std::chrono::steady_clock::now();
     for (size_t i = 0; i < receiver_num; i++) {
         receivers.emplace_back([i, &queue] {
-            photon::set_cpu_affinity(i);
+            photon::set_cpu_affinity(i * 2);
             std::chrono::nanoseconds rspent(std::chrono::nanoseconds(0));
             for (size_t x = 0; x < items_num / receiver_num; x++) {
                 int t;
@@ -109,7 +111,7 @@ int test_queue(const char *name, QType &queue) {
     }
     for (size_t i = 0; i < sender_num; i++) {
         senders.emplace_back([i, &queue] {
-            photon::set_cpu_affinity(i);
+            photon::set_cpu_affinity(i * 2 + 1);
             std::chrono::nanoseconds wspent{std::chrono::nanoseconds(0)};
             for (size_t x = 0; x < items_num / sender_num; x++) {
                 auto tm = std::chrono::high_resolution_clock::now();
@@ -135,9 +137,8 @@ int test_queue(const char *name, QType &queue) {
     LOG_DEBUG("` ` p ` c, ` items, Spent ` us", name, sender_num, receiver_num, items_num,
            std::chrono::duration_cast<std::chrono::microseconds>(end - begin).count());
     for (size_t i = 0; i < items_num / sender_num; i++) {
-        if (sc[i] != rc[i] || sc[i] != sender_num) {
-            LOG_DEBUG("MISMATCH ` ` `", i, sc[i].load(), rc[i].load());
-        }
+        EXPECT_EQ(sc[i], rc[i]);
+        EXPECT_EQ(sc[i].load(), (int)sender_num);
         sc[i] = 0;
         rc[i] = 0;
     }
@@ -153,7 +154,7 @@ int test_queue_batch(const char *name, QType &queue) {
     auto begin = std::chrono::steady_clock::now();
     for (size_t i = 0; i < receiver_num; i++) {
         receivers.emplace_back([i, &queue] {
-            photon::set_cpu_affinity(i);
+            photon::set_cpu_affinity(i * 2);
             int buffer[32];
             size_t size;
             int amount = items_num / receiver_num;
@@ -181,7 +182,7 @@ int test_queue_batch(const char *name, QType &queue) {
     }
     for (size_t i = 0; i < sender_num; i++) {
         senders.emplace_back([i, &queue] {
-            photon::set_cpu_affinity(i);
+            photon::set_cpu_affinity(i * 2 + 1);
             std::vector<int> vec;
             vec.resize(items_num / sender_num);
             for (size_t x = 0; x < items_num / sender_num; x++) {
@@ -215,26 +216,33 @@ int test_queue_batch(const char *name, QType &queue) {
     LOG_DEBUG("` ` p ` c, ` items, Spent ` us", name, sender_num, receiver_num, items_num,
            std::chrono::duration_cast<std::chrono::microseconds>(end - begin).count());
     for (size_t i = 0; i < items_num / sender_num; i++) {
-        if (sc[i] != rc[i] || sc[i] != sender_num) {
-            LOG_DEBUG("MISMATCH ` ` `", i, sc[i].load(), rc[i].load());
-        }
+        EXPECT_EQ(sc[i], rc[i]);
+        EXPECT_EQ(sc[i].load(), (int)sender_num);
         sc[i] = 0;
         rc[i] = 0;
     }
     std::this_thread::sleep_for(std::chrono::seconds(1));
     return 0;
 }
 
-int main() {
+TEST(lockfree_queue, test_queue) {
 #ifdef TESTING_ENABLE_BOOST
     test_queue<NoLock, NoLock>("BoostQueue", bqueue);
 #endif
     test_queue<NoLock, NoLock>("PhotonLockfreeMPMCQueue", lqueue);
     test_queue<NoLock, NoLock>("PhotonLockfreeBatchMPMCQueue", lbqueue);
-    test_queue_batch<NoLock, NoLock>("PhotonLockfreeBatchMPMCQueue+Batch", lbqueue);
+    test_queue_batch<NoLock, NoLock>("PhotonLockfreeBatchMPMCQueue+Batch",
+                                     lbqueue);
 #ifdef TESTING_ENABLE_BOOST
     test_queue<WithLock, WithLock>("BoostSPSCQueue", squeue);
 #endif
     test_queue<WithLock, WithLock>("PhotonSPSCQueue", cqueue);
     test_queue_batch<WithLock, WithLock>("PhotonSPSCQueue+Batch", cqueue);
 }
+
+int main(int argc, char **arg) {
+    if (!photon::is_using_default_engine()) return 0;
+    ::testing::InitGoogleTest(&argc, arg);
+    gflags::ParseCommandLineFlags(&argc, &arg, true);
+    return RUN_ALL_TESTS();
+}