From 881b060d1f64df3027861ba39e8e92de78390311 Mon Sep 17 00:00:00 2001
From: mutouyun <mark.lonr@tom.com>
Date: Wed, 23 Jan 2019 11:44:26 +0800
Subject: [PATCH] refactor

---
 include/elem_circ.h |   5 +-
 include/elem_link.h |   5 ++
 include/ipc.h       |   8 +-
 include/queue.h     | 184 +++++++++++++++++++++++++++++---------------
 include/rw_lock.h   |   4 +-
 src/ipc.cpp         |  68 ++++++++--------
 6 files changed, 171 insertions(+), 103 deletions(-)

diff --git a/include/elem_circ.h b/include/elem_circ.h
index 659a323..cfca3ab 100644
--- a/include/elem_circ.h
+++ b/include/elem_circ.h
@@ -4,6 +4,7 @@
 #include <thread>
 #include <cstring>
 #include <utility>
+#include <type_traits>
 
 #include "def.h"
 #include "rw_lock.h"
@@ -299,7 +300,7 @@ public:
     using base_t::cursor;
 
     template <typename F, typename... P>
-    bool push(F&& f, P&&...) noexcept {
+    bool push(F&& f) noexcept {
         return base_t::push(this, std::forward<F>(f), block_);
     }
 
@@ -312,6 +313,8 @@ public:
 
 template <relat Rp, relat Rc, trans Ts>
 struct prod_cons {
+    using is_fixed = std::true_type;
+
     template <std::size_t DataSize>
     using elems_t = elem_array<DataSize, prod_cons_circ<Rp, Rc, Ts>>;
 };
diff --git a/include/elem_link.h b/include/elem_link.h
index 0f5703c..9148d29 100644
--- a/include/elem_link.h
+++ b/include/elem_link.h
@@ -10,4 +10,9 @@
 
 namespace ipc {
 
+namespace link {
+
+
+
+} // namespace link
 } // namespace ipc
diff --git a/include/ipc.h b/include/ipc.h
index 6f3bc8d..e5f1f30 100644
--- a/include/ipc.h
+++ b/include/ipc.h
@@ -7,14 +7,14 @@
 #include "def.h"
 #include "buffer.h"
 #include "shm.h"
-#include "queue.h"
+#include "elem_circ.h"
 
 namespace ipc {
 
 using handle_t = void*;
 using buff_t   = buffer;
 
-template <template <typename...> class Queue, typename Policy>
+template <typename Policy>
 struct IPC_EXPORT channel_detail {
     static handle_t connect   (char const * name);
     static void     disconnect(handle_t h);
@@ -138,7 +138,7 @@ public:
  * (one producer/server/sender to multi consumers/clients/receivers)
 */
 using route = channel_impl<channel_detail<
-    ipc::queue, ipc::circ::prod_cons<relat::single, relat::multi, trans::broadcast>
+    ipc::circ::prod_cons<relat::single, relat::multi, trans::broadcast>
 >>;
 
 /*
@@ -150,7 +150,7 @@ using route = channel_impl<channel_detail<
 */
 
 using channel = channel_impl<channel_detail<
-    ipc::queue, ipc::circ::prod_cons<relat::multi, relat::multi, trans::broadcast>
+    ipc::circ::prod_cons<relat::multi, relat::multi, trans::broadcast>
 >>;
 
 } // namespace ipc
diff --git a/include/queue.h b/include/queue.h
index 9b05d5b..4f04397 100644
--- a/include/queue.h
+++ b/include/queue.h
@@ -17,54 +17,101 @@
 #include "platform/waiter.h"
 
 namespace ipc {
+namespace detail {
 
-template <typename T,
-          typename Policy = ipc::circ::prod_cons<relat::single, relat::multi, trans::broadcast>>
-class queue {
-public:
-    using elems_t  = typename Policy::template elems_t<sizeof(T)>;
-    using policy_t = typename elems_t::policy_t;
-
-private:
-    elems_t* elems_ = nullptr;
+class queue_waiter {
+protected:
     ipc::detail::waiter_impl waiter_, cc_waiter_;
-    decltype(std::declval<elems_t>().cursor()) cursor_ = 0;
     std::atomic_bool connected_ { false };
 
-public:
-    queue() = default;
-
-    explicit queue(elems_t* els, char const * name = nullptr) : queue() {
-        attach(els, name);
+    template <typename Elems>
+    void open(Elems* elems, char const * name) {
+        waiter_.attach(&(elems->waiter()));
+        waiter_.open((std::string{ "__IPC_WAITER__" } + name).c_str());
+        cc_waiter_.attach(&(elems->conn_waiter()));
+        cc_waiter_.open((std::string{ "__IPC_CC_WAITER__" } + name).c_str());
     }
 
-    queue(const queue&) = delete;
-    queue& operator=(const queue&) = delete;
+    void close() {
+        waiter_.close();
+        waiter_.attach(nullptr);
+        cc_waiter_.close();
+        cc_waiter_.attach(nullptr);
+    }
+
+public:
+    queue_waiter() = default;
+    queue_waiter(const queue_waiter&) = delete;
+    queue_waiter& operator=(const queue_waiter&) = delete;
+
+    bool connected() const noexcept {
+        return connected_.load(std::memory_order_acquire);
+    }
+
+    template <typename Elems>
+    std::size_t connect(Elems* elems) {
+        if (elems == nullptr) return invalid_value;
+        if (connected_.exchange(true, std::memory_order_acq_rel)) {
+            // if it's already connected, just return an error count
+            return invalid_value;
+        }
+        auto ret = elems->connect();
+        cc_waiter_.broadcast();
+        return ret;
+    }
+
+    template <typename Elems>
+    std::size_t disconnect(Elems* elems) {
+        if (elems == nullptr) return invalid_value;
+        if (!connected_.exchange(false, std::memory_order_acq_rel)) {
+            // if it's already disconnected, just return an error count
+            return invalid_value;
+        }
+        auto ret = elems->disconnect();
+        cc_waiter_.broadcast();
+        return ret;
+    }
+
+    template <typename Elems>
+    bool wait_for_connect(Elems* elems, std::size_t count) {
+        if (elems == nullptr) return false;
+        for (unsigned k = 0; elems->conn_count() < count;) {
+            ipc::sleep(k, [this] { return cc_waiter_.wait(); });
+        }
+        return true;
+    }
+};
+
+template <typename Elems>
+class queue_base : public queue_waiter {
+    using base_t = queue_waiter;
+
+public:
+    using elems_t  = Elems;
+    using policy_t = typename elems_t::policy_t;
+
+protected:
+    elems_t* elems_ = nullptr;
+    decltype(std::declval<elems_t>().cursor()) cursor_ = 0;
+
+public:
+    using base_t::base_t;
+
+    explicit queue_base(elems_t* els, char const * name = nullptr)
+        : queue_base() {
+        attach(els, name);
+    }
 
     constexpr elems_t * elems() const noexcept {
         return elems_;
     }
 
     std::size_t connect() {
-        if (elems_ == nullptr) return invalid_value;
-        if (connected_.exchange(true, std::memory_order_acq_rel)) {
-            // if it's already connected, just return an error count
-            return invalid_value;
-        }
-        auto ret = elems_->connect();
-        cc_waiter_.broadcast();
-        return ret;
+        return base_t::connect(elems_);
     }
 
     std::size_t disconnect() {
-        if (elems_ == nullptr) return invalid_value;
-        if (!connected_.exchange(false, std::memory_order_acq_rel)) {
-            // if it's already disconnected, just return an error count
-            return invalid_value;
-        }
-        auto ret = elems_->disconnect();
-        cc_waiter_.broadcast();
-        return ret;
+        return base_t::disconnect(elems_);
     }
 
     std::size_t conn_count() const noexcept {
@@ -72,37 +119,21 @@ public:
     }
 
     bool wait_for_connect(std::size_t count) {
-        if (elems_ == nullptr) return false;
-        for (unsigned k = 0; elems_->conn_count() < count;) {
-            ipc::sleep(k, [this] { return cc_waiter_.wait(); });
-        }
-        return true;
+        return base_t::wait_for_connect(elems_, count);
     }
 
     bool empty() const noexcept {
         return (elems_ == nullptr) ? true : (cursor_ == elems_->cursor());
     }
 
-    bool connected() const noexcept {
-        return connected_.load(std::memory_order_acquire);
-    }
-
     elems_t* attach(elems_t* els, char const * name = nullptr) noexcept {
         if (els == nullptr) return nullptr;
         auto old = elems_;
-        elems_  = els;
+        elems_ = els;
         if (name == nullptr) {
-            waiter_.close();
-            waiter_.attach(nullptr);
-            cc_waiter_.close();
-            cc_waiter_.attach(nullptr);
-        }
-        else {
-            waiter_.attach(&(elems_->waiter()));
-            waiter_.open((std::string{ "__IPC_WAITER__" } + name).c_str());
-            cc_waiter_.attach(&(elems_->conn_waiter()));
-            cc_waiter_.open((std::string{ "__IPC_CC_WAITER__" } + name).c_str());
+            base_t::close();
         }
+        else base_t::open(elems_, name);
         cursor_ = elems_->cursor();
         return old;
     }
@@ -113,33 +144,64 @@ public:
         elems_ = nullptr;
         return old;
     }
+};
 
-    template <typename... P>
+template <typename Elems, typename IsFixed>
+class queue : public queue_base<Elems> {
+    using base_t = queue_base<Elems>;
+
+public:
+    using is_fixed = IsFixed;
+
+    using base_t::base_t;
+
+    template <typename T, typename... P>
     auto push(P&&... params) {
-        if (elems_ == nullptr) return false;
-        if (elems_->push([&](void* p) {
+        if (this->elems_ == nullptr) return false;
+        if (this->elems_->push([&](void* p) {
             ::new (p) T(std::forward<P>(params)...);
-        }, std::forward<P>(params)...)) {
-            waiter_.broadcast();
+        })) {
+            this->waiter_.broadcast();
             return true;
         }
         return false;
     }
 
+    template <typename T>
     T pop() {
-        if (elems_ == nullptr) {
+        if (this->elems_ == nullptr) {
             return {};
         }
         T item;
         for (unsigned k = 0;;) {
-            if (elems_->pop(&cursor_, [&item](void* p) {
+            if (this->elems_->pop(&this->cursor_, [&item](void* p) {
                 ::new (&item) T(std::move(*static_cast<T*>(p)));
             })) {
                 return item;
             }
-            ipc::sleep(k, [this] { return waiter_.wait(); });
+            ipc::sleep(k, [this] { return this->waiter_.wait(); });
         }
     }
 };
 
+} // namespace detail
+
+template <typename T,
+          typename Policy = ipc::circ::prod_cons<relat::single, relat::multi, trans::broadcast>>
+class queue : public detail::queue<typename Policy::template elems_t<sizeof(T)>, typename Policy::is_fixed> {
+    using base_t = detail::queue<typename Policy::template elems_t<sizeof(T)>, typename Policy::is_fixed>;
+
+public:
+    using base_t::base_t;
+
+    template <typename... P>
+    auto push(P&&... params) {
+        return base_t::push<T>(std::forward<P>(params)...);
+    }
+
+    T pop() {
+        return base_t::pop<T>();
+    }
+};
+
 } // namespace ipc
diff --git a/include/rw_lock.h b/include/rw_lock.h
index f75690f..f9546aa 100644
--- a/include/rw_lock.h
+++ b/include/rw_lock.h
@@ -97,7 +97,7 @@ inline void sleep(K& k) noexcept {
 namespace ipc {
 
 class spin_lock {
-    std::atomic<std::size_t> lc_ { 0 };
+    std::atomic<unsigned> lc_ { 0 };
 
 public:
     void lock(void) noexcept {
@@ -112,7 +112,7 @@ public:
 };
 
 class rw_lock {
-    using lc_ui_t = std::size_t;
+    using lc_ui_t = unsigned;
     std::atomic<lc_ui_t> lc_ { 0 };
 
     enum : lc_ui_t {
diff --git a/src/ipc.cpp b/src/ipc.cpp
index 8405414..02db9c0 100644
--- a/src/ipc.cpp
+++ b/src/ipc.cpp
@@ -9,6 +9,7 @@
 #include "def.h"
 #include "shm.h"
 #include "tls_pointer.h"
+#include "queue.h"
 
 #include "memory/resource.hpp"
 
@@ -23,11 +24,8 @@ inline auto acc_of_msg() {
     return static_cast<std::atomic<msg_id_t>*>(g_shm.get());
 }
 
-template <template <typename...> class Queue, typename Policy>
-struct detail_impl;
-
-template <typename Policy>
-struct detail_impl<ipc::queue, Policy> {
+template <typename Policy, typename IsFixed = typename Policy::is_fixed>
+struct detail_impl {
 
 #pragma pack(1)
 struct msg_t {
@@ -217,56 +215,56 @@ static buff_t recv(ipc::handle_t h) {
     }
 }
 
-}; // detail_impl<ipc::queue>
+}; // detail_impl<Policy>
 
 } // internal-linkage
 
 namespace ipc {
 
-template <template <typename...> class Queue, typename Policy>
-ipc::handle_t channel_detail<Queue, Policy>::connect(char const * name) {
-    return detail_impl<Queue, Policy>::connect(name);
+template <typename Policy>
+ipc::handle_t channel_detail<Policy>::connect(char const * name) {
+    return detail_impl<Policy>::connect(name);
 }
 
-template <template <typename...> class Queue, typename Policy>
-void channel_detail<Queue, Policy>::disconnect(ipc::handle_t h) {
-    detail_impl<Queue, Policy>::disconnect(h);
+template <typename Policy>
+void channel_detail<Policy>::disconnect(ipc::handle_t h) {
+    detail_impl<Policy>::disconnect(h);
 }
 
-template <template <typename...> class Queue, typename Policy>
-std::size_t channel_detail<Queue, Policy>::recv_count(ipc::handle_t h) {
-    return detail_impl<Queue, Policy>::recv_count(h);
+template <typename Policy>
+std::size_t channel_detail<Policy>::recv_count(ipc::handle_t h) {
+    return detail_impl<Policy>::recv_count(h);
 }
 
-template <template <typename...> class Queue, typename Policy>
-bool channel_detail<Queue, Policy>::wait_for_recv(ipc::handle_t h, std::size_t r_count) {
-    return detail_impl<Queue, Policy>::wait_for_recv(h, r_count);
+template <typename Policy>
+bool channel_detail<Policy>::wait_for_recv(ipc::handle_t h, std::size_t r_count) {
+    return detail_impl<Policy>::wait_for_recv(h, r_count);
 }
 
-template <template <typename...> class Queue, typename Policy>
-void channel_detail<Queue, Policy>::clear_recv(ipc::handle_t h) {
-    detail_impl<Queue, Policy>::clear_recv(h);
+template <typename Policy>
+void channel_detail<Policy>::clear_recv(ipc::handle_t h) {
+    detail_impl<Policy>::clear_recv(h);
 }
 
-template <template <typename...> class Queue, typename Policy>
-void channel_detail<Queue, Policy>::clear_recv(char const * name) {
-    detail_impl<Queue, Policy>::clear_recv(name);
+template <typename Policy>
+void channel_detail<Policy>::clear_recv(char const * name) {
+    detail_impl<Policy>::clear_recv(name);
 }
 
-template <template <typename...> class Queue, typename Policy>
-bool channel_detail<Queue, Policy>::send(ipc::handle_t h, void const * data, std::size_t size) {
-    return detail_impl<Queue, Policy>::send(h, data, size);
+template <typename Policy>
+bool channel_detail<Policy>::send(ipc::handle_t h, void const * data, std::size_t size) {
+    return detail_impl<Policy>::send(h, data, size);
 }
 
-template <template <typename...> class Queue, typename Policy>
-buff_t channel_detail<Queue, Policy>::recv(ipc::handle_t h) {
-    return detail_impl<Queue, Policy>::recv(h);
+template <typename Policy>
+buff_t channel_detail<Policy>::recv(ipc::handle_t h) {
+    return detail_impl<Policy>::recv(h);
 }
 
-template struct channel_detail<ipc::queue, ipc::circ::prod_cons<relat::single, relat::single, trans::unicast  >>;
-template struct channel_detail<ipc::queue, ipc::circ::prod_cons<relat::single, relat::multi , trans::unicast  >>;
-template struct channel_detail<ipc::queue, ipc::circ::prod_cons<relat::multi , relat::multi , trans::unicast  >>;
-template struct channel_detail<ipc::queue, ipc::circ::prod_cons<relat::single, relat::multi , trans::broadcast>>;
-template struct channel_detail<ipc::queue, ipc::circ::prod_cons<relat::multi , relat::multi , trans::broadcast>>;
+template struct channel_detail<ipc::circ::prod_cons<relat::single, relat::single, trans::unicast  >>;
+template struct channel_detail<ipc::circ::prod_cons<relat::single, relat::multi , trans::unicast  >>;
+template struct channel_detail<ipc::circ::prod_cons<relat::multi , relat::multi , trans::unicast  >>;
+template struct channel_detail<ipc::circ::prod_cons<relat::single, relat::multi , trans::broadcast>>;
+template struct channel_detail<ipc::circ::prod_cons<relat::multi , relat::multi , trans::broadcast>>;
 
 } // namespace ipc