Making sure every parcel is handled only once

STEllAR-GROUP · May 25, 2017 · d490a92 · d490a92
1 parent 4218539
commit d490a92
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 0 deletions.
diff --git a/hpx/runtime/parcelset/detail/parcel_await.hpp b/hpx/runtime/parcelset/detail/parcel_await.hpp
@@ -13,6 +13,7 @@
 #include <hpx/runtime/serialization/output_archive.hpp>
 
 #include <cstddef>
+#include <cstdint>
 #include <memory>
 #include <utility>
 #include <vector>
@@ -32,6 +33,7 @@ namespace hpx { namespace parcelset { namespace detail {
 
         put_parcel_type put_parcel_;
         std::vector<parcel> parcels_;
+        std::vector<std::uint8_t> handled_;
         int archive_flags_;
         std::size_t idx_;
     };

diff --git a/src/runtime/parcelset/detail/parcel_await.cpp b/src/runtime/parcelset/detail/parcel_await.cpp
@@ -21,12 +21,14 @@ namespace hpx { namespace parcelset { namespace detail {
         idx_(0)
     {
         parcels_.push_back(std::move(p));
+        handled_.push_back(0);
     }
 
     parcel_await::parcel_await(std::vector<parcel>&& parcels, int archive_flags,
             parcel_await::put_parcel_type pp)
       : put_parcel_(std::move(pp)),
         parcels_(std::move(parcels)),
+        handled_(parcels_.size(), 0),
         archive_flags_(archive_flags),
         idx_(0)
     {}
@@ -35,6 +37,10 @@ namespace hpx { namespace parcelset { namespace detail {
     {
         for (/*idx_*/; idx_ != parcels_.size(); ++idx_)
         {
+            // send every parcel only once
+            if (handled_[idx_])
+                continue;
+
             // collect the necessary size the serialization operation
             hpx::serialization::detail::preprocess preprocess;
             hpx::serialization::output_archive archive(preprocess, archive_flags_);
@@ -62,6 +68,7 @@ namespace hpx { namespace parcelset { namespace detail {
             parcels_[idx_].num_chunks() = archive.get_num_chunks();
             parcels_[idx_].set_split_gids(std::move(preprocess.split_gids_));
 
+            handled_[idx_] = 1;
             put_parcel_(std::move(parcels_[idx_]));
         }
     }