Refactor locking_deque to new interface.

lib/pls/include/pls/dataflow/internal/graph.h

@@ -98,7 +98,7 @@ class graph<inputs<I0, I...>, outputs<O0, O...>> : public node {
   int num_successors() const override {
   int num_successors() const override {
     return 0;
     return 0;
   }
   }
-  node *successor_at(int pos) const override {
+  node *successor_at(int) const override {
     PLS_ERROR("A graph instance has no direct successor!")
     PLS_ERROR("A graph instance has no direct successor!")
   }
   }
 
 
@@ -110,7 +110,7 @@ class graph<inputs<I0, I...>, outputs<O0, O...>> : public node {
     return sizeof(invocation_memory);
     return sizeof(invocation_memory);
   }
   }
   void init_instance_buffer(void *memory) const override {
   void init_instance_buffer(void *memory) const override {
-    auto invocation = new(memory) invocation_memory{};
+    new(memory) invocation_memory{};
   };
   };
 
 
  private:
  private:

lib/pls/include/pls/dataflow/internal/in_port.h

@@ -74,7 +74,7 @@ class multi_in_port<CB, N, I0, I...> : public in_port<I0> {
   using child_type = multi_in_port<CB, N + 1, I...>;
   using child_type = multi_in_port<CB, N + 1, I...>;
   using value_type = I0;
   using value_type = I0;
 
 
-  explicit multi_in_port(node *owning_node, CB *cb) : in_port<I0>{owning_node}, cb_{cb}, rec_{owning_node, cb} {};
+  explicit multi_in_port(node *owning_node, CB *cb) : in_port<I0>{owning_node}, rec_{owning_node, cb}, cb_{cb} {};
 
 
   void token_pushed(token<I0> token) override {
   void token_pushed(token<I0> token) override {
     cb_->template token_pushed<N, I0>(token);
     cb_->template token_pushed<N, I0>(token);

lib/pls/include/pls/dataflow/internal/out_port.h

@@ -100,7 +100,7 @@ class multi_out_port {
   template<typename DUMMY>
   template<typename DUMMY>
   struct next_node<num_outputs, DUMMY> {
   struct next_node<num_outputs, DUMMY> {
     const out_port_tupel_type *outputs_;
     const out_port_tupel_type *outputs_;
-    node *get(int i) {
+    node *get(int) {
       PLS_ERROR("Try to access invalid successor node index!")
       PLS_ERROR("Try to access invalid successor node index!")
     }
     }
   };
   };

lib/pls/include/pls/dataflow/internal/split_node.h

@@ -57,8 +57,8 @@ class split_node : public node {
   int instance_buffer_size() const override {
   int instance_buffer_size() const override {
     return 0;
     return 0;
   }
   }
-  void init_instance_buffer(void *memory) const override {
-    // No need for memory, we simply forward
+  void init_instance_buffer(void *) const override {
+    // No need for memory, we simply forward entries without buffering
   };
   };
 
 
  private:
  private:

lib/pls/include/pls/internal/data_structures/deque.h

@@ -9,8 +9,8 @@ namespace pls {
 namespace internal {
 namespace internal {
 namespace data_structures {
 namespace data_structures {
 
 
-template<typename Item>
-using deque = work_stealing_deque<Item>;
+template<typename Task>
+using deque = work_stealing_deque<Task>;
 
 
 }
 }
 }
 }

lib/pls/include/pls/internal/data_structures/locking_deque.h

@@ -11,6 +11,8 @@ namespace pls {
 namespace internal {
 namespace internal {
 namespace data_structures {
 namespace data_structures {
 
 
+using deque_offset = aligned_stack::stack_offset;
+
 /**
 /**
  * Wraps any object into a deque item.
  * Wraps any object into a deque item.
  */
  */
@@ -45,23 +47,26 @@ class locking_deque {
   locking_deque_item<Item> *head_;
   locking_deque_item<Item> *head_;
   locking_deque_item<Item> *tail_;
   locking_deque_item<Item> *tail_;
 
 
+  locking_deque_item<Item> *last_inserted_;
+
   base::spin_lock lock_;
   base::spin_lock lock_;
 
 
  public:
  public:
-  using state = aligned_stack::stack_offset;
-
   explicit locking_deque(aligned_stack *stack)
   explicit locking_deque(aligned_stack *stack)
       : stack_{stack}, head_{nullptr}, tail_{nullptr}, lock_{} {}
       : stack_{stack}, head_{nullptr}, tail_{nullptr}, lock_{} {}
 
 
   template<typename T, typename ...ARGS>
   template<typename T, typename ...ARGS>
-  T *push_tail(ARGS &&... args);
-  template<typename T, typename Function, typename ...ARGS>
-  T *push_tail_cb(const Function &after_creation, ARGS &&... args);
-  Item *pop_tail();
-  Item *pop_head();
-
-  void release_memory_until(state state);
-  state save_state();
+  T *push_task(ARGS &&... args);
+  template<typename T, typename ...ARGS>
+  T *push_object(ARGS &&... args);
+  void *push_bytes(size_t size);
+  void publish_last_task();
+
+  Item *pop_local_task();
+  Item *pop_external_task();
+
+  void reset_offset(deque_offset state);
+  deque_offset save_offset();
 };
 };
 
 
 }
 }

lib/pls/include/pls/internal/data_structures/locking_deque_impl.h

@@ -6,37 +6,52 @@ namespace pls {
 namespace internal {
 namespace internal {
 namespace data_structures {
 namespace data_structures {
 
 
-template<typename Item>
+template<typename Task>
 template<typename T, typename ...ARGS>
 template<typename T, typename ...ARGS>
-T *locking_deque<Item>::push_tail(ARGS &&... args) {
-  return push_tail_cb<T>([](T *) {}, std::forward<ARGS>(args)...);
+T *locking_deque<Task>::push_task(ARGS &&...args) {
+  static_assert(std::is_same<Task, T>::value || std::is_base_of<Task, T>::value,
+                "Must only push types of <Item> onto work_stealing_deque<Item>");
+
+  // Allocate object
+  auto deque_item = stack_->push<locking_deque_container<Task, T>>(std::forward<ARGS>(args)...);
+  deque_item->item_ = &deque_item->content_;
+
+  // Keep for later publishing
+  last_inserted_ = deque_item;
+
+  // ...actual data reference
+  return &deque_item->content_;
 }
 }
 
 
-template<typename Item>
-template<typename T, typename Function, typename ...ARGS>
-T *locking_deque<Item>::push_tail_cb(const Function &after_creation, ARGS &&...args) {
-  static_assert(std::is_same<Item, T>::value || std::is_base_of<Item, T>::value,
-                "Must only push types of <Item> onto work_stealing_deque<Item>");
+template<typename Task>
+template<typename T, typename ...ARGS>
+T *locking_deque<Task>::push_object(ARGS &&... args) {
+  // Simply add data to the stack, do not publish it in any way
+  return stack_->push<T>(std::forward(args)...);
+}
+
+template<typename Task>
+void *locking_deque<Task>::push_bytes(size_t size) {
+  // Simply add data to the stack, do not publish it in any way
+  return stack_->push_bytes(size);
+}
 
 
+template<typename Task>
+void locking_deque<Task>::publish_last_task() {
   std::lock_guard<base::spin_lock> lock{lock_};
   std::lock_guard<base::spin_lock> lock{lock_};
-  auto deque_item = stack_->push<locking_deque_container<Item, T>>(std::forward<ARGS>(args)...);
-  deque_item->item_ = &deque_item->content_;
-  after_creation(&deque_item->content_);
 
 
   if (tail_ != nullptr) {
   if (tail_ != nullptr) {
-    tail_->next_ = deque_item;
+    tail_->next_ = last_inserted_;
   } else {
   } else {
-    head_ = deque_item;
+    head_ = last_inserted_;
   }
   }
-  deque_item->prev_ = tail_;
-  deque_item->next_ = nullptr;
-  tail_ = deque_item;
-
-  return &deque_item->content_;
+  last_inserted_->prev_ = tail_;
+  last_inserted_->next_ = nullptr;
+  tail_ = last_inserted_;
 }
 }
 
 
-template<typename Item>
-Item *locking_deque<Item>::pop_tail() {
+template<typename Task>
+Task *locking_deque<Task>::pop_local_task() {
   std::lock_guard<base::spin_lock> lock{lock_};
   std::lock_guard<base::spin_lock> lock{lock_};
 
 
   if (tail_ == nullptr) {
   if (tail_ == nullptr) {
@@ -54,8 +69,8 @@ Item *locking_deque<Item>::pop_tail() {
   return result->item_;
   return result->item_;
 }
 }
 
 
-template<typename Item>
-Item *locking_deque<Item>::pop_head() {
+template<typename Task>
+Task *locking_deque<Task>::pop_external_task() {
   std::lock_guard<base::spin_lock> lock{lock_};
   std::lock_guard<base::spin_lock> lock{lock_};
 
 
   if (head_ == nullptr) {
   if (head_ == nullptr) {
@@ -73,12 +88,13 @@ Item *locking_deque<Item>::pop_head() {
   return result->item_;
   return result->item_;
 }
 }
 
 
-template<typename Item>
-void locking_deque<Item>::release_memory_until(state state) {
+template<typename Task>
+void locking_deque<Task>::reset_offset(deque_offset state) {
   stack_->reset_offset(state);
   stack_->reset_offset(state);
 }
 }
-template<typename Item>
-typename locking_deque<Item>::state locking_deque<Item>::save_state() {
+
+template<typename Task>
+deque_offset locking_deque<Task>::save_offset() {
   return stack_->save_offset();
   return stack_->save_offset();
 }
 }
 
 

test/data_structures_test.cpp

@@ -191,77 +191,98 @@ TEST_CASE("locking_deque functions correctly", "[internal/data_structures/lockin
     int one = 1, two = 2, three = 3, four = 4;
     int one = 1, two = 2, three = 3, four = 4;
 
 
     SECTION("add and remove items form the tail") {
     SECTION("add and remove items form the tail") {
-      deque.push_tail<int>(one);
-      deque.push_tail<int>(two);
-      deque.push_tail<int>(three);
+      deque.push_task<int>(one);
+      deque.publish_last_task();
+      deque.push_task<int>(two);
+      deque.publish_last_task();
+      deque.push_task<int>(three);
+      deque.publish_last_task();
 
 
-      REQUIRE(*deque.pop_tail() == three);
-      REQUIRE(*deque.pop_tail() == two);
-      REQUIRE(*deque.pop_tail() == one);
+      REQUIRE(*deque.pop_local_task() == three);
+      REQUIRE(*deque.pop_local_task() == two);
+      REQUIRE(*deque.pop_local_task() == one);
     }
     }
 
 
     SECTION("handles getting empty by popping the tail correctly") {
     SECTION("handles getting empty by popping the tail correctly") {
-      deque.push_tail<int>(one);
-      REQUIRE(*deque.pop_tail() == one);
+      deque.push_task<int>(one);
+      deque.publish_last_task();
+      REQUIRE(*deque.pop_local_task() == one);
 
 
-      deque.push_tail<int>(two);
-      REQUIRE(*deque.pop_tail() == two);
+      deque.push_task<int>(two);
+      deque.publish_last_task();
+      REQUIRE(*deque.pop_local_task() == two);
     }
     }
 
 
     SECTION("remove items form the head") {
     SECTION("remove items form the head") {
-      deque.push_tail<int>(one);
-      deque.push_tail<int>(two);
-      deque.push_tail<int>(three);
+      deque.push_task<int>(one);
+      deque.publish_last_task();
+      deque.push_task<int>(two);
+      deque.publish_last_task();
+      deque.push_task<int>(three);
+      deque.publish_last_task();
 
 
-      REQUIRE(*deque.pop_head() == one);
-      REQUIRE(*deque.pop_head() == two);
-      REQUIRE(*deque.pop_head() == three);
+      REQUIRE(*deque.pop_external_task() == one);
+      REQUIRE(*deque.pop_external_task() == two);
+      REQUIRE(*deque.pop_external_task() == three);
     }
     }
 
 
     SECTION("handles getting empty by popping the head correctly") {
     SECTION("handles getting empty by popping the head correctly") {
-      deque.push_tail<int>(one);
-      REQUIRE(*deque.pop_head() == one);
+      deque.push_task<int>(one);
+      deque.publish_last_task();
+      REQUIRE(*deque.pop_external_task() == one);
 
 
-      deque.push_tail<int>(two);
-      REQUIRE(*deque.pop_head() == two);
+      deque.push_task<int>(two);
+      deque.publish_last_task();
+      REQUIRE(*deque.pop_external_task() == two);
     }
     }
 
 
     SECTION("handles getting empty by popping the head and tail correctly") {
     SECTION("handles getting empty by popping the head and tail correctly") {
-      deque.push_tail<int>(one);
-      REQUIRE(*deque.pop_tail() == one);
+      deque.push_task<int>(one);
+      deque.publish_last_task();
+      REQUIRE(*deque.pop_local_task() == one);
 
 
-      deque.push_tail<int>(two);
-      REQUIRE(*deque.pop_head() == two);
+      deque.push_task<int>(two);
+      deque.publish_last_task();
+      REQUIRE(*deque.pop_external_task() == two);
 
 
-      deque.push_tail<int>(three);
-      REQUIRE(*deque.pop_tail() == three);
+      deque.push_task<int>(three);
+      deque.publish_last_task();
+      REQUIRE(*deque.pop_local_task() == three);
     }
     }
 
 
     SECTION("handles jumps bigger 1 correctly") {
     SECTION("handles jumps bigger 1 correctly") {
-      deque.push_tail<int>(one);
-      deque.push_tail<int>(two);
-      REQUIRE(*deque.pop_tail() == two);
-
-      deque.push_tail<int>(three);
-      deque.push_tail<int>(four);
-      REQUIRE(*deque.pop_head() == one);
-      REQUIRE(*deque.pop_head() == three);
-      REQUIRE(*deque.pop_head() == four);
+      deque.push_task<int>(one);
+      deque.publish_last_task();
+      deque.push_task<int>(two);
+      deque.publish_last_task();
+      REQUIRE(*deque.pop_local_task() == two);
+
+      deque.push_task<int>(three);
+      deque.publish_last_task();
+      deque.push_task<int>(four);
+      deque.publish_last_task();
+      REQUIRE(*deque.pop_external_task() == one);
+      REQUIRE(*deque.pop_external_task() == three);
+      REQUIRE(*deque.pop_external_task() == four);
     }
     }
 
 
     SECTION("handles stack reset 1 correctly when emptied by tail") {
     SECTION("handles stack reset 1 correctly when emptied by tail") {
-      deque.push_tail<int>(one);
-      auto state = deque.save_state();
-      deque.push_tail<int>(two);
-      REQUIRE(*deque.pop_tail() == two);
-
-      deque.release_memory_until(state);
-      REQUIRE(*deque.pop_tail() == one);
-
-      deque.push_tail<int>(three);
-      deque.push_tail<int>(four);
-      REQUIRE(*deque.pop_head() == three);
-      REQUIRE(*deque.pop_tail() == four);
+      deque.push_task<int>(one);
+      deque.publish_last_task();
+      auto state = deque.save_offset();
+      deque.push_task<int>(two);
+      deque.publish_last_task();
+      REQUIRE(*deque.pop_local_task() == two);
+
+      deque.reset_offset(state);
+      REQUIRE(*deque.pop_local_task() == one);
+
+      deque.push_task<int>(three);
+      deque.publish_last_task();
+      deque.push_task<int>(four);
+      deque.publish_last_task();
+      REQUIRE(*deque.pop_external_task() == three);
+      REQUIRE(*deque.pop_local_task() == four);
     }
     }
   }
   }
 }
 }