[#139] local-parallel transform + decoupled wait/flush

pnnl · Apr 27, 2019 · 48bfaea · 48bfaea
1 parent 21c0a77
commit 48bfaea
Show file tree

Hide file tree

Showing 7 changed files with 57 additions and 14 deletions.
diff --git a/include/shad/core/impl/modifyng_sequence_ops.h b/include/shad/core/impl/modifyng_sequence_ops.h
@@ -305,6 +305,7 @@ ForwardIt2 dseq_kernel(std::false_type, ForwardIt1 first, ForwardIt1 last,
   auto begin = local_range.begin();
   auto end = local_range.end();
   auto res = std::transform(begin, end, d_first, op);
+  wait_iterator(res);
   flush_iterator(res);
   return res;
 }
@@ -314,7 +315,27 @@ ForwardIt2 dseq_kernel(std::false_type, ForwardIt1 first, ForwardIt1 last,
 template <class ForwardIt1, class ForwardIt2, class UnaryOperation>
 ForwardIt2 dpar_kernel(std::false_type, ForwardIt1 first, ForwardIt1 last,
                        ForwardIt2 d_first, UnaryOperation op) {
-  return dseq_kernel(std::false_type{}, first, last, d_first, op);
+  using itr_traits = distributed_iterator_traits<ForwardIt1>;
+  using local_iterator_t = typename itr_traits::local_iterator_type;
+
+  // local map
+  auto lrange = itr_traits::local_range(first, last);
+
+  auto map_res = local_map_init(
+      // range
+      lrange.begin(), lrange.end(),
+      // kernel
+      [&](local_iterator_t b, local_iterator_t e) {
+        auto res = std::transform(b, e, d_first, op);
+        wait_iterator(res);
+        return res;
+      },
+      // init value
+      d_first);
+
+  // local reduce
+  flush_iterator(map_res.back());
+  return map_res.back();
 }
 
 ////////////////////////////////////////////////////////////////////////////////

diff --git a/include/shad/core/impl/utils.h b/include/shad/core/impl/utils.h
@@ -46,6 +46,14 @@ void advance_output_iterator(shad::insert_iterator<T> &, It2, It2) {}
 template <typename T, typename It2>
 void advance_output_iterator(shad::buffered_insert_iterator<T>, It2, It2) {}
 
+template <typename It>
+void wait_iterator(It &) {}
+
+template <typename T>
+void wait_iterator(shad::buffered_insert_iterator<T> &it) {
+  it.wait();
+}
+
 template <typename It>
 void flush_iterator(It &) {}
 

diff --git a/include/shad/core/iterator.h b/include/shad/core/iterator.h
@@ -84,7 +84,7 @@ class insert_iterator
   typename internal_container_t::ObjectID global_id_;
   Iterator iterator_;
   internal_container_t* local_container_ptr_ = nullptr;
-  rt::Locality locality_;
+  rt::Locality locality_{rt::numLocalities()};
 };
 
 /// @brief Buffered insert iterator.
@@ -126,19 +126,25 @@ class buffered_insert_iterator : public insert_iterator<Container> {
     if (!this->local_container_ptr_ || this->locality_ != rt::thisLocality()) {
       this->locality_ = rt::thisLocality();
       this->local_container_ptr_ = Container::from_global_id(this->global_id_);
-      rt::Handle h;
-      handle_ = h;
+      handle_ = rt::Handle{};  // reset
     }
     this->local_container_ptr_->buffered_async_insert(handle_, value);
     return *this;
   }
 
+  /// @brief Wait for pending insertions to the container.
+  void wait() {
+    if (this->local_container_ptr_ != nullptr &&
+        this->locality_ == rt::thisLocality()) {
+      this->local_container_ptr_->buffered_async_wait(handle_);
+    }
+  }
+
   /// @brief Flushes pending insertions to the container.
   void flush() {
     if (this->local_container_ptr_ != nullptr &&
         this->locality_ == rt::thisLocality()) {
-      // if(!handle_.IsNull()) FIXME
-      this->local_container_ptr_->buffered_async_flush(handle_);
+      this->local_container_ptr_->buffered_async_flush();
     }
   }
 

diff --git a/include/shad/data_structures/hashmap.h b/include/shad/data_structures/hashmap.h
@@ -334,10 +334,9 @@ class Hashmap : public AbstractDataStructure<
     BufferedAsyncInsert(h, value.first, value.second);
   }
 
-  void buffered_async_flush(rt::Handle &h) {
-    rt::waitForCompletion(h);
-    WaitForBufferedInsert();
-  }
+  void buffered_async_wait(rt::Handle &h) { rt::waitForCompletion(h); }
+
+  void buffered_async_flush() { WaitForBufferedInsert(); }
 
  private:
   ObjectID oid_;

diff --git a/include/shad/data_structures/set.h b/include/shad/data_structures/set.h
@@ -238,10 +238,9 @@ class Set : public AbstractDataStructure<Set<T, ELEM_COMPARE>> {
     BufferedAsyncInsert(h, value);
   }
 
-  void buffered_async_flush(rt::Handle& h) {
-    rt::waitForCompletion(h);
-    WaitForBufferedInsert();
-  }
+  void buffered_async_wait(rt::Handle& h) { rt::waitForCompletion(h); }
+
+  void buffered_async_flush() { WaitForBufferedInsert(); }
 
  private:
   ObjectID oid_;

diff --git a/test/unit_tests/core/common.h b/test/unit_tests/core/common.h
@@ -141,6 +141,7 @@ struct create_set_<shad::unordered_set<U>, even> {
     auto res = std::make_shared<T>(size);
     shad::buffered_insert_iterator<T> ins(*res, res->end());
     for (size_t i = 0; i < size; ++i) ins = (2 * i + !even);
+    ins.wait();
     ins.flush();
     return res;
   }
@@ -163,6 +164,7 @@ struct create_map_<shad::unordered_map<U, V>, even> {
     auto res = std::make_shared<T>(size);
     shad::buffered_insert_iterator<T> ins(*res, res->begin());
     for (size_t i = 0; i < size; i++) ins = std::make_pair(i, 2 * i + !even);
+    ins.wait();
     ins.flush();
     return res;
   }
@@ -236,6 +238,7 @@ struct subseq_from_<shad::unordered_set<U>> {
       ins = *first;
       ++first;
     }
+    ins.wait();
     ins.flush();
     return res;
   }
@@ -274,6 +277,7 @@ struct subseq_from_<shad::unordered_map<U, V>> {
       ins = std::make_pair((*first).first, (*first).second);
       ++first;
     }
+    ins.wait();
     ins.flush();
     return res;
   }

diff --git a/test/unit_tests/core/iterator_test.cc b/test/unit_tests/core/iterator_test.cc
@@ -105,6 +105,7 @@ TEST(shad_uset, buffered_insert_iterator) {
   for (auto i = batch_size; i > 0; --i) {
     ins_begin = i;
   }
+  ins_begin.wait();
   ins_begin.flush();
   for (auto i = batch_size; i > 0; --i) {
     ASSERT_TRUE(shad_test_stl::find_(cnt.begin(),
@@ -117,6 +118,7 @@ TEST(shad_uset, buffered_insert_iterator) {
   for (auto i = batch_size; i > 0; --i) {
     ins_begin_ = first + i;
   }
+  ins_begin_.wait();
   ins_begin_.flush();
   for (auto i = batch_size; i > 0; --i) {
     ASSERT_TRUE(shad_test_stl::find_(cnt.begin(),
@@ -129,6 +131,7 @@ TEST(shad_uset, buffered_insert_iterator) {
   for (auto i = batch_size; i > 0; --i) {
     ins_end = first + i;
   }
+  ins_end.wait();
   ins_end.flush();
   for (auto i = batch_size; i > 0; --i) {
     ASSERT_TRUE(shad_test_stl::find_(cnt.begin(),
@@ -183,6 +186,7 @@ TEST(shad_umap, buffered_insert_iterator) {
   for (auto i = batch_size; i > 0; --i) {
     ins_begin = kv(i);
   }
+  ins_begin.wait();
   ins_begin.flush();
   for (auto i = batch_size; i > 0; --i)
     ASSERT_TRUE(shad_test_stl::find_(cnt.begin(),
@@ -194,6 +198,7 @@ TEST(shad_umap, buffered_insert_iterator) {
   for (auto i = batch_size; i > 0; --i) {
     ins_begin_ = kv(first + i);
   }
+  ins_begin_.wait();
   ins_begin_.flush();
   for (auto i = batch_size; i > 0; --i)
     ASSERT_TRUE(shad_test_stl::find_(cnt.begin(),
@@ -205,6 +210,7 @@ TEST(shad_umap, buffered_insert_iterator) {
   for (auto i = batch_size; i > 0; --i) {
     ins_end = kv(first + i);
   }
+  ins_end.wait();
   ins_end.flush();
   for (auto i = batch_size; i > 0; --i)
     ASSERT_TRUE(shad_test_stl::find_(cnt.begin(), cnt.end(), kv(first + i)) !=