feat(core): add resource-limited execution to graph nodes

CMakeLists.txt

@@ -157,10 +157,13 @@ set(CMAKE_INSTALL_RPATH_USE_LINK_PATH ON)
 # add_compile_options(-fprofile-instr-generate -fcoverage-mapping)
 
 list(PREPEND CMAKE_MODULE_PATH ${PROJECT_SOURCE_DIR}/Modules)
+include(${PROJECT_SOURCE_DIR}/Modules/private/PhlexTBB.cmake)
 
 # Dependencies required by public API
 find_package(Boost REQUIRED COMPONENTS json EXPORT)
 find_package(TBB REQUIRED EXPORT)
+phlex_check_tbb_resource_limiting()
+
 find_package(fmt REQUIRED EXPORT)
 find_package(spdlog REQUIRED EXPORT)
 find_package(Microsoft.GSL REQUIRED EXPORT)

Modules/private/PhlexTBB.cmake

@@ -0,0 +1,39 @@
+# Provides phlex_check_tbb_resource_limiting(), which verifies that the
+# TBB preview resource-limiting API required by Phlex is available.
+
+include_guard()
+
+include(CheckCXXSourceCompiles)
+
+function(phlex_check_tbb_resource_limiting)
+  set(_phlex_required_libraries_save ${CMAKE_REQUIRED_LIBRARIES})
+  set(_phlex_required_includes_save ${CMAKE_REQUIRED_INCLUDES})
+  set(_phlex_required_flags_save "${CMAKE_REQUIRED_FLAGS}")
+
+  get_target_property(_phlex_tbb_includes TBB::tbb INTERFACE_INCLUDE_DIRECTORIES)
+  if(_phlex_tbb_includes)
+    set(CMAKE_REQUIRED_INCLUDES ${_phlex_tbb_includes})
+  endif()
+  set(CMAKE_REQUIRED_LIBRARIES TBB::tbb)
+  set(CMAKE_REQUIRED_FLAGS "-std=c++${CMAKE_CXX_STANDARD}")
+
+  check_cxx_source_compiles(
+    "
+    #define TBB_PREVIEW_FLOW_GRAPH_RESOURCE_LIMITING 1
+    #include <oneapi/tbb/flow_graph.h>
+
+    using type = oneapi::tbb::flow::resource_limiter<int>;
+
+    int main() {}
+    "
+    HAVE_TBB_RESOURCE_LIMITING
+  )
+
+  set(CMAKE_REQUIRED_LIBRARIES ${_phlex_required_libraries_save})
+  set(CMAKE_REQUIRED_INCLUDES ${_phlex_required_includes_save})
+  set(CMAKE_REQUIRED_FLAGS "${_phlex_required_flags_save}")
+
+  if(NOT HAVE_TBB_RESOURCE_LIMITING)
+    message(FATAL_ERROR "Phlex requires TBB with flow::resource_limiter support")
+  endif()
+endfunction()

phlex/core/CMakeLists.txt

@@ -59,12 +59,14 @@ install(
     message.hpp
     multilayer_join_node.hpp
     index_router.hpp
+    node_builder.hpp
     node_catalog.hpp
     product_selector.hpp
     products_consumer.hpp
     provider_node.hpp
     registrar.hpp
     registration_api.hpp
+    resource_api.hpp
     source.hpp
     upstream_predicates.hpp
   DESTINATION include/phlex/core
@@ -79,6 +81,7 @@ install(
   DESTINATION include/phlex/core/detail
 )
 target_include_directories(phlex_core PRIVATE ${PROJECT_SOURCE_DIR})
+target_compile_definitions(phlex_core PUBLIC TBB_PREVIEW_FLOW_GRAPH_RESOURCE_LIMITING=1)
 phlex_apply_symbol_visibility(phlex_core)
 phlex_apply_optimizations(phlex_core)
 
@@ -96,6 +99,10 @@ phlex_make_internal_library(
   PUBLIC TBB::tbb phlex::metaprogramming phlex_model_internal phlex_utilities_internal
   PRIVATE Boost::json spdlog::spdlog
 )
+target_compile_definitions(
+  phlex_core_internal
+  INTERFACE TBB_PREVIEW_FLOW_GRAPH_RESOURCE_LIMITING=1
+)
 add_library(phlex::core_internal ALIAS phlex_core_internal)
 
 # Interface library

phlex/core/declared_fold.hpp

@@ -10,6 +10,7 @@
 #include "phlex/core/fwd.hpp"
 #include "phlex/core/input_arguments.hpp"
 #include "phlex/core/message.hpp"
+#include "phlex/core/node_builder.hpp"
 #include "phlex/core/product_selector.hpp"
 #include "phlex/core/products_consumer.hpp"
 #include "phlex/model/algorithm_name.hpp"
@@ -73,15 +74,24 @@ namespace phlex::detail {
 
   // =====================================================================================
 
-  template <typename AlgorithmBits, typename InitTuple>
+  template <typename AlgorithmBits, typename InitTuple, typename... Resources>
   class fold_node : public declared_fold {
     using all_parameter_types = AlgorithmBits::input_parameter_types;
     using result_type = std::decay_t<std::tuple_element_t<0, all_parameter_types>>;
-    using input_parameter_types = skip_first_type<all_parameter_types>; // Skip fold object
-    static constexpr auto num_inputs = std::tuple_size_v<input_parameter_types>;
+    using non_result_parameter_types = skip_first_type<all_parameter_types>;
+    static constexpr std::size_t num_resources = sizeof...(Resources);
+    static constexpr std::size_t num_input_products =
+      std::tuple_size_v<non_result_parameter_types> - num_resources;
+    using input_product_types =
+      boost::mp11::mp_take_c<non_result_parameter_types, num_input_products>;
 
     static constexpr std::size_t num_outputs = 1; // hard-coded for now
     using function_t = AlgorithmBits::bound_type;
+    using builder = node_builder<accumulator_with_messages<result_type, num_input_products>,
+                                 function_t,
+                                 no_outputs_t,
+                                 std::tuple<Resources...>>;
+    using node_t = typename builder::node_t;
 
   public:
     fold_node(phlex::experimental::algorithm_name algo_name,
@@ -92,7 +102,8 @@ namespace phlex::detail {
               InitTuple initializer,
               product_selectors input_products,
               std::vector<std::string> output,
-              std::string partition_layer) :
+              std::string partition_layer,
+              resource_catalog& resources) :
       declared_fold{std::move(algo_name),
                     std::move(predicates),
                     std::move(input_products),
@@ -105,17 +116,20 @@ namespace phlex::detail {
             this->output(),
             make_initializer<result_type>(
               std::move(initializer), std::make_index_sequence<std::tuple_size_v<InitTuple>>{})},
-      fold_{g,
-            concurrency,
-            [this, ft = alg.release_algorithm()](
-              accumulator_with_messages<result_type, num_inputs> const& accum_with_msgs) {
-              std::size_t const partition_hash = apply_fold(ft, accum_with_msgs);
-
-              ++calls_;
-
-              join_.notify_result_repeater_port().try_put(partition_hash);
-              return tbb::flow::continue_msg{};
-            }}
+      fold_{builder::make(
+        g,
+        concurrency,
+        resources,
+        alg.release_algorithm(),
+        [this](function_t const& ft,
+               accumulator_with_messages<result_type, num_input_products> const& accum_with_msgs,
+               auto*... resource_tokens) {
+          std::size_t const partition_hash = apply_fold(ft, accum_with_msgs, resource_tokens...);
+
+          ++calls_;
+
+          join_.notify_result_repeater_port().try_put(partition_hash);
+        })}
     {
       make_edge(join_, fold_);
     }
@@ -138,23 +152,24 @@ namespace phlex::detail {
 
     std::size_t apply_fold(
       function_t const& ft,
-      accumulator_with_messages<result_type, num_inputs> const& accum_with_msgs)
+      accumulator_with_messages<result_type, num_input_products> const& accum_with_msgs,
+      auto*... resource_tokens)
     {
       // We have to do awkward index management until we can use structured bindings with packs.
       auto& accumulator = std::get<0>(accum_with_msgs);
       [&]<std::size_t... Is>(std::index_sequence<Is...>) {
         accumulator.partial_result->call(
-          ft, std::get<Is>(input_).retrieve(std::get<Is + 1>(accum_with_msgs))...);
-      }(std::make_index_sequence<num_inputs>{});
+          ft,
+          std::get<Is>(input_).retrieve(std::get<Is + 1>(accum_with_msgs))...,
+          resource_tokens...);
+      }(std::make_index_sequence<num_input_products>{});
       return accumulator.index->hash();
     }
 
-    input_retriever_types<input_parameter_types> input_{input_arguments<input_parameter_types>()};
+    input_retriever_types<input_product_types> input_{input_arguments<input_product_types>()};
     product_specifications output_;
-    fold_join_node<result_type, num_inputs> join_;
-    tbb::flow::function_node<accumulator_with_messages<result_type, num_inputs>,
-                             tbb::flow::continue_msg>
-      fold_;
+    fold_join_node<result_type, num_input_products> join_;
+    node_t fold_;
     std::atomic<std::size_t> calls_;
   };
 }

phlex/core/declared_observer.hpp

@@ -8,8 +8,10 @@
 #include "phlex/core/input_arguments.hpp"
 #include "phlex/core/message.hpp"
 #include "phlex/core/multilayer_join_node.hpp"
+#include "phlex/core/node_builder.hpp"
 #include "phlex/core/product_selector.hpp"
 #include "phlex/core/products_consumer.hpp"
+#include "phlex/core/resource_api.hpp"
 #include "phlex/metaprogramming/type_deduction.hpp"
 #include "phlex/model/algorithm_name.hpp"
 #include "phlex/model/data_cell_index.hpp"
@@ -44,11 +46,15 @@ namespace phlex::detail {
 
   // =====================================================================================
 
-  template <typename AlgorithmBits>
+  template <typename AlgorithmBits, typename... Resources>
   class observer_node : public declared_observer {
-    using input_args = AlgorithmBits::input_parameter_types;
     using function_t = AlgorithmBits::bound_type;
-    static constexpr auto num_inputs = AlgorithmBits::number_inputs;
+    static constexpr auto num_resources = sizeof...(Resources);
+    static constexpr auto num_products = AlgorithmBits::number_inputs - num_resources;
+    using input_product_types = AlgorithmBits::template input_parameters<num_products>;
+    using builder =
+      node_builder<messages_t<num_products>, function_t, no_outputs_t, std::tuple<Resources...>>;
+    using node_t = typename builder::node_t;
 
   public:
     static constexpr auto number_output_products = 0;
@@ -59,52 +65,58 @@ namespace phlex::detail {
                   std::vector<std::string> predicates,
                   tbb::flow::graph& g,
                   AlgorithmBits alg,
-                  product_selectors input_products) :
+                  product_selectors input_products,
+                  resource_catalog& resources) :
       declared_observer{std::move(algo_name), std::move(predicates), std::move(input_products)},
-      join_{make_join_or_none<num_inputs>(g, name().to_string(), layers())},
-      observer_{g,
-                concurrency,
-                [this, ft = alg.release_algorithm()](
-                  messages_t<num_inputs> const& messages) -> oneapi::tbb::flow::continue_msg {
-                  call(ft, messages, std::make_index_sequence<num_inputs>{});
-                  ++calls_;
-                  return {};
-                }}
+      join_{make_join_or_none<num_products>(g, name().to_string(), layers())},
+      observer_{builder::make(
+        g,
+        concurrency,
+        resources,
+        alg.release_algorithm(),
+        [this](function_t const& ft,
+               messages_t<num_products> const& messages,
+               auto*... resource_tokens) {
+          call(ft, messages, std::make_index_sequence<num_products>{}, resource_tokens...);
+          ++calls_;
+        })}
     {
-      if constexpr (num_inputs > 1ull) {
+      if constexpr (num_products > 1ull) {
         make_edge(join_, observer_);
       }
     }
 
   private:
     tbb::flow::receiver<message>& port_for(product_selector const& input_product) override
     {
-      return receiver_for<num_inputs>(join_, input(), input_product, observer_);
+      return receiver_for<num_products>(join_, input(), input_product, observer_);
     }
 
     std::vector<tbb::flow::receiver<message>*> ports() override
     {
-      return input_ports<num_inputs>(join_, observer_);
+      return input_ports<num_products>(join_, observer_);
     }
 
     template <std::size_t... Is>
     void call(function_t const& ft,
-              messages_t<num_inputs> const& messages,
-              std::index_sequence<Is...>)
+              messages_t<num_products> const& messages,
+              std::index_sequence<Is...>,
+              auto*... resource_tokens)
     {
-      if constexpr (num_inputs == 1ull) {
-        std::invoke(ft, std::get<Is>(input_).retrieve(messages)...);
+      if constexpr (num_products == 1ull) {
+        std::invoke(ft, std::get<Is>(input_).retrieve(messages)..., resource_tokens...);
       } else {
-        std::invoke(ft, std::get<Is>(input_).retrieve(std::get<Is>(messages))...);
+        std::invoke(
+          ft, std::get<Is>(input_).retrieve(std::get<Is>(messages))..., resource_tokens...);
       }
     }
 
     named_index_ports index_ports() final { return join_.index_ports(); }
     std::size_t num_calls() const final { return calls_.load(); }
 
-    input_retriever_types<input_args> input_{input_arguments<input_args>()};
-    join_or_none_t<num_inputs> join_;
-    tbb::flow::function_node<messages_t<num_inputs>> observer_;
+    input_retriever_types<input_product_types> input_{input_arguments<input_product_types>()};
+    join_or_none_t<num_products> join_;
+    node_t observer_;
     std::atomic<std::size_t> calls_;
   };
 }