pass the testsnew-lf-queue

2 files changed, 22 insertions, 12 deletions

diff --git a/include/salticidae/queue.h b/include/salticidae/queue.h
index 3df1529..0417750 100644
--- a/include/salticidae/queue.h
+++ b/include/salticidae/queue.h
@@ -105,7 +105,7 @@ class MPMCQueue {
     bool _enqueue(U &&e, bool unbounded = true) {
         for (;;)
         {
-            auto t = tail.load(std::memory_order_relaxed);
+            auto t = tail.load(std::memory_order_acquire);
             auto tcnt = t->refcnt.load(std::memory_order_relaxed);
             if (!tcnt) continue;
             if (!t->refcnt.compare_exchange_weak(tcnt, tcnt + 1, std::memory_order_relaxed))
@@ -188,7 +188,7 @@ class MPMCQueue {
     bool try_dequeue(T &e) {
         for (;;)
         {
-            auto h = this->head.load(std::memory_order_relaxed);
+            auto h = this->head.load(std::memory_order_acquire);
             auto hcnt = h->refcnt.load(std::memory_order_relaxed);
             if (!hcnt) continue;
             if (!h->refcnt.compare_exchange_weak(hcnt, hcnt + 1, std::memory_order_relaxed))
@@ -199,19 +199,21 @@ class MPMCQueue {
             if (hh >= tt)
             {
                 if (tt < MPMCQ_SIZE) { blks.release_ref(h); return false; }
-                auto hnext = h->next.load(std::memory_order_relaxed);
+                auto hnext = h->next.load(std::memory_order_acquire);
                 if (hnext == nullptr) { blks.release_ref(h); return false; }
-                if (this->head.compare_exchange_weak(h, hnext, std::memory_order_relaxed))
+                auto h2 = h;
+                if (this->head.compare_exchange_weak(h2, hnext, std::memory_order_acq_rel))
                     this->blks.push(h);
                 blks.release_ref(h);
                 continue;
             }
-            while (!h->avail[hh].load(std::memory_order_acquire))
-                std::this_thread::yield();
             auto hh2 = hh;
             if (h->head.compare_exchange_weak(hh2, hh2 + 1, std::memory_order_relaxed))
             {
+                while (!h->avail[hh].load(std::memory_order_acquire))
+                    std::this_thread::yield();
                 e = std::move(h->elem[hh]);
+                h->avail[hh].store(false, std::memory_order_relaxed);
                 blks.release_ref(h);
                 break;
             }
@@ -241,9 +243,9 @@ struct MPSCQueue: public MPMCQueue<T> {
                 this->blks.push(h);
                 continue;
             }
+            h->head.store(hh + 1, std::memory_order_relaxed);
             while (!h->avail[hh].load(std::memory_order_acquire))
                 std::this_thread::yield();
-            h->head.store(hh + 1, std::memory_order_relaxed);
             e = std::move(h->elem[hh]);
             h->avail[hh].store(false, std::memory_order_relaxed);
             break;
diff --git a/test/test_queue.cpp b/test/test_queue.cpp
index fac3bc7..11699c1 100644
--- a/test/test_queue.cpp
+++ b/test/test_queue.cpp
@@ -8,20 +8,25 @@
 using salticidae::TimerEvent;
 using salticidae::Config;
 
-void test_mpsc(int nproducers, int nops, size_t burst_size) {
+void test_mpsc(int nproducers, int nops, size_t burst_size, bool test_rewind) {
     size_t total = nproducers * nops;
     salticidae::EventContext ec;
     std::atomic<size_t> collected(0);
     using queue_t = salticidae::MPSCQueueEventDriven<int>;
     queue_t q;
     q.set_capacity(65536);
-    q.reg_handler(ec, [&collected, burst_size](queue_t &q) {
+    q.reg_handler(ec, [&collected, burst_size, test_rewind](queue_t &q) {
         size_t cnt = burst_size;
         int x;
         while (q.try_dequeue(x))
         {
-            printf("%d\n", x);
-            collected.fetch_add(1);
+            if (test_rewind && (rand() & 1))
+                q.rewind(x);
+            else
+            {
+                collected.fetch_add(1);
+                printf("%d\n", x);
+            }
             if (!--cnt) return true;
         }
         return false;
@@ -119,13 +124,16 @@ int main(int argc, char **argv) {
     auto opt_nops = Config::OptValInt::create(100000);
     auto opt_mpmc = Config::OptValFlag::create(false);
     auto opt_help = Config::OptValFlag::create(false);
+    auto opt_rewind = Config::OptValFlag::create(false);
     config.add_opt("nproducers", opt_nproducers, Config::SET_VAL);
     config.add_opt("nconsumers", opt_nconsumers, Config::SET_VAL);
     config.add_opt("burst-size", opt_burst_size, Config::SET_VAL);
     config.add_opt("nops", opt_nops, Config::SET_VAL);
     config.add_opt("mpmc", opt_mpmc, Config::SWITCH_ON);
+    config.add_opt("rewind", opt_rewind, Config::SWITCH_ON);
     config.add_opt("help", opt_help, Config::SWITCH_ON, 'h', "show this help info");
     config.parse(argc, argv);
+    srand(time(0));
     if (opt_help->get())
     {
         config.print_help();
@@ -136,7 +144,7 @@ int main(int argc, char **argv) {
     {
         SALTICIDAE_LOG_INFO("testing an MPSC queue...");
         test_mpsc(opt_nproducers->get(), opt_nops->get(),
-                opt_burst_size->get());
+                opt_burst_size->get(), opt_rewind->get());
     }
     else
     {