From 5c9a8e031116661929ddcf93a7ca3987b4688409 Mon Sep 17 00:00:00 2001
From: mutouyun <mark.lonr@tom.com>
Date: Thu, 24 Jan 2019 17:55:23 +0800
Subject: [PATCH] remove is_fixed (TBD)

---
 build/ipc.pro         |   4 +-
 include/def.h         |   3 +-
 include/pool_alloc.h  |  86 +++++++++++++++++++++++++++++++++
 include/rw_lock.h     |   1 +
 src/circ/elem_array.h |  74 ++++++++++++++++++++--------
 src/circ/elem_chan.h  |  11 -----
 src/circ/elem_def.h   |   2 +-
 src/ipc.cpp           |  24 +++------
 src/memory/wrapper.h  |   1 -
 src/platform/detail.h |   8 +++
 src/policy.h          |   2 -
 src/prod_cons.h       |  10 ++--
 src/queue.h           | 110 +++++++++++++++++++++++-------------------
 test/test_circ.cpp    |   6 ++-
 14 files changed, 232 insertions(+), 110 deletions(-)
 delete mode 100644 src/circ/elem_chan.h

diff --git a/build/ipc.pro b/build/ipc.pro
index 5ece147..f936a9b 100644
--- a/build/ipc.pro
+++ b/build/ipc.pro
@@ -32,9 +32,9 @@ HEADERS += \
     ../src/platform/waiter.h \
     ../src/circ/elem_def.h \
     ../src/circ/elem_array.h \
-    ../src/circ/elem_chan.h \
     ../src/prod_cons.h \
-    ../src/policy.h
+    ../src/policy.h \
+    ../src/queue.h
 
 SOURCES += \
     ../src/shm.cpp \
diff --git a/include/def.h b/include/def.h
index 5466cf9..2a17ecc 100644
--- a/include/def.h
+++ b/include/def.h
@@ -46,7 +46,8 @@ using uint_t = typename uint<N>::type;
 
 enum : std::size_t {
     invalid_value = (std::numeric_limits<std::size_t>::max)(),
-    data_length   = 16
+    data_length   = 16,
+    name_length   = 64
 };
 
 enum class relat { // multiplicity of the relationship
diff --git a/include/pool_alloc.h b/include/pool_alloc.h
index 8747e08..8d96237 100644
--- a/include/pool_alloc.h
+++ b/include/pool_alloc.h
@@ -1,5 +1,8 @@
 #pragma once
 
+#include <new>
+#include <utility>
+
 #include "export.h"
 #include "def.h"
 
@@ -13,5 +16,88 @@ public:
     static void  free(void* p, std::size_t size);
 };
 
+////////////////////////////////////////////////////////////////
+/// construct/destruct an object
+////////////////////////////////////////////////////////////////
+
+namespace detail {
+
+template <typename T>
+struct impl {
+    template <typename... P>
+    static T* construct(T* p, P&&... params) {
+        ::new (p) T(std::forward<P>(params)...);
+        return p;
+    }
+
+    static void destruct(T* p) {
+        reinterpret_cast<T*>(p)->~T();
+    }
+};
+
+template <typename T, size_t N>
+struct impl<T[N]> {
+    using type = T[N];
+
+    template <typename... P>
+    static type* construct(type* p, P&&... params) {
+        for (size_t i = 0; i < N; ++i) {
+            impl<T>::construct(&((*p)[i]), std::forward<P>(params)...);
+        }
+        return p;
+    }
+
+    static void destruct(type* p) {
+        for (size_t i = 0; i < N; ++i) {
+            impl<T>::destruct(&((*p)[i]));
+        }
+    }
+};
+
+} // namespace detail
+
+template <typename T, typename... P>
+T* construct(T* p, P&&... params) {
+    return detail::impl<T>::construct(p, std::forward<P>(params)...);
+}
+
+template <typename T, typename... P>
+T* construct(void* p, P&&... params) {
+    return construct(static_cast<T*>(p), std::forward<P>(params)...);
+}
+
+template <typename T>
+void destruct(T* p) {
+    return detail::impl<T>::destruct(p);
+}
+
+template <typename T>
+void destruct(void* p) {
+    destruct(static_cast<T*>(p));
+}
+
+////////////////////////////////////////////////////////////////
+/// general alloc/free
+////////////////////////////////////////////////////////////////
+
+inline void* alloc(std::size_t size) {
+    return pool_alloc::alloc(size);
+}
+
+template <typename T, typename... P>
+T* alloc(P&&... params) {
+    return construct<T>(pool_alloc::alloc(sizeof(T)), std::forward<P>(params)...);
+}
+
+inline void free(void* p, std::size_t size) {
+    pool_alloc::free(p, size);
+}
+
+template <typename T>
+void free(T* p) {
+    destruct(p);
+    pool_alloc::free(p, sizeof(T));
+}
+
 } // namespace mem
 } // namespace ipc
diff --git a/include/rw_lock.h b/include/rw_lock.h
index f9546aa..d01f70c 100644
--- a/include/rw_lock.h
+++ b/include/rw_lock.h
@@ -113,6 +113,7 @@ public:
 
 class rw_lock {
     using lc_ui_t = unsigned;
+
     std::atomic<lc_ui_t> lc_ { 0 };
 
     enum : lc_ui_t {
diff --git a/src/circ/elem_array.h b/src/circ/elem_array.h
index b3e134f..dd97281 100644
--- a/src/circ/elem_array.h
+++ b/src/circ/elem_array.h
@@ -1,37 +1,31 @@
 #pragma once
 
-#include <atomic>
-#include <thread>
-#include <cstring>
+#include <limits>
 #include <utility>
 #include <type_traits>
 
 #include "def.h"
-#include "rw_lock.h"
 
 #include "circ/elem_def.h"
 #include "platform/detail.h"
 
 namespace ipc {
 namespace circ {
-
-////////////////////////////////////////////////////////////////
-/// element-array implementation
-////////////////////////////////////////////////////////////////
+namespace detail {
 
 template <typename Policy, std::size_t DataSize>
-class elem_array : public ipc::circ::conn_head {
+class elem_array {
 public:
-    using base_t   = ipc::circ::conn_head;
     using policy_t = Policy;
+    using cursor_t = decltype(std::declval<policy_t>().cursor());
+
 #if __cplusplus >= 201703L
-    using elem_t   = ipc::circ::elem_t<policy_t::template elem_param<DataSize>>;
+    using elem_t = ipc::circ::elem_t<policy_t::template elem_param<DataSize>>;
 #else /*__cplusplus < 201703L*/
-    using elem_t   = ipc::circ::elem_t<policy_t::template elem_param<DataSize>::value>;
+    using elem_t = ipc::circ::elem_t<policy_t::template elem_param<DataSize>::value>;
 #endif/*__cplusplus < 201703L*/
 
     enum : std::size_t {
-        head_size  = sizeof(base_t) + sizeof(policy_t),
         data_size  = DataSize,
         elem_max   = (std::numeric_limits<uint_t<8>>::max)() + 1, // default is 255 + 1
         elem_size  = sizeof(elem_t),
@@ -43,17 +37,57 @@ private:
     elem_t   block_[elem_max];
 
 public:
-    auto cursor() const noexcept { return head_.cursor(); }
+    cursor_t cursor() const noexcept {
+        return head_.cursor();
+    }
 
-    template <typename F>
-    bool push(F&& f) {
-        return head_.push(this, std::forward<F>(f), block_);
+    template <typename E, typename F>
+    bool push(E* elems, F&& f) {
+        return head_.push(elems, std::forward<F>(f), block_);
+    }
+
+    template <typename E, typename F>
+    bool pop(E* elems, cursor_t* cur, F&& f) {
+        if (cur == nullptr) return false;
+        return head_.pop(elems, *cur, std::forward<F>(f), block_);
+    }
+};
+
+} // namespace detail
+
+template <typename Policy, std::size_t DataSize>
+class elem_array : public ipc::circ::conn_head {
+public:
+    using base_t   = ipc::circ::conn_head;
+    using array_t  = detail::elem_array<Policy, DataSize>;
+    using policy_t = typename array_t::policy_t;
+    using cursor_t = typename array_t::cursor_t;
+    using elem_t   = typename array_t::elem_t;
+
+    enum : std::size_t {
+        head_size  = sizeof(base_t) + sizeof(policy_t),
+        data_size  = array_t::data_size,
+        elem_max   = array_t::elem_max,
+        elem_size  = array_t::elem_size,
+        block_size = array_t::block_size
+    };
+
+private:
+    array_t array_;
+
+public:
+    cursor_t cursor() const noexcept {
+        return array_.cursor();
     }
 
     template <typename F>
-    bool pop(decltype(std::declval<policy_t>().cursor())* cur, F&& f) {
-        if (cur == nullptr) return false;
-        return head_.pop(this, *cur, std::forward<F>(f), block_);
+    bool push(F&& f) {
+        return array_.push(this, std::forward<F>(f));
+    }
+
+    template <typename F>
+    bool pop(cursor_t* cur, F&& f) {
+        return array_.pop(this, cur, std::forward<F>(f));
     }
 };
 
diff --git a/src/circ/elem_chan.h b/src/circ/elem_chan.h
deleted file mode 100644
index a59b38c..0000000
--- a/src/circ/elem_chan.h
+++ /dev/null
@@ -1,11 +0,0 @@
-#pragma once
-
-#include "circ/elem_array.h"
-
-namespace ipc {
-namespace circ {
-
-
-
-} // namespace circ
-} // namespace ipc
diff --git a/src/circ/elem_def.h b/src/circ/elem_def.h
index b9ee247..025b054 100644
--- a/src/circ/elem_def.h
+++ b/src/circ/elem_def.h
@@ -38,7 +38,7 @@ constexpr u1_t index_of(u2_t c) noexcept {
 
 class conn_head {
     ipc::detail::waiter      cc_waiter_, waiter_;
-    std::atomic<std::size_t> cc_ { 0 }; // connection counter
+    std::atomic<std::size_t> cc_; // connection counter
 
 public:
     conn_head() = default;
diff --git a/src/ipc.cpp b/src/ipc.cpp
index f3053fd..48b4b34 100644
--- a/src/ipc.cpp
+++ b/src/ipc.cpp
@@ -9,9 +9,10 @@
 #include "def.h"
 #include "shm.h"
 #include "tls_pointer.h"
+#include "pool_alloc.h"
 #include "queue.h"
-
 #include "policy.h"
+
 #include "memory/resource.h"
 
 namespace {
@@ -25,7 +26,7 @@ inline auto acc_of_msg() {
     return static_cast<std::atomic<msg_id_t>*>(g_shm.get());
 }
 
-template <typename Policy, typename IsFixed = typename Policy::is_fixed>
+template <typename Policy>
 struct detail_impl {
 
 #pragma pack(1)
@@ -39,10 +40,6 @@ struct msg_t {
 
 using queue_t = ipc::queue<msg_t, Policy>;
 
-struct shm_info_t {
-    typename queue_t::elems_t elems_;  // the elements in shm
-};
-
 constexpr static void* head_of(queue_t* que) {
     return static_cast<void*>(que->elems());
 }
@@ -52,9 +49,9 @@ constexpr static queue_t* queue_of(ipc::handle_t h) {
 }
 
 static buff_t make_cache(void const * data, std::size_t size) {
-    auto ptr = mem::sync_pool_alloc::alloc(size);
+    auto ptr = mem::alloc(size);
     std::memcpy(ptr, data, size);
-    return { ptr, size, mem::sync_pool_alloc::free };
+    return { ptr, size, mem::free };
 }
 
 struct cache_t {
@@ -93,11 +90,7 @@ static auto& queues_cache() {
 /* API implementations */
 
 static ipc::handle_t connect(char const * name) {
-    auto mem = shm::acquire(name, sizeof(shm_info_t));
-    if (mem == nullptr) {
-        return nullptr;
-    }
-    return new queue_t { &(static_cast<shm_info_t*>(mem)->elems_), name };
+    return mem::alloc<queue_t>(name);
 }
 
 static void disconnect(ipc::handle_t h) {
@@ -106,8 +99,7 @@ static void disconnect(ipc::handle_t h) {
         return;
     }
     que->disconnect(); // needn't to detach, cause it will be deleted soon.
-    shm::release(head_of(que), sizeof(shm_info_t));
-    delete que;
+    mem::free(que);
 }
 
 static std::size_t recv_count(ipc::handle_t h) {
@@ -131,7 +123,7 @@ static void clear_recv(ipc::handle_t h) {
     if (head == nullptr) {
         return;
     }
-    std::memset(head, 0, sizeof(shm_info_t));
+    std::memset(head, 0, sizeof(queue_t::elems_t));
 }
 
 static void clear_recv(char const * name) {
diff --git a/src/memory/wrapper.h b/src/memory/wrapper.h
index c632da5..71c7996 100644
--- a/src/memory/wrapper.h
+++ b/src/memory/wrapper.h
@@ -2,7 +2,6 @@
 
 #include <limits>
 #include <new>
-#include <mutex>
 #include <tuple>
 #include <map>
 #include <vector>
diff --git a/src/platform/detail.h b/src/platform/detail.h
index 760544a..3f82d61 100644
--- a/src/platform/detail.h
+++ b/src/platform/detail.h
@@ -2,6 +2,7 @@
 
 #include <memory>
 #include <mutex>
+#include <shared_mutex>
 #include <type_traits>
 
 #include "def.h"
@@ -21,6 +22,7 @@ namespace detail {
 
 using std::unique_ptr;
 using std::unique_lock;
+using std::shared_lock;
 
 #else /*__cplusplus < 201703L*/
 
@@ -39,6 +41,12 @@ constexpr auto unique_lock(T&& lc) {
     return std::unique_lock<std::decay_t<T>> { std::forward<T>(lc) };
 }
 
+// deduction guides for std::shared_lock
+template <typename T>
+constexpr auto shared_lock(T&& lc) {
+    return std::shared_lock<std::decay_t<T>> { std::forward<T>(lc) };
+}
+
 #endif/*__cplusplus < 201703L*/
 
 template <typename F, typename D>
diff --git a/src/policy.h b/src/policy.h
index b89ad80..79da3ae 100644
--- a/src/policy.h
+++ b/src/policy.h
@@ -16,8 +16,6 @@ struct choose;
 
 template <typename Flag>
 struct choose<circ::elem_array, Flag> {
-    using is_fixed = std::true_type;
-
     template <std::size_t DataSize>
     using elems_t = circ::elem_array<ipc::prod_cons_impl<Flag>, DataSize>;
 };
diff --git a/src/prod_cons.h b/src/prod_cons.h
index 5b657de..ba43fbc 100644
--- a/src/prod_cons.h
+++ b/src/prod_cons.h
@@ -20,8 +20,8 @@ struct prod_cons_impl;
 
 template <>
 struct prod_cons_impl<prod_cons<relat::single, relat::single, trans::unicast>> {
-    std::atomic<circ::u2_t> rd_ { 0 }; // read index
-    std::atomic<circ::u2_t> wt_ { 0 }; // write index
+    std::atomic<circ::u2_t> rd_; // read index
+    std::atomic<circ::u2_t> wt_; // write index
 
 #if __cplusplus >= 201703L
     template <std::size_t DataSize>
@@ -89,7 +89,7 @@ template <>
 struct prod_cons_impl<prod_cons<relat::multi , relat::multi, trans::unicast>>
      : prod_cons_impl<prod_cons<relat::single, relat::multi, trans::unicast>> {
 
-    std::atomic<circ::u2_t> ct_ { 0 }; // commit index
+    std::atomic<circ::u2_t> ct_; // commit index
 
     template <typename E, typename F, typename EB>
     bool push(E* /*elems*/, F&& f, EB* elem_start) {
@@ -119,7 +119,7 @@ struct prod_cons_impl<prod_cons<relat::multi , relat::multi, trans::unicast>>
 
 template <>
 struct prod_cons_impl<prod_cons<relat::single, relat::multi, trans::broadcast>> {
-    std::atomic<circ::u2_t> wt_ { 0 }; // write index
+    std::atomic<circ::u2_t> wt_; // write index
 
 #if __cplusplus >= 201703L
     template <std::size_t DataSize>
@@ -183,7 +183,7 @@ template <>
 struct prod_cons_impl<prod_cons<relat::multi , relat::multi, trans::broadcast>>
      : prod_cons_impl<prod_cons<relat::single, relat::multi, trans::broadcast>> {
 
-    std::atomic<circ::u2_t> ct_ { 0 }; // commit index
+    std::atomic<circ::u2_t> ct_; // commit index
 
     template <typename E, typename F, typename EB>
     bool push(E* elems, F&& f, EB* elem_start) {
diff --git a/src/queue.h b/src/queue.h
index b0563b9..5f2cb62 100644
--- a/src/queue.h
+++ b/src/queue.h
@@ -9,8 +9,10 @@
 #include <thread>
 #include <chrono>
 #include <string>
+#include <cassert>
 
 #include "def.h"
+#include "shm.h"
 #include "rw_lock.h"
 
 #include "platform/waiter.h"
@@ -20,11 +22,29 @@ namespace detail {
 
 class queue_waiter {
 protected:
-    ipc::detail::waiter_impl waiter_, cc_waiter_;
-    std::atomic_bool connected_ { false };
+    ipc::detail::waiter_impl waiter_;
+    ipc::detail::waiter_impl cc_waiter_;
+
+    bool connected_ = false;
+    bool dismiss_   = true;
 
     template <typename Elems>
-    void open(Elems* elems, char const * name) {
+    Elems* open(char const * name) {
+        auto elems = static_cast<Elems*>(shm::acquire(name, sizeof(Elems)));
+        if (elems == nullptr) {
+            return nullptr;
+        }
+        dismiss_ = false;
+        return elems;
+    }
+
+    template <typename Elems>
+    void open(Elems*(& elems), char const * name) {
+        assert(name != nullptr && name[0] != '\0');
+        if (elems == nullptr) {
+            elems = open<Elems>(name);
+        }
+        assert(elems != nullptr);
         waiter_.attach(&(elems->waiter()));
         waiter_.open((std::string{ "__IPC_WAITER__" } + name).c_str());
         cc_waiter_.attach(&(elems->conn_waiter()));
@@ -38,22 +58,32 @@ protected:
         cc_waiter_.attach(nullptr);
     }
 
+    template <typename Elems>
+    void close(Elems* elems) {
+        if (!dismiss_ && (elems != nullptr)) {
+            shm::release(elems, sizeof(Elems));
+        }
+        dismiss_ = true;
+        close();
+    }
+
 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);
+        return connected_;
     }
 
     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 (connected_) {
             // if it's already connected, just return an error count
             return invalid_value;
         }
+        connected_ = true;
         auto ret = elems->connect();
         cc_waiter_.broadcast();
         return ret;
@@ -62,10 +92,11 @@ public:
     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 (!connected_) {
             // if it's already disconnected, just return an error count
             return invalid_value;
         }
+        connected_ = false;
         auto ret = elems->disconnect();
         cc_waiter_.broadcast();
         return ret;
@@ -98,11 +129,20 @@ public:
 
     queue_base() = default;
 
+    explicit queue_base(char const * name)
+        : queue_base() {
+        attach(nullptr, name);
+    }
+
     explicit queue_base(elems_t* els, char const * name = nullptr)
         : queue_base() {
         attach(els, name);
     }
 
+    /* not virtual */ ~queue_base(void) {
+        base_t::close(elems_);
+    }
+
     constexpr elems_t * elems() const noexcept {
         return elems_;
     }
@@ -123,33 +163,39 @@ public:
         return base_t::wait_for_connect(elems_, count);
     }
 
+    bool valid() const noexcept {
+        return elems_ != nullptr;
+    }
+
     bool empty() const noexcept {
         return (elems_ == nullptr) ? true : (cursor_ == elems_->cursor());
     }
 
     elems_t* attach(elems_t* els, char const * name = nullptr) noexcept {
-        if (els == nullptr) return nullptr;
         auto old = elems_;
         elems_ = els;
-        if (name == nullptr) {
-            base_t::close();
+        if (name == nullptr || name[0] == '\0') {
+            base_t::close(old);
         }
         else base_t::open(elems_, name);
-        cursor_ = elems_->cursor();
+        if (elems_ != nullptr) {
+            cursor_ = elems_->cursor();
+        }
         return old;
     }
 
     elems_t* detach() noexcept {
         if (elems_ == nullptr) return nullptr;
+        base_t::close<elems_t>(nullptr); // not release shm
         auto old = elems_;
         elems_ = nullptr;
         return old;
     }
 
-    template <typename T, typename F, typename... P>
-    auto push(F&& f, P&&... params) {
+    template <typename T, typename... P>
+    auto push(P&&... params) {
         if (elems_ == nullptr) return false;
-        if (std::forward<F>(f)([&](void* p) {
+        if (elems_->push([&](void* p) {
             ::new (p) T(std::forward<P>(params)...);
         })) {
             this->waiter_.broadcast();
@@ -175,45 +221,11 @@ public:
     }
 };
 
-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) {
-        return base_t::template push<T>([this](auto&& f) {
-            return this->elems_->push(std::forward<decltype(f)>(f));
-        }, std::forward<P>(params)...);
-    }
-};
-
-template <typename Elems>
-class queue<Elems, std::false_type> : public queue_base<Elems> {
-    using base_t = queue_base<Elems>;
-
-public:
-    using is_fixed = std::false_type;
-
-    using base_t::base_t;
-
-    template <typename T, typename... P>
-    auto push(P&&... params) {
-        return base_t::template push<T>([this](auto&& f) {
-            return this->elems_->template push<sizeof(T), alignof(T)>(std::forward<decltype(f)>(f));
-        }, std::forward<P>(params)...);
-    }
-};
-
 } // namespace detail
 
 template <typename T, typename Policy>
-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>;
+class queue : public detail::queue_base<typename Policy::template elems_t<sizeof(T)>> {
+    using base_t = detail::queue_base<typename Policy::template elems_t<sizeof(T)>>;
 
 public:
     using base_t::base_t;
diff --git a/test/test_circ.cpp b/test/test_circ.cpp
index f93bd56..dee2808 100644
--- a/test/test_circ.cpp
+++ b/test/test_circ.cpp
@@ -176,7 +176,7 @@ struct test_cq<ipc::queue<T...>> {
     ca_t* ca_;
 
     test_cq(void*) : ca_(reinterpret_cast<ca_t*>(cq__)) {
-        ::new (ca_) ca_t;
+        std::memset(ca_, 0, sizeof(ca_t));
     }
 
     cn_t* connect() {
@@ -243,6 +243,7 @@ constexpr int LoopCount = 1000000;
 void Unit::initTestCase() {
     TestSuite::initTestCase();
     cq__ = new cq_t;
+    std::memset(cq__, 0, sizeof(cq_t));
 }
 
 void Unit::cleanupTestCase() {
@@ -378,7 +379,8 @@ void Unit::test_queue() {
     QCOMPARE(queue.pop(), msg_t{});
     QVERIFY(sizeof(decltype(queue)::elems_t) <= sizeof(*cq__));
 
-    auto cq = ::new (cq__) decltype(queue)::elems_t;
+    std::memset(cq__, 0, sizeof(decltype(queue)::elems_t));
+    auto cq = reinterpret_cast<decltype(queue)::elems_t*>(cq__);
     queue.attach(cq);
     QVERIFY(queue.detach() != nullptr);