Adapting to HPX V1.5

config/project.cmake

@@ -448,7 +448,8 @@ endif()
 
 list(APPEND FLECSI_LIBRARY_DEPENDENCIES ${COLORING_LIBRARIES})
 
-if(FLECSI_RUNTIME_MODEL STREQUAL "mpi")
+if(FLECSI_RUNTIME_MODEL STREQUAL "mpi" OR
+   FLECSI_RUNTIME_MODEL STREQUAL "hpx")
   #------------------------------------------------------------------------------#
   # Use lazy aggregated dense field communication
   #------------------------------------------------------------------------------#
@@ -582,7 +583,7 @@ if (ENABLE_KOKKOS)
   target_link_libraries(
     FleCSI PUBLIC ${Kokkos_LIBRARIES}
   )
-  if(ENABLE_FLECSI_TUTORIAL) 
+  if(ENABLE_FLECSI_TUTORIAL)
     target_link_libraries(FleCSI-Tut PUBLIC ${Kokkos_LIBRARIES})
   endif()
 endif()

flecsi/data/hpx/data_handle_policy.h

@@ -8,6 +8,7 @@
 //! @date Initial file creation: Apr 04, 2017
 //----------------------------------------------------------------------------//
 
+#include <flecsi/execution/hpx/future.h>
 #include <flecsi/runtime/types.h>
 
 namespace flecsi {
@@ -23,6 +24,9 @@ struct hpx_data_handle_policy_t {
   size_t index_space;
   size_t data_client_hash;
   bool * ghost_is_readable;
+
+  execution::hpx_future_u<void> future;
+
 }; // class mpi_data_handle_policy_t
 
 } // namespace flecsi

flecsi/data/hpx/sparse_data_handle_policy.h

@@ -3,6 +3,9 @@
 
 #pragma once
 
+#include <flecsi/execution/hpx/future.h>
+#include <flecsi/runtime/types.h>
+
 //----------------------------------------------------------------------------//
 //! @file
 //! @date Initial file creation: Apr 04, 2017
@@ -19,6 +22,9 @@ struct hpx_sparse_data_handle_policy_t {
 
   field_id_t fid;
   bool * ghost_is_readable;
+
+  execution::hpx_future_u<void> future;
+
 }; // class mpi_sparse_data_handle_policy_t
 
 } // namespace flecsi

flecsi/execution/hpx/context_policy.cc

@@ -64,8 +64,8 @@ hpx_context_policy_t::hpx_main(void (*driver)(int, char *[]),
   int argc,
   char * argv[]) {
 
-  color_ = hpx::get_locality_id();
-  colors_ = hpx::get_num_localities(hpx::launch::sync);
+  MPI_Comm_rank(MPI_COMM_WORLD, &color_);
+  MPI_Comm_size(MPI_COMM_WORLD, &colors_);
 
   // execute user code (driver)
   (*driver)(argc, argv);

flecsi/execution/hpx/execution_policy.h

@@ -20,6 +20,7 @@
 
 #include <cinchlog.h>
 #include <functional>
+#include <stdexcept>
 #include <tuple>
 #include <unordered_map>
 #include <utility>
@@ -30,6 +31,8 @@
 #include <flecsi/execution/hpx/finalize_handles.h>
 #include <flecsi/execution/hpx/future.h>
 #include <flecsi/execution/hpx/reduction_wrapper.h>
+#include <flecsi/execution/hpx/task_add_dependencies.h>
+#include <flecsi/execution/hpx/task_collect_dependencies.h>
 #include <flecsi/execution/hpx/task_epilog.h>
 #include <flecsi/execution/hpx/task_prolog.h>
 #include <flecsi/utils/annotation.h>
@@ -145,22 +148,42 @@ struct FLECSI_EXPORT hpx_execution_policy_t {
     typename... ARGS>
   static decltype(auto) execute_task(ARGS &&... args) {
 
+#if defined(ENABLE_CALIPER)
+    auto tname = context_.function_name(TASK);
+#else
+    /* using a placeholder so we do not have to maintain
+       function_name_registry when annotations are disabled. */
+    std::string tname{""};
+#endif
+
+    using annotation = flecsi::utils::annotation;
+
     // Make a tuple from the task arguments.
     using decayed_args = utils::convert_tuple_t<ARG_TUPLE, std::decay_t>;
 
-    auto func = [function = context_t::instance().function(TASK),
-                  task_args = decayed_args(
-                    std::make_tuple(std::forward<ARGS>(args)...))]() mutable {
-      context_t & context_ = context_t::instance();
+    auto task_args = decayed_args(std::make_tuple(std::forward<ARGS>(args)...));
+
+    // collect dependencies from all arguments for this task
+    annotation::begin<annotation::execute_task_collect_dependencies>(tname);
+    task_collect_dependencies_t task_collect_dependencies;
+    task_collect_dependencies.walk(task_args);
+    annotation::end<annotation::execute_task_collect_dependencies>();
+
+    // add this task as a dependency to all arguments, if needed
+    annotation::begin<annotation::execute_task_add_dependencies>(tname);
+    task_add_dependencies_t task_add_dependencies;
+    task_add_dependencies.walk(task_args);
+    annotation::end<annotation::execute_task_add_dependencies>();
+
+    auto func = [tname, function = context_t::instance().function(TASK),
+                  task_args = std::move(task_args)](
+                  auto && dependencies) mutable {
+      // propagate exceptions
+      for(auto && f : dependencies) {
+        f.get();
+      }
 
-      using annotation = flecsi::utils::annotation;
-#if defined(ENABLE_CALIPER)
-      auto tname = context_.function_name(TASK);
-#else
-      /* using a placeholder so we do not have to maintain
-         function_name_registry when annotations are disabled. */
-      std::string tname{""};
-#endif
+      context_t & context_ = context_t::instance();
 
       annotation::begin<annotation::execute_task_prolog>(tname);
       // run task_prolog to copy ghost cells.
@@ -219,9 +242,29 @@ struct FLECSI_EXPORT hpx_execution_policy_t {
       return future;
     };
 
-    // for unwrapping o returned future
-    hpx_future_u<RETURN> future = hpx::async(std::move(func));
-    future.wait();
+    // force unwrapping of returned future
+    hpx_future_u<RETURN> future = hpx::dataflow(
+      std::move(func), std::move(task_collect_dependencies.dependencies_));
+
+    // make sure the task dependencies are triggered once this future has
+    // become ready
+    if(task_add_dependencies.has_dependencies) {
+      future.then(
+        [p = std::move(task_add_dependencies.promise)](auto && f) mutable {
+          if(f.has_exception()) {
+            try {
+              f.get(); // rethrow exception
+            }
+            catch(...) {
+              p.set_exception(std::current_exception());
+            }
+          }
+          else {
+            p.set_value();
+          }
+        });
+    }
+
     return future;
   } // execute_task
 

flecsi/execution/hpx/task_add_dependencies.h

@@ -0,0 +1,202 @@
+/*
+    @@@@@@@@  @@           @@@@@@   @@@@@@@@ @@
+   /@@/////  /@@          @@////@@ @@////// /@@
+   /@@       /@@  @@@@@  @@    // /@@       /@@
+   /@@@@@@@  /@@ @@///@@/@@       /@@@@@@@@@/@@
+   /@@////   /@@/@@@@@@@/@@       ////////@@/@@
+   /@@       /@@/@@//// //@@    @@       /@@/@@
+   /@@       @@@//@@@@@@ //@@@@@@  @@@@@@@@ /@@
+   //       ///  //////   //////  ////////  //
+
+   Copyright (c) 2020, Los Alamos National Security, LLC
+   All rights reserved.
+                                                                              */
+#pragma once
+
+/*! @file */
+
+#include <hpx/include/lcos.hpp>
+
+#include <flecsi/data/common/data_reference.h>
+#include <flecsi/data/dense_accessor.h>
+#include <flecsi/data/global_accessor.h>
+#include <flecsi/data/ragged_accessor.h>
+#include <flecsi/data/ragged_mutator.h>
+#include <flecsi/data/sparse_accessor.h>
+#include <flecsi/data/sparse_mutator.h>
+
+/*!
+ @file
+ @date Initial file creation: November 16, 2020
+ */
+
+#include <cstdint>
+#include <string>
+#include <vector>
+
+#include "mpi.h"
+
+#include <flecsi/coloring/dcrs_utils.h>
+#include <flecsi/data/data.h>
+#include <flecsi/data/dense_accessor.h>
+#include <flecsi/execution/context.h>
+
+#include "flecsi/utils/mpi_type_traits.h"
+#include <flecsi/utils/tuple_walker.h>
+#include <flecsi/utils/type_traits.h>
+
+namespace flecsi {
+namespace execution {
+
+/*!
+ The task_add_dependencies_t type can be called to walk the task args before
+ the task has run. This allows to ensure task dependencies be added to the
+ execution flow.
+
+ @ingroup execution
+ */
+
+struct task_add_dependencies_t
+  : public flecsi::utils::tuple_walker_u<task_add_dependencies_t> {
+
+  /*!
+   Construct a task_add_dependencies_t instance.
+   */
+
+  task_add_dependencies_t()
+    : has_dependencies(false), promise(), future(promise.get_future()) {}
+
+  /*!
+   FIXME: Need a description.
+
+   @tparam T                     The data type referenced by the handle.
+   @tparam EXCLUSIVE_PERMISSIONS The permissions required on the exclusive
+                                 indices of the index partition.
+   @tparam SHARED_PERMISSIONS    The permissions required on the shared
+                                 indices of the index partition.
+   @tparam GHOST_PERMISSIONS     The permissions required on the ghost
+                                 indices of the index partition.
+
+   */
+
+  template<typename T,
+    size_t EXCLUSIVE_PERMISSIONS,
+    size_t SHARED_PERMISSIONS,
+    size_t GHOST_PERMISSIONS>
+  void handle(dense_accessor<T,
+    EXCLUSIVE_PERMISSIONS,
+    SHARED_PERMISSIONS,
+    GHOST_PERMISSIONS> & a) {
+
+    auto & h = a.handle;
+    // Skip Read Only handles
+    if constexpr((SHARED_PERMISSIONS == ro) || (GHOST_PERMISSIONS == rw) ||
+                 (GHOST_PERMISSIONS == wo)) {
+      return;
+    }
+
+    h.future = future;
+    has_dependencies = true;
+  } // handle
+
+  template<typename T, size_t PERMISSIONS>
+  void handle(global_accessor_u<T, PERMISSIONS> & a) {
+    auto & h = a.handle;
+
+    // Skip Read Only handles
+    if(PERMISSIONS == ro)
+      return;
+
+    h.future = future;
+    has_dependencies = true;
+  } // handle
+
+  template<typename T,
+    size_t EXCLUSIVE_PERMISSIONS,
+    size_t SHARED_PERMISSIONS,
+    size_t GHOST_PERMISSIONS>
+  void handle(ragged_accessor<T,
+    EXCLUSIVE_PERMISSIONS,
+    SHARED_PERMISSIONS,
+    GHOST_PERMISSIONS> & a) {
+    auto & h = a.handle;
+
+    // Skip Read Only handles
+    if constexpr((SHARED_PERMISSIONS == ro) || (GHOST_PERMISSIONS == rw) ||
+                 (GHOST_PERMISSIONS == wo)) {
+      return;
+    }
+
+    h.future = future;
+    has_dependencies = true;
+  } // handle
+
+  template<typename T,
+    size_t EXCLUSIVE_PERMISSIONS,
+    size_t SHARED_PERMISSIONS,
+    size_t GHOST_PERMISSIONS>
+  void handle(sparse_accessor<T,
+    EXCLUSIVE_PERMISSIONS,
+    SHARED_PERMISSIONS,
+    GHOST_PERMISSIONS> & a) {
+    handle(a.ragged);
+  } // handle
+
+  template<typename T>
+  void handle(ragged_mutator<T> & m) {} // handle
+
+  template<typename T>
+  void handle(sparse_mutator<T> & m) {
+    handle(m.ragged);
+  }
+
+  /*!
+   Handle individual list items
+   */
+  template<typename T,
+    std::size_t N,
+    template<typename, std::size_t>
+    typename Container,
+    typename =
+      std::enable_if_t<std::is_base_of<data::data_reference_base_t, T>::value>>
+  void handle(Container<T, N> & list) {
+    for(auto & item : list) {
+      handle(item);
+    }
+  }
+
+  /*!
+   * Handle tuple of items
+   */
+
+  template<typename... Ts, size_t... I>
+  void handle_tuple_items(std::tuple<Ts...> & items,
+    std::index_sequence<I...>) {
+    (handle(std::get<I>(items)), ...);
+  }
+
+  template<typename... Ts,
+    typename = std::enable_if_t<
+      utils::are_base_of_t<data::data_reference_base_t, Ts...>::value>>
+  void handle(std::tuple<Ts...> & items) {
+    handle_tuple_items(items, std::make_index_sequence<sizeof...(Ts)>{});
+  }
+
+  /*!
+    This method is called on any task arguments that are not handles, e.g.
+    scalars or those that did not need any special handling.
+   */
+  template<typename T>
+  void handle(T &) {} // handle
+
+  /*!
+    This future is used as a dependency for all arguments, if needed
+   */
+  bool has_dependencies;
+  hpx::lcos::local::promise<void> promise;
+  hpx_future_u<void> future;
+
+}; // struct task_add_dependencies_t
+
+} // namespace execution
+} // namespace flecsi