Draft of aggregators

Aggregators are meant to replace reduction terminals where the sole purpose
is to collect a number of inputs before passing them into the
task. An aggregator is created by wrapping the input type in a
ttg::aggregator, which will cause values to be aggregated until either
the target count is reached (if provided) or the stream size has been reached.
The aggregator is then passed directly into the task and can be iterated over.

In combination with make_tt, an aggregator is created using
`ttg::make_aggregator(inedge)` or `ttg::make_aggregator(inedge, target)`.
These calls will return a specialized Edge that allows to check at
compile-time and create aggregators in the backend.

Example:

```
ttg::Edge<int, int> edge;
auto tt = make_tt(
  [](const int& key, ttg::aggregator<int>& agg){
    for (auto&& v : agg) {
       ...
    }
  },
  ttg::edges(ttg::make_aggregator(edge)), ...);
```

Signed-off-by: Joseph Schuchart <schuchart@icl.utk.edu>

Author: devreal

tests/unit/tt.cc

@@ -145,6 +145,27 @@ namespace tt_i_iv {
   };
 }  // namespace tt_i_iv
 
+// {task_id,data} = {int, aggregator}
+namespace tt_i_i_a {
+
+  class tt : public ttg::TT<int, std::tuple<>, tt, ttg::typelist<ttg::Aggregator<int>>> {
+    using baseT = typename TT::ttT;
+
+   public:
+    tt(const typename baseT::input_edges_type &inedges, const typename baseT::output_edges_type &outedges,
+       const std::string &name)
+        : baseT(inedges, outedges, name, {"aggregator<int>"}, {}) {}
+
+    static constexpr const bool have_cuda_op = false;
+
+    void op(const int &key, const baseT::input_refs_tuple_type &data, baseT::output_terminals_type &outs) {
+      static_assert(ttg::detail::is_aggregator_v<std::decay_t<std::tuple_element_t<0, baseT::input_refs_tuple_type>>>);
+    }
+
+    ~tt() {}
+  };
+}  // namespace tt_i_i_a
+
 TEST_CASE("TemplateTask", "[core]") {
   SECTION("constructors") {
     {  // void task id, void data
@@ -249,5 +270,28 @@ TEST_CASE("TemplateTask", "[core]") {
           },
           ttg::edges(in), ttg::edges()));
     }
+    {  // nonvoid task id, aggregator input
+      ttg::Edge<int, int> in;
+      size_t count = 16;
+      CHECK_NOTHROW(std::make_unique<tt_i_i_a::tt>(ttg::edges(ttg::make_aggregator(in)), ttg::edges(), ""));
+      CHECK_NOTHROW(
+          ttg::make_tt(
+            [](const int &key, const ttg::Aggregator<int> &datum, std::tuple<> &outs) {
+              for (auto&& v : datum)
+              { }
+
+              for (const auto& v : datum)
+              { }
+            }, ttg::edges(ttg::make_aggregator(in)), ttg::edges()));
+      CHECK_NOTHROW(
+          ttg::make_tt(
+            [](const int &key, ttg::Aggregator<int> &&datum, std::tuple<> &outs) {
+              for (auto&& v : datum)
+              { }
+
+              for (const auto& v : datum)
+              { }
+            }, ttg::edges(ttg::make_aggregator(in)), ttg::edges()));
+    }
   }
 }

ttg/ttg/make_tt.h

@@ -3,6 +3,7 @@
 #ifndef TTG_MAKE_TT_H
 #define TTG_MAKE_TT_H
 
+#if 0
 namespace detail {
 
   template <typename... FromEdgeTypesT, std::size_t... I>
@@ -22,6 +23,7 @@ namespace detail {
 
   inline auto edge_base_tuple(const std::tuple<> &empty) { return empty; }
 }  // namespace detail
+#endif // 0
 
 // Class to wrap a callable with signature
 //
@@ -346,9 +348,9 @@ auto make_tt_tpl(funcT &&func, const std::tuple<ttg::Edge<keyT, input_edge_value
   static_assert(std::is_same_v<decayed_input_args_t, std::tuple<input_edge_valuesT...>>,
                 "ttg::make_tt_tpl(func, inedges, outedges): inedges value types do not match argument types of func");
 
-  auto input_edges = detail::edge_base_tuple(inedges);
+  //auto input_edges = detail::edge_base_tuple(inedges);
 
-  return std::make_unique<wrapT>(std::forward<funcT>(func), input_edges, outedges, name, innames, outnames);
+  return std::make_unique<wrapT>(std::forward<funcT>(func), inedges, outedges, name, innames, outnames);
 }
 
 /// @brief Factory function to assist in wrapping a callable with signature
@@ -423,9 +425,9 @@ auto make_tt(funcT &&func, const std::tuple<ttg::Edge<keyT, input_edge_valuesT>.
   using wrapT = typename CallableWrapTTArgsAsTypelist<funcT, have_outterm_tuple, keyT, output_terminals_type,
                                                       full_input_args_t>::type;
 
-  auto input_edges = detail::edge_base_tuple(inedges);
+  //auto input_edges = detail::edge_base_tuple(inedges);
 
-  return std::make_unique<wrapT>(std::forward<funcT>(func), input_edges, outedges, name, innames, outnames);
+  return std::make_unique<wrapT>(std::forward<funcT>(func), inedges, outedges, name, innames, outnames);
 }
 
 template <typename keyT, typename funcT, typename... input_valuesT, typename... output_edgesT>

ttg/ttg/parsec/ttg.h

@@ -27,6 +27,7 @@
 #include "ttg/util/print.h"
 #include "ttg/util/trace.h"
 #include "ttg/util/typelist.h"
+#include "ttg/aggregator.h"
 
 #include "ttg/serialization/data_descriptor.h"
 
@@ -744,7 +745,9 @@ namespace ttg_parsec {
                   "The fourth template for ttg::TT must be a ttg::typelist containing the input types");
     // create a virtual control input if the input list is empty, to be used in invoke()
     using actual_input_tuple_type = std::conditional_t<!ttg::meta::typelist_is_empty_v<input_valueTs>,
-                                                       ttg::meta::typelist_to_tuple_t<input_valueTs>, std::tuple<void>>;
+                                                          ttg::meta::remove_wrapper_tuple_t<
+                                                              ttg::meta::typelist_to_tuple_t<input_valueTs>>,
+                                                          std::tuple<void>>;
     using input_tuple_type = ttg::meta::typelist_to_tuple_t<input_valueTs>;
     static_assert(ttg::meta::is_tuple_v<output_terminalsT>,
                   "Second template argument for ttg::TT must be std::tuple containing the output terminal types");
@@ -786,6 +789,8 @@ namespace ttg_parsec {
     using input_values_tuple_type = ttg::meta::drop_void_t<ttg::meta::decayed_typelist_t<input_tuple_type>>;
     using input_refs_tuple_type = ttg::meta::drop_void_t<ttg::meta::add_glvalue_reference_tuple_t<input_tuple_type>>;
 
+    using aggregator_factory_tuple_type = ttg::meta::aggregator_factory_tuple_type_t<input_edges_type>;
+
     static constexpr int numinvals =
         std::tuple_size_v<input_refs_tuple_type>;  // number of input arguments with values (i.e. omitting the control
                                                    // input, if any)
@@ -810,6 +815,7 @@ namespace ttg_parsec {
 
     input_terminals_type input_terminals;
     output_terminalsT output_terminals;
+    aggregator_factory_tuple_type aggregator_factories;
 
    protected:
     const auto &get_output_terminals() const { return output_terminals; }
@@ -1295,10 +1301,13 @@ namespace ttg_parsec {
       task_t *task;
       auto &world_impl = world.impl();
       auto &reducer = std::get<i>(input_reducers);
+      constexpr bool is_aggregator = ttg::detail::is_aggregator_v<
+                                        typename std::tuple_element_t<i, input_edges_type>::value_type>;
       bool release = false;
       bool remove_from_hash = true;
+      bool use_hash_table = is_aggregator || (numins > 1) || reducer;
       /* If we have only one input and no reducer on that input we can skip the hash table */
-      if (numins > 1 || reducer) {
+      if (use_hash_table) {
         parsec_hash_table_lock_bucket(&tasks_table, hk);
         if (nullptr == (task = (task_t *)parsec_hash_table_nolock_find(&tasks_table, hk))) {
           task = create_new_task(key);
@@ -1310,17 +1319,52 @@ namespace ttg_parsec {
           remove_from_hash = false;
           release = true;
         }
-        parsec_hash_table_unlock_bucket(&tasks_table, hk);
+        /* we'll keep the lock for later */
       } else {
         task = create_new_task(key);
         world_impl.increment_created();
         remove_from_hash = false;
       }
 
-      if (reducer) {  // is this a streaming input? reduce the received value
+      if constexpr (is_aggregator) {
+
+        /* we use the lock to ensure mutual exclusion when inserting into the aggregator */
+
+        using aggregator_t = typename std::tuple_element_t<i, input_edges_type>::value_type;
+        aggregator_t* agg;
+        ttg_data_copy_t *agg_copy;
+        if (nullptr == (agg_copy = reinterpret_cast<ttg_data_copy_t *>(task->parsec_task.data[i].data_in))) {
+          /* create a new aggregator */
+          agg_copy = detail::create_new_datacopy(std::get<i>(aggregator_factories)());
+          task->parsec_task.data[i].data_in = agg_copy;
+        }
+        agg = reinterpret_cast<aggregator_t *>(agg_copy->device_private);
+
+        ttg_data_copy_t *copy;
+        if (nullptr != copy_in) {
+          /* register this copy with the task */
+          copy = detail::register_data_copy<std::decay_t<Value>>(copy_in, task, std::is_const_v<typename aggregator_t::value_type>);
+        } else {
+          copy = detail::create_new_datacopy(std::forward<Value>(value));
+        }
+        /* put the value into the aggregator */
+        agg->add_value(*reinterpret_cast<std::decay_t<Value> *>(copy->device_private));
+        if (agg->has_target()) {
+          /* the target has a fixed target size set */
+          release = (agg->size() == agg->target());
+        } else {
+          /* fall back to the stream size */
+          release = (agg->size() == task->stream[i].goal);
+        }
+
+        /* release the hash table bucket */
+        parsec_hash_table_unlock_bucket(&tasks_table, hk);
+
+      } else if (reducer) {  // is this a streaming input? reduce the received value
         // N.B. Right now reductions are done eagerly, without spawning tasks
         //      this means we must lock
-        parsec_hash_table_lock_bucket(&tasks_table, hk);
+
+        /* we use the lock for mutual exclusion for the reducer */
 
         if constexpr (!ttg::meta::is_void_v<valueT>) {  // for data values
           // have a value already? if not, set, otherwise reduce
@@ -1334,7 +1378,8 @@ namespace ttg_parsec {
             }
             task->parsec_task.data[i].data_in = copy;
           } else {
-            reducer(*reinterpret_cast<std::decay_t<valueT> *>(copy->device_private), value);
+            using decay_valueT = std::decay_t<valueT>;
+            reducer(*reinterpret_cast<decay_valueT *>(copy->device_private), value);
           }
         } else {
           reducer();  // even if this was a control input, must execute the reducer for possible side effects
@@ -1347,6 +1392,8 @@ namespace ttg_parsec {
         }
         parsec_hash_table_unlock_bucket(&tasks_table, hk);
       } else {
+        /* release the lock, not needed anymore */
+        parsec_hash_table_unlock_bucket(&tasks_table, hk);
         /* whether the task needs to be deferred or not */
         bool needs_deferring = false;
         if constexpr (!valueT_is_Void) {
@@ -2234,7 +2281,8 @@ namespace ttg_parsec {
    public:
     template <typename keymapT = ttg::detail::default_keymap<keyT>,
               typename priomapT = ttg::detail::default_priomap<keyT>>
-    TT(const std::string &name, const std::vector<std::string> &innames, const std::vector<std::string> &outnames,
+    TT(const input_edges_type &inedges,
+       const std::string &name, const std::vector<std::string> &innames, const std::vector<std::string> &outnames,
        ttg::World world, keymapT &&keymap_ = keymapT(), priomapT &&priomap_ = priomapT())
         : ttg::TTBase(name, numinedges, numouts)
         , world(world)
@@ -2243,7 +2291,8 @@ namespace ttg_parsec {
                      ? decltype(keymap)(ttg::detail::default_keymap<keyT>(world))
                      : decltype(keymap)(std::forward<keymapT>(keymap_)))
         , priomap(decltype(keymap)(std::forward<priomapT>(priomap_)))
-        , static_stream_goal() {
+        , static_stream_goal()
+        , aggregator_factories(ttg::meta::make_aggregator_factory_tuple(inedges)) {
       // Cannot call these in base constructor since terminals not yet constructed
       if (innames.size() != numinedges) throw std::logic_error("ttg_parsec::TT: #input names != #input terminals");
       if (outnames.size() != numouts) throw std::logic_error("ttg_parsec::TT: #output names != #output terminals");
@@ -2345,15 +2394,16 @@ namespace ttg_parsec {
               typename priomapT = ttg::detail::default_priomap<keyT>>
     TT(const std::string &name, const std::vector<std::string> &innames, const std::vector<std::string> &outnames,
        keymapT &&keymap = keymapT(ttg::default_execution_context()), priomapT &&priomap = priomapT())
-        : TT(name, innames, outnames, ttg::default_execution_context(), std::forward<keymapT>(keymap),
+        : TT(input_edges_type(), name, innames, outnames, ttg::default_execution_context(), std::forward<keymapT>(keymap),
              std::forward<priomapT>(priomap)) {}
 
     template <typename keymapT = ttg::detail::default_keymap<keyT>,
               typename priomapT = ttg::detail::default_priomap<keyT>>
     TT(const input_edges_type &inedges, const output_edges_type &outedges, const std::string &name,
        const std::vector<std::string> &innames, const std::vector<std::string> &outnames, ttg::World world,
        keymapT &&keymap_ = keymapT(), priomapT &&priomap = priomapT())
-        : TT(name, innames, outnames, world, std::forward<keymapT>(keymap_), std::forward<priomapT>(priomap)) {
+        : TT(inedges, name, innames, outnames, world, std::forward<keymapT>(keymap_), std::forward<priomapT>(priomap))
+         {
       connect_my_inputs_to_incoming_edge_outputs(std::make_index_sequence<numinedges>{}, inedges);
       connect_my_outputs_to_outgoing_edge_inputs(std::make_index_sequence<numouts>{}, outedges);
     }

ttg/ttg/terminal.h

@@ -230,13 +230,13 @@ namespace ttg {
   namespace detail {
     template <typename keyT, typename... valuesT>
     struct input_terminals_tuple {
-      using type = std::tuple<ttg::In<keyT, valuesT>...>;
+      using type = std::tuple<ttg::In<keyT, ttg::meta::remove_wrapper_t<valuesT>>...>;
     };
 
     template <typename keyT, typename... valuesT>
-    struct input_terminals_tuple<keyT, std::tuple<valuesT...>> {
-      using type = std::tuple<ttg::In<keyT, valuesT>...>;
-    };
+    struct input_terminals_tuple<keyT, std::tuple<valuesT...>>
+    : input_terminals_tuple<keyT, valuesT...>
+    { };
 
     template <typename keyT, typename... valuesT>
     using input_terminals_tuple_t = typename input_terminals_tuple<keyT, valuesT...>::type;

ttg/ttg/util/meta.h

@@ -755,6 +755,94 @@ namespace ttg {
     constexpr bool is_invocable_typelist_r_v<ReturnType, Callable, ttg::typelist<Args...>> =
         std::is_invocable_r_v<ReturnType, Callable, Args...>;
 
+    ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+    // remove any wrapper from a type, specializations provided where the wrapper is implemented (e.g., aggregator, ...)
+    ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+    template<typename T>
+    struct remove_wrapper {
+      using type = T;
+    };
+
+    template<typename T>
+    using remove_wrapper_t = typename remove_wrapper<T>::type;
+
+    template<typename T>
+    struct remove_wrapper_tuple;
+
+    template<typename...Ts>
+    struct remove_wrapper_tuple<std::tuple<Ts...>> {
+      using type = std::tuple<remove_wrapper_t<Ts>...>;
+    };
+
+    template<typename T>
+    using remove_wrapper_tuple_t = typename remove_wrapper_tuple<T>::type;
+
+    ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+    // type of a aggregator factory (returned by Edge::aggregator_factory()), with empty default
+    ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+    template<typename EdgeT, typename Enabler = void>
+    struct edge_has_aggregator_factory : std::false_type
+    { };
+
+    template<typename EdgeT>
+    struct edge_has_aggregator_factory<EdgeT, std::enable_if_t<std::is_invocable_v<decltype(std::declval<EdgeT>().aggregator_factory())>>>
+    : std::true_type
+    { };
+
+    template<typename T>
+    constexpr bool edge_has_aggregator_factory_v = edge_has_aggregator_factory<T>::value;
+
+    template<typename EdgeT, bool HasAggregatorFactory>
+    struct aggregator_factory {
+      using type = std::byte;
+    };
+
+    template<typename EdgeT>
+    struct aggregator_factory<EdgeT, true> {
+      using type = decltype(std::declval<EdgeT>().aggregator_factory());
+    };
+
+    template<typename EdgeT>
+    using aggregator_factory_t = typename aggregator_factory<EdgeT, edge_has_aggregator_factory_v<EdgeT>>::type;
+
+    template<typename T>
+    struct aggregator_factory_tuple_type;
+
+    template<typename... ValueTs>
+    struct aggregator_factory_tuple_type<ttg::typelist<ValueTs...>> {
+      using type = std::tuple<aggregator_factory_t<ValueTs>...>;
+    };
+
+    template<typename... ValueTs>
+    struct aggregator_factory_tuple_type<std::tuple<ValueTs...>> {
+      using type = std::tuple<aggregator_factory_t<ValueTs>...>;
+    };
+
+    template<typename T>
+    using aggregator_factory_tuple_type_t = typename aggregator_factory_tuple_type<T>::type;
+
+    namespace detail {
+
+      template<typename EdgeT>
+      auto make_aggregator(const EdgeT& edge) {
+        if constexpr (edge_has_aggregator_factory_v<EdgeT>) {
+          return edge.aggregator_factory();
+        } else {
+          return std::byte();
+        }
+      }
+
+      template<typename... EdgesT, std::size_t... Is>
+      auto make_aggregator_factory_tuple(const std::tuple<EdgesT...>& edges, std::index_sequence<Is...>) {
+        return std::make_tuple(make_aggregator(std::get<Is>(edges))...);
+      }
+    } // namespace detail
+    template<typename... EdgesT>
+    auto make_aggregator_factory_tuple(const std::tuple<EdgesT...>& edges) {
+      return detail::make_aggregator_factory_tuple(edges, std::make_index_sequence<sizeof...(EdgesT)>());
+    }
+
   }  // namespace meta
 }  // namespace ttg