From 8e104ded0ff1ad39d851f23d4f613bf85137aecc Mon Sep 17 00:00:00 2001
From: mutouyun <orz@orzz.org>
Date: Sat, 5 Jan 2019 16:09:56 +0800
Subject: [PATCH] add prod_cons<relat::multi, relat::multi, trans::unicast>

---
 include/circ_elems_array.h | 38 +++++++++++++++++++++++++++++++++++---
 test/test_circ.cpp         | 16 +++++++++++++---
 2 files changed, 48 insertions(+), 6 deletions(-)

diff --git a/include/circ_elems_array.h b/include/circ_elems_array.h
index be45ee5..9884c05 100644
--- a/include/circ_elems_array.h
+++ b/include/circ_elems_array.h
@@ -73,7 +73,7 @@ struct prod_cons<relat::single, relat::single, trans::unicast> {
     bool push(E* /*elems*/, F&& f, detail::elem_t<S>* elem_start) {
         auto cur_wt = detail::index_of(wt_.load(std::memory_order_acquire));
         if (cur_wt == detail::index_of(rd_.load(std::memory_order_relaxed) - 1)) {
-            return false;
+            return false; // full
         }
         std::forward<F>(f)(elem_start + cur_wt);
         wt_.fetch_add(1, std::memory_order_release);
@@ -84,7 +84,7 @@ struct prod_cons<relat::single, relat::single, trans::unicast> {
     bool pop(E* /*elems*/, detail::u2_t& /*cur*/, F&& f, detail::elem_t<S>* elem_start) noexcept {
         auto cur_rd = detail::index_of(rd_.load(std::memory_order_acquire));
         if (cur_rd == detail::index_of(wt_.load(std::memory_order_relaxed))) {
-            return false;
+            return false; // empty
         }
         std::forward<F>(f)(elem_start + cur_rd);
         rd_.fetch_add(1, std::memory_order_release);
@@ -103,7 +103,7 @@ struct prod_cons<relat::single, relat::multi, trans::unicast>
             auto cur_rd = rd_.load(std::memory_order_acquire);
             if (detail::index_of(cur_rd) ==
                 detail::index_of(wt_.load(std::memory_order_relaxed))) {
-                return false;
+                return false; // empty
             }
             std::memcpy(buff, elem_start + detail::index_of(cur_rd), sizeof(buff));
             if (rd_.compare_exchange_weak(cur_rd, cur_rd + 1, std::memory_order_release)) {
@@ -115,6 +115,38 @@ struct prod_cons<relat::single, relat::multi, trans::unicast>
     }
 };
 
+template <>
+struct prod_cons<relat::multi, relat::multi, trans::unicast>
+     : prod_cons<relat::single, relat::multi, trans::unicast> {
+
+    std::atomic<detail::u2_t> ct_ { 0 }; // commit index
+
+    template <typename E, typename F, std::size_t S>
+    bool push(E* /*elems*/, F&& f, detail::elem_t<S>* elem_start) {
+        detail::u2_t cur_ct, nxt_ct;
+        while (1) {
+            cur_ct = ct_.load(std::memory_order_acquire);
+            if (detail::index_of(nxt_ct = cur_ct + 1) ==
+                detail::index_of(rd_.load(std::memory_order_relaxed))) {
+                return false; // full
+            }
+            if (ct_.compare_exchange_weak(cur_ct, nxt_ct, std::memory_order_relaxed)) {
+                break;
+            }
+            std::this_thread::yield();
+        }
+        std::forward<F>(f)(elem_start + detail::index_of(cur_ct));
+        while (1) {
+            auto exp_wt = cur_ct;
+            if (wt_.compare_exchange_weak(exp_wt, nxt_ct, std::memory_order_release)) {
+                break;
+            }
+            std::this_thread::yield();
+        }
+        return true;
+    }
+};
+
 template <>
 struct prod_cons<relat::single, relat::multi, trans::broadcast> {
     std::atomic<detail::u2_t> wt_ { 0 }; // write index
diff --git a/test/test_circ.cpp b/test/test_circ.cpp
index 4ce7ca0..1d0d946 100644
--- a/test/test_circ.cpp
+++ b/test/test_circ.cpp
@@ -64,9 +64,8 @@ struct test_verify<cq_t> {
     }
 };
 
-template <>
-struct test_verify<ipc::circ::prod_cons<
-        ipc::circ::relat::single,
+template <ipc::circ::relat Rp>
+struct test_verify<ipc::circ::prod_cons<Rp,
         ipc::circ::relat::multi,
         ipc::circ::trans::unicast>
     > : test_verify<cq_t> {
@@ -344,6 +343,17 @@ void Unit::test_prod_cons_1v3() {
     benchmark_prod_cons<1, 3, LoopCount, decltype(el_arr_smn)::policy_t>(&el_arr_smn);
     benchmark_prod_cons<1, 3, LoopCount, void>(&el_arr_smn);
 
+    ipc::circ::elems_array<
+        sizeof(msg_t),
+        ipc::circ::prod_cons<ipc::circ::relat::multi,
+                             ipc::circ::relat::multi,
+                             ipc::circ::trans::unicast>
+    > el_arr_mmn;
+    benchmark_prod_cons<1, 3, LoopCount, decltype(el_arr_mmn)::policy_t>(&el_arr_mmn);
+    benchmark_prod_cons<1, 3, LoopCount, void>(&el_arr_mmn);
+    benchmark_prod_cons<3, 3, LoopCount, decltype(el_arr_mmn)::policy_t>(&el_arr_mmn);
+    benchmark_prod_cons<3, 3, LoopCount, void>(&el_arr_mmn);
+
     ipc::circ::elems_array<
         sizeof(msg_t),
         ipc::circ::prod_cons<ipc::circ::relat::single,