ipc::circ::queue & test_queue

build/test.pro

@@ -17,6 +17,6 @@ HEADERS += \
 SOURCES += \
     ../test/main.cpp \
     ../test/test_shm.cpp \
-    ../test/test_circ_elem_array.cpp
+    ../test/test_circ.cpp
 
 LIBS += -L$${DESTDIR} -lipc

include/circ_queue.h

@@ -1,14 +1,68 @@
 #pragma once
 
+#include <type_traits>
+#include <new>
+#include <exception>
+#include <utility>
+
 #include "circ_elem_array.h"
 
 namespace ipc {
+
+enum : std::size_t {
+    error_count = std::numeric_limits<std::size_t>::max()
+};
+
 namespace circ {
 
 template <typename T>
 class queue {
+public:
+    using array_t = elem_array<sizeof(T)>;
+
 private:
-    elem_array<sizeof(T)> elems_;
+    array_t * elems_ = nullptr;
+    typename std::result_of<decltype(&array_t::cursor)(array_t)>::type cursor_ = 0;
+
+public:
+    std::size_t connect(array_t* arr) {
+        if (arr == nullptr) return error_count;
+        elems_  = arr;
+        cursor_ = elems_->cursor();
+        return elems_->connect();
+    }
+
+    std::size_t disconnect(void) {
+        if (elems_ == nullptr) return error_count;
+        auto ret = elems_->disconnect();
+        elems_ = nullptr;
+        return ret;
+    }
+
+    std::size_t conn_count(void) const {
+        return (elems_ == nullptr) ? error_count : elems_->conn_count();
+    }
+
+    template <typename... P>
+    void push(P&&... params) {
+        if (elems_ == nullptr) return;
+        auto ptr = elems_->acquire();
+        ::new (ptr) T { std::forward<P>(params)... };
+        elems_->commit(ptr);
+    }
+
+    T pop(void) {
+        if (elems_ == nullptr) throw std::invalid_argument {
+            "This queue hasn't connected to any elem_array."
+        };
+        while (cursor_ == elems_->cursor()) {
+            std::this_thread::yield();
+        }
+        auto item_ptr = static_cast<T*>(elems_->take(cursor_));
+        T item { *item_ptr };
+        elems_->put(item_ptr);
+        return item;
+    }
 };
 
 } // namespace circ

test/test_circ.cpp

@@ -8,6 +8,7 @@
 #include <functional>
 
 #include "circ_elem_array.h"
+#include "circ_queue.h"
 #include "test.h"
 #include "stopwatch.hpp"
 
@@ -24,9 +25,11 @@ private slots:
     void test_prod_cons_1v1(void);
     void test_prod_cons_1v3(void);
     void test_prod_cons_performance(void);
+
+    void test_queue(void);
 } unit__;
 
-#include "test_circ_elem_array.moc"
+#include "test_circ.moc"
 
 using cq_t = ipc::circ::elem_array<12>;
 cq_t* cq__;
@@ -57,6 +60,11 @@ void Unit::test_inst(void) {
              static_cast<std::size_t>(cq_t::elem_size));
 }
 
+struct msg_t {
+    int pid_;
+    int dat_;
+};
+
 template <int N, int M, bool Confirmation = true, int Loops = 1000000>
 void test_prod_cons(void) {
     ::new (cq__) cq_t;
@@ -65,11 +73,6 @@ void test_prod_cons(void) {
     std::atomic_int fini { 0 };
     capo::stopwatch<> sw;
 
-    struct msg_t {
-        int pid_;
-        int dat_;
-    };
-
     std::unordered_map<int, std::vector<int>> list[std::extent<decltype(consumers)>::value];
     auto push_data = Confirmation ? [](std::vector<int>& l, int dat) {
         l.push_back(dat);
@@ -176,4 +179,14 @@ void Unit::test_prod_cons_performance(void) {
     test_performance<1, 10>::start();
 }
 
+void Unit::test_queue(void) {
+    ipc::circ::queue<msg_t> queue;
+    queue.push(1, 2);
+    QVERIFY_EXCEPTION_THROWN(queue.pop(), std::exception);
+    QVERIFY(sizeof(decltype(queue)::array_t) <= sizeof(*cq__));
+
+    auto cq = ::new (cq__) decltype(queue)::array_t;
+    QVERIFY(queue.connect(cq) != ipc::error_count);
+}
+
 } // internal-linkage