diff --git a/CMakeLists.txt b/CMakeLists.txt index 332669d09..72cda4921 100644 --- a/CMakeLists.txt +++ b/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) diff --git a/Modules/private/PhlexTBB.cmake b/Modules/private/PhlexTBB.cmake new file mode 100644 index 000000000..8ec81b9ef --- /dev/null +++ b/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 + + using type = oneapi::tbb::flow::resource_limiter; + + 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() diff --git a/phlex/core/CMakeLists.txt b/phlex/core/CMakeLists.txt index b35c04c0f..1ec53f860 100644 --- a/phlex/core/CMakeLists.txt +++ b/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 diff --git a/phlex/core/declared_fold.hpp b/phlex/core/declared_fold.hpp index c7cfa4593..77826321e 100644 --- a/phlex/core/declared_fold.hpp +++ b/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 + template class fold_node : public declared_fold { using all_parameter_types = AlgorithmBits::input_parameter_types; using result_type = std::decay_t>; - using input_parameter_types = skip_first_type; // Skip fold object - static constexpr auto num_inputs = std::tuple_size_v; + using non_result_parameter_types = skip_first_type; + static constexpr std::size_t num_resources = sizeof...(Resources); + static constexpr std::size_t num_input_products = + std::tuple_size_v - num_resources; + using input_product_types = + boost::mp11::mp_take_c; static constexpr std::size_t num_outputs = 1; // hard-coded for now using function_t = AlgorithmBits::bound_type; + using builder = node_builder, + function_t, + no_outputs_t, + std::tuple>; + 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 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( std::move(initializer), std::make_index_sequence>{})}, - fold_{g, - concurrency, - [this, ft = alg.release_algorithm()]( - accumulator_with_messages 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 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 const& accum_with_msgs) + accumulator_with_messages 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::index_sequence) { accumulator.partial_result->call( - ft, std::get(input_).retrieve(std::get(accum_with_msgs))...); - }(std::make_index_sequence{}); + ft, + std::get(input_).retrieve(std::get(accum_with_msgs))..., + resource_tokens...); + }(std::make_index_sequence{}); return accumulator.index->hash(); } - input_retriever_types input_{input_arguments()}; + input_retriever_types input_{input_arguments()}; product_specifications output_; - fold_join_node join_; - tbb::flow::function_node, - tbb::flow::continue_msg> - fold_; + fold_join_node join_; + node_t fold_; std::atomic calls_; }; } diff --git a/phlex/core/declared_observer.hpp b/phlex/core/declared_observer.hpp index 3f00ffd9f..1b8085a88 100644 --- a/phlex/core/declared_observer.hpp +++ b/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 + template 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; + using builder = + node_builder, function_t, no_outputs_t, std::tuple>; + using node_t = typename builder::node_t; public: static constexpr auto number_output_products = 0; @@ -59,19 +65,23 @@ namespace phlex::detail { std::vector 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(g, name().to_string(), layers())}, - observer_{g, - concurrency, - [this, ft = alg.release_algorithm()]( - messages_t const& messages) -> oneapi::tbb::flow::continue_msg { - call(ft, messages, std::make_index_sequence{}); - ++calls_; - return {}; - }} + join_{make_join_or_none(g, name().to_string(), layers())}, + observer_{builder::make( + g, + concurrency, + resources, + alg.release_algorithm(), + [this](function_t const& ft, + messages_t const& messages, + auto*... resource_tokens) { + call(ft, messages, std::make_index_sequence{}, resource_tokens...); + ++calls_; + })} { - if constexpr (num_inputs > 1ull) { + if constexpr (num_products > 1ull) { make_edge(join_, observer_); } } @@ -79,32 +89,34 @@ namespace phlex::detail { private: tbb::flow::receiver& port_for(product_selector const& input_product) override { - return receiver_for(join_, input(), input_product, observer_); + return receiver_for(join_, input(), input_product, observer_); } std::vector*> ports() override { - return input_ports(join_, observer_); + return input_ports(join_, observer_); } template void call(function_t const& ft, - messages_t const& messages, - std::index_sequence) + messages_t const& messages, + std::index_sequence, + auto*... resource_tokens) { - if constexpr (num_inputs == 1ull) { - std::invoke(ft, std::get(input_).retrieve(messages)...); + if constexpr (num_products == 1ull) { + std::invoke(ft, std::get(input_).retrieve(messages)..., resource_tokens...); } else { - std::invoke(ft, std::get(input_).retrieve(std::get(messages))...); + std::invoke( + ft, std::get(input_).retrieve(std::get(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_{input_arguments()}; - join_or_none_t join_; - tbb::flow::function_node> observer_; + input_retriever_types input_{input_arguments()}; + join_or_none_t join_; + node_t observer_; std::atomic calls_; }; } diff --git a/phlex/core/declared_predicate.hpp b/phlex/core/declared_predicate.hpp index ab63465c8..7657d6fb1 100644 --- a/phlex/core/declared_predicate.hpp +++ b/phlex/core/declared_predicate.hpp @@ -9,8 +9,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" @@ -49,11 +51,17 @@ namespace phlex::detail { // ===================================================================================== - template + template class predicate_node : public declared_predicate { - 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; + using builder = node_builder, + function_t, + std::tuple, + std::tuple>; + using node_t = typename builder::node_t; public: static constexpr auto number_output_products = 0ull; @@ -64,22 +72,28 @@ namespace phlex::detail { std::vector predicates, tbb::flow::graph& g, AlgorithmBits alg, - product_selectors input_products) : + product_selectors input_products, + resource_catalog& resources) : declared_predicate{std::move(algo_name), std::move(predicates), std::move(input_products)}, - join_{make_join_or_none(g, name().to_string(), layers())}, - predicate_{g, - concurrency, - [this, ft = alg.release_algorithm()]( - messages_t const& messages) -> predicate_result { - auto const& msg = most_derived(messages); - auto const& [store, message_id] = std::tie(msg.store, msg.id); - - bool const rc = call(ft, messages, std::make_index_sequence{}); - ++calls_; - return {message_id, rc}; - }} + join_{make_join_or_none(g, name().to_string(), layers())}, + predicate_{builder::make( + g, + concurrency, + resources, + alg.release_algorithm(), + [this](function_t const& ft, + messages_t const& messages, + auto*... resource_tokens) -> predicate_result { + auto const& msg = most_derived(messages); + auto const& [store, message_id] = std::tie(msg.store, msg.id); + + bool const rc = + call(ft, messages, std::make_index_sequence{}, resource_tokens...); + ++calls_; + return {message_id, rc}; + })} { - if constexpr (num_inputs > 1ull) { + if constexpr (num_products > 1ull) { make_edge(join_, predicate_); } } @@ -87,33 +101,38 @@ namespace phlex::detail { private: tbb::flow::receiver& port_for(product_selector const& input_product) override { - return receiver_for(join_, input(), input_product, predicate_); + return receiver_for(join_, input(), input_product, predicate_); + } + tbb::flow::sender& sender() override + { + return builder::output_port(predicate_); } - tbb::flow::sender& sender() override { return predicate_; } std::vector*> ports() override { - return input_ports(join_, predicate_); + return input_ports(join_, predicate_); } template bool call(function_t const& ft, - messages_t const& messages, - std::index_sequence) + messages_t const& messages, + std::index_sequence, + auto*... resource_tokens) { - if constexpr (num_inputs == 1ull) { - return std::invoke(ft, std::get(input_).retrieve(messages)...); + if constexpr (num_products == 1ull) { + return std::invoke(ft, std::get(input_).retrieve(messages)..., resource_tokens...); } else { - return std::invoke(ft, std::get(input_).retrieve(std::get(messages))...); + return std::invoke( + ft, std::get(input_).retrieve(std::get(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_{input_arguments()}; - join_or_none_t join_; - tbb::flow::function_node, predicate_result> predicate_; + input_retriever_types input_{input_arguments()}; + join_or_none_t join_; + node_t predicate_; std::atomic calls_; }; diff --git a/phlex/core/declared_transform.hpp b/phlex/core/declared_transform.hpp index b5fd24c71..87cf8b997 100644 --- a/phlex/core/declared_transform.hpp +++ b/phlex/core/declared_transform.hpp @@ -11,8 +11,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" @@ -33,6 +35,7 @@ #include #include #include +#include #include #include @@ -55,13 +58,19 @@ namespace phlex::detail { // ===================================================================================== - template + template class transform_node : public declared_transform { using function_t = AlgorithmBits::bound_type; - using input_parameter_types = AlgorithmBits::input_parameter_types; - static constexpr auto num_inputs = AlgorithmBits::number_inputs; - static constexpr auto num_outputs = number_output_objects; + static constexpr auto num_resources = sizeof...(Resources); + static constexpr auto num_products = AlgorithmBits::number_inputs - num_resources; + static constexpr auto num_outputs = AlgorithmBits::number_outputs; + using input_product_types = AlgorithmBits::template input_parameters; + using builder = node_builder, + function_t, + std::tuple, + std::tuple>; + using node_t = typename builder::node_t; public: using node_ptr_type = declared_transform_ptr; @@ -73,20 +82,26 @@ namespace phlex::detail { tbb::flow::graph& g, AlgorithmBits alg, product_selectors input_products, - std::vector output) : + std::vector output, + resource_catalog& resources) : declared_transform{std::move(algo_name), std::move(predicates), std::move(input_products)}, output_{ to_product_specifications(name(), std::move(output), make_output_type_ids())}, - join_{make_join_or_none(g, name().to_string(), layers())}, - transform_{ + join_{make_join_or_none(g, name().to_string(), layers())}, + transform_{builder::make( g, concurrency, - [this, ft = alg.release_algorithm()](messages_t const& messages) -> message { + resources, + alg.release_algorithm(), + [this](function_t const& ft, + messages_t const& messages, + auto*... resource_tokens) -> message { using namespace phlex::experimental::detail; auto const& msg = most_derived(messages); auto const& [store, message_id] = std::tie(msg.store, msg.id); - auto result = call(ft, messages, std::make_index_sequence{}); + auto result = + call(ft, messages, std::make_index_sequence{}, resource_tokens...); ++calls_; ++product_count_[store->index()->layer_hash()]; @@ -96,9 +111,9 @@ namespace phlex::detail { store->index(), name(), std::move(new_products)); return {.store = std::move(new_store), .id = message_id}; - }} + })} { - if constexpr (num_inputs > 1ull) { + if constexpr (num_products > 1ull) { make_edge(join_, transform_); } } @@ -106,26 +121,28 @@ namespace phlex::detail { private: tbb::flow::receiver& port_for(product_selector const& input_product) override { - return receiver_for(join_, input(), input_product, transform_); + return receiver_for(join_, input(), input_product, transform_); } std::vector*> ports() override { - return input_ports(join_, transform_); + return input_ports(join_, transform_); } - tbb::flow::sender& output_port() override { return transform_; } + tbb::flow::sender& output_port() override { return builder::output_port(transform_); } product_specifications const& output() const override { return output_; } template auto call(function_t const& ft, - messages_t const& messages, - std::index_sequence) + messages_t const& messages, + std::index_sequence, + auto*... resource_tokens) { - if constexpr (num_inputs == 1ull) { - return std::invoke(ft, std::get(input_).retrieve(messages)...); + if constexpr (num_products == 1ull) { + return std::invoke(ft, std::get(input_).retrieve(messages)..., resource_tokens...); } else { - return std::invoke(ft, std::get(input_).retrieve(std::get(messages))...); + return std::invoke( + ft, std::get(input_).retrieve(std::get(messages))..., resource_tokens...); } } @@ -140,10 +157,10 @@ namespace phlex::detail { return result; } - input_retriever_types input_{input_arguments()}; + input_retriever_types input_{input_arguments()}; product_specifications output_; - join_or_none_t join_; - tbb::flow::function_node, message> transform_; + join_or_none_t join_; + node_t transform_; std::atomic calls_; tbb::concurrent_unordered_map> product_count_; }; diff --git a/phlex/core/declared_unfold.hpp b/phlex/core/declared_unfold.hpp index aa34a6596..6b0f94403 100644 --- a/phlex/core/declared_unfold.hpp +++ b/phlex/core/declared_unfold.hpp @@ -8,6 +8,7 @@ #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/products_consumer.hpp" #include "phlex/model/algorithm_name.hpp" #include "phlex/model/data_cell_index.hpp" @@ -80,11 +81,18 @@ namespace phlex::detail { // ===================================================================================== - template + template class unfold_node : public declared_unfold { - using input_args = constructor_parameter_types; - static constexpr std::size_t num_inputs = std::tuple_size_v; + using all_input_args = constructor_parameter_types; + static constexpr std::size_t num_inputs = std::tuple_size_v; + using input_args = all_input_args; static constexpr std::size_t num_outputs = number_output_objects; + using function_t = std::decay_t; + using builder = node_builder, + function_t, + multifunction_outputs, + std::tuple>; + using node_t = typename builder::node_t; public: unfold_node(phlex::experimental::algorithm_name algo_name, @@ -95,7 +103,8 @@ namespace phlex::detail { Unfold&& unfold, product_selectors input_products, std::vector output_product_suffixes, - std::string child_layer_name) : + std::string child_layer_name, + resource_catalog& resources) : declared_unfold{std::move(algo_name), std::move(predicates), std::move(input_products), @@ -104,20 +113,29 @@ namespace phlex::detail { std::move(output_product_suffixes), make_type_ids>>())}, join_{make_join_or_none(g, name().to_string(), layers())}, - unfold_{g, - concurrency, - [this, p = std::move(predicate), ufold = std::move(unfold)]( - messages_t const& messages, auto& outputs) { - auto const& msg = most_derived(messages); - auto const& store = msg.store; - - generator gen{store, name(), child_layer()}; - call( - p, ufold, store->index(), gen, messages, std::make_index_sequence{}); - std::get<2>(outputs).try_put({.index = store->index(), - .layer_hash = gen.child_layer_hash(), - .count = gen.child_count()}); - }} + unfold_{builder::make(g, + concurrency, + resources, + std::move(unfold), + [this, p = std::move(predicate)](function_t const& ufold, + messages_t const& messages, + auto& outputs, + auto*... resource_tokens) { + auto const& msg = most_derived(messages); + auto const& store = msg.store; + + generator gen{store, name(), child_layer()}; + call(p, + ufold, + store->index(), + gen, + messages, + std::make_index_sequence{}, + resource_tokens...); + std::get<2>(outputs).try_put({.index = store->index(), + .layer_hash = gen.child_layer_hash(), + .count = gen.child_count()}); + })} { if constexpr (num_inputs > 1ull) { make_edge(join_, unfold_); @@ -154,7 +172,8 @@ namespace phlex::detail { data_cell_index_ptr const& unfolded_id, generator& g, messages_t const& messages, - std::index_sequence) + std::index_sequence, + auto*... resource_tokens) { ++calls_; Object obj = [this, &messages]() { @@ -169,12 +188,15 @@ namespace phlex::detail { while (std::invoke(predicate, obj, running_value)) { products new_products{num_outputs}; auto new_id = unfolded_id->make_child(child_layer(), counter); - if constexpr (requires { std::invoke(unfold, obj, running_value, *new_id); }) { - auto [next_value, prods] = std::invoke(unfold, obj, running_value, *new_id); + if constexpr (requires { + std::invoke(unfold, obj, running_value, *new_id, resource_tokens...); + }) { + auto [next_value, prods] = + std::invoke(unfold, obj, running_value, *new_id, resource_tokens...); new_products.add_all(output_, std::move(prods)); running_value = next_value; } else { - auto [next_value, prods] = std::invoke(unfold, obj, running_value); + auto [next_value, prods] = std::invoke(unfold, obj, running_value, resource_tokens...); new_products.add_all(output_, std::move(prods)); running_value = next_value; } @@ -194,9 +216,7 @@ namespace phlex::detail { input_retriever_types input_{input_arguments()}; product_specifications output_; join_or_none_t join_; - tbb::flow::multifunction_node, - std::tuple> - unfold_; + node_t unfold_; std::atomic msg_counter_; // Is this sufficient? Probably not. std::atomic calls_; std::atomic product_count_; diff --git a/phlex/core/framework_graph.hpp b/phlex/core/framework_graph.hpp index d01670d19..126469b27 100644 --- a/phlex/core/framework_graph.hpp +++ b/phlex/core/framework_graph.hpp @@ -8,6 +8,7 @@ #include "phlex/core/index_router.hpp" #include "phlex/core/message.hpp" #include "phlex/core/node_catalog.hpp" +#include "phlex/core/resource_api.hpp" #include "phlex/driver.hpp" #include "phlex/model/data_cell_tracker.hpp" #include "phlex/model/data_layer_hierarchy.hpp" @@ -31,6 +32,7 @@ #include #include #include +#include #include #include @@ -71,7 +73,7 @@ namespace phlex::detail { module_graph_proxy module_proxy(configuration const& config) { - return {config, graph_, nodes_, registration_errors_}; + return {config, graph_, nodes_, registration_errors_, resources_}; } source_bundle source_proxy(configuration const& config) @@ -79,6 +81,7 @@ namespace phlex::detail { return {.config = config, .graph = graph_, .nodes = nodes_, + .resources = resources_, .registration_errors = registration_errors_}; } @@ -146,6 +149,13 @@ namespace phlex::detail { return make_glue().template add_source(name, std::forward(args)...); } + template + requires(!std::is_const_v && std::constructible_from) + void add_resource(Args&&... args) + { + resources_.template add(std::forward(args)...); + } + template glue make(Args&&... args) { @@ -185,7 +195,7 @@ namespace phlex::detail { if constexpr (is_bound_object && Construct) { bound_object = std::make_shared(std::forward(args)...); } - return {graph_, nodes_, std::move(bound_object), registration_errors_}; + return {graph_, nodes_, std::move(bound_object), registration_errors_, resources_}; } void run(); @@ -200,10 +210,12 @@ namespace phlex::detail { resource_usage graph_resource_usage_; max_allowed_parallelism parallelism_limit_; fixed_hierarchy fixed_hierarchy_; + // The graph_ object uses the filters_, nodes_, resources_, and hierarchy_ objects implicitly. + // These must be declared before graph_ so that they outlive it during destruction. data_layer_hierarchy hierarchy_{}; + resource_catalog resources_; node_catalog nodes_; std::map filters_; - // The graph_ object uses the filters_, nodes_, and hierarchy_ objects implicitly. tbb::flow::graph graph_{}; std::optional driver_; std::vector registration_errors_; diff --git a/phlex/core/glue.hpp b/phlex/core/glue.hpp index 88ea9d653..a99418662 100644 --- a/phlex/core/glue.hpp +++ b/phlex/core/glue.hpp @@ -7,6 +7,7 @@ #include "phlex/core/concepts.hpp" #include "phlex/core/registrar.hpp" #include "phlex/core/registration_api.hpp" +#include "phlex/core/resource_api.hpp" #include "phlex/core/source.hpp" #include "phlex/metaprogramming/delegate.hpp" @@ -48,8 +49,14 @@ namespace phlex::detail { node_catalog& nodes, std::shared_ptr bound_obj, std::vector& errors, + resource_catalog& resources, configuration const* config = nullptr) : - graph_{g}, nodes_{nodes}, bound_obj_{std::move(bound_obj)}, errors_{errors}, config_{config} + graph_{g}, + nodes_{nodes}, + bound_obj_{std::move(bound_obj)}, + errors_{errors}, + resources_{resources}, + config_{config} { } @@ -70,6 +77,7 @@ namespace phlex::detail { graph_, nodes_, errors_, + resources_, std::move(partition), std::forward(init_args)...}; } @@ -82,8 +90,14 @@ namespace phlex::detail { concurrency c) { internal::verify_name(name, config_); - return make_registration( - config_, name, algorithm_bits{bound_obj_, std::move(f)}, c, graph_, nodes_, errors_); + return make_registration(config_, + name, + algorithm_bits{bound_obj_, std::move(f)}, + c, + graph_, + nodes_, + errors_, + resources_); } // 'f' is a by-value sink: it is moved into algorithm_bits. The clang-tidy @@ -94,8 +108,14 @@ namespace phlex::detail { concurrency c) { internal::verify_name(name, config_); - return provider_api{ - config_, name, algorithm_bits{bound_obj_, std::move(f)}, c, graph_, nodes_, errors_}; + return provider_api{config_, + name, + algorithm_bits{bound_obj_, std::move(f)}, + c, + graph_, + nodes_, + errors_, + resources_}; } // 'f' is a by-value sink: it is moved into algorithm_bits. The clang-tidy @@ -106,8 +126,14 @@ namespace phlex::detail { concurrency c) { internal::verify_name(name, config_); - return make_registration( - config_, name, algorithm_bits{bound_obj_, std::move(f)}, c, graph_, nodes_, errors_); + return make_registration(config_, + name, + algorithm_bits{bound_obj_, std::move(f)}, + c, + graph_, + nodes_, + errors_, + resources_); } // 'f' is a by-value sink: it is moved into algorithm_bits. The clang-tidy @@ -118,8 +144,14 @@ namespace phlex::detail { concurrency c) { internal::verify_name(name, config_); - return make_registration( - config_, name, algorithm_bits{bound_obj_, std::move(f)}, c, graph_, nodes_, errors_); + return make_registration(config_, + name, + algorithm_bits{bound_obj_, std::move(f)}, + c, + graph_, + nodes_, + errors_, + resources_); } auto unfold(std::string_view name, @@ -139,6 +171,7 @@ namespace phlex::detail { graph_, nodes_, errors_, + resources_, std::move(destination_data_layer)}; } @@ -168,6 +201,7 @@ namespace phlex::detail { node_catalog& nodes_; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members) std::shared_ptr bound_obj_; std::vector& errors_; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members) + resource_catalog& resources_; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members) configuration const* config_; }; } diff --git a/phlex/core/graph_proxy.hpp b/phlex/core/graph_proxy.hpp index ed429def2..c2d557771 100644 --- a/phlex/core/graph_proxy.hpp +++ b/phlex/core/graph_proxy.hpp @@ -39,7 +39,8 @@ namespace phlex::detail { graph_proxy(configuration const& config, tbb::flow::graph& g, node_catalog& nodes, - std::vector& errors) + std::vector& errors, + resource_catalog& resources) requires(not is_bound_object); /// @brief Binds a user algorithm object of type @p U to this proxy. @@ -110,7 +111,8 @@ namespace phlex::detail { tbb::flow::graph& g, node_catalog& nodes, std::shared_ptr bound_obj, - std::vector& errors) + std::vector& errors, + resource_catalog& resources) requires(is_bound_object); private: @@ -123,15 +125,17 @@ namespace phlex::detail { node_catalog& nodes_; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members) std::shared_ptr bound_obj_; std::vector& errors_; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members) + resource_catalog& resources_; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members) }; template graph_proxy::graph_proxy(configuration const& config, tbb::flow::graph& g, node_catalog& nodes, - std::vector& errors) + std::vector& errors, + resource_catalog& resources) requires(not is_bound_object) - : config_{&config}, graph_{g}, nodes_{nodes}, errors_{errors} + : config_{&config}, graph_{g}, nodes_{nodes}, errors_{errors}, resources_{resources} { } @@ -191,7 +195,7 @@ namespace phlex::detail { std::string destination_data_layer, concurrency c) const { - return glue{graph_, nodes_, nullptr, errors_, config_}.unfold( + return glue{graph_, nodes_, nullptr, errors_, resources_, config_}.unfold( name, std::move(pred), std::move(unf), c, std::move(destination_data_layer)); } @@ -216,8 +220,12 @@ namespace phlex::detail { Proxy graph_proxy::bind_to(Args&&... args) const requires(not is_bound_object) { - return Proxy{ - config_, graph_, nodes_, std::make_shared(std::forward(args)...), errors_}; + return Proxy{config_, + graph_, + nodes_, + std::make_shared(std::forward(args)...), + errors_, + resources_}; } template @@ -225,16 +233,24 @@ namespace phlex::detail { tbb::flow::graph& g, node_catalog& nodes, std::shared_ptr bound_obj, - std::vector& errors) + std::vector& errors, + resource_catalog& resources) requires(is_bound_object) - : config_{config}, graph_{g}, nodes_{nodes}, bound_obj_{std::move(bound_obj)}, errors_{errors} + : + config_{config}, + graph_{g}, + nodes_{nodes}, + bound_obj_{std::move(bound_obj)}, + errors_{errors}, + resources_{resources} { } template glue graph_proxy::create_glue(bool use_bound_object) const { - return glue{graph_, nodes_, (use_bound_object ? bound_obj_ : nullptr), errors_, config_}; + return glue{ + graph_, nodes_, (use_bound_object ? bound_obj_ : nullptr), errors_, resources_, config_}; } } diff --git a/phlex/core/node_builder.hpp b/phlex/core/node_builder.hpp new file mode 100644 index 000000000..9b282485d --- /dev/null +++ b/phlex/core/node_builder.hpp @@ -0,0 +1,174 @@ +#ifndef PHLEX_CORE_NODE_BUILDER_HPP +#define PHLEX_CORE_NODE_BUILDER_HPP + +#include "phlex/core/resource_api.hpp" + +#include "oneapi/tbb/flow_graph.h" + +#include +#include +#include +#include + +namespace phlex::detail { + + using no_outputs_t = std::tuple<>; + using no_resources_t = std::tuple<>; + + template + struct multifunction_outputs {}; + + template + struct node_builder; + + template + struct node_builder { + using node_t = tbb::flow::function_node; + + template + static node_t make(tbb::flow::graph& g, + std::size_t concurrency, + resource_catalog&, + Function ft, + NodeBody node_body) + { + return {g, + concurrency, + [ft = std::move(ft), node_body = std::move(node_body)]( + InputMessages const& messages) mutable -> oneapi::tbb::flow::continue_msg { + std::invoke(node_body, ft, messages); + return {}; + }}; + } + }; + + template + struct node_builder> { + using node_t = tbb::flow::resource_limited_node; + + template + static node_t make(tbb::flow::graph& g, + std::size_t concurrency, + resource_catalog& resources, + Function ft, + NodeBody node_body) + { + return {g, + concurrency, + std::tie(resources.template limiter_for(), + resources.template limiter_for()...), + [ft = std::move(ft), node_body = std::move(node_body)]( + InputMessages const& messages, auto&, auto*... resource_tokens) mutable { + std::invoke(node_body, ft, messages, resource_tokens...); + }}; + } + }; + + template + struct node_builder, no_resources_t> { + using node_t = tbb::flow::function_node; + + template + static node_t make(tbb::flow::graph& g, + std::size_t concurrency, + resource_catalog&, + Function ft, + NodeBody node_body) + { + return {g, + concurrency, + [ft = std::move(ft), + node_body = std::move(node_body)](InputMessages const& messages) mutable { + return std::invoke(node_body, ft, messages); + }}; + } + + static tbb::flow::sender& output_port(node_t& node) { return node; } + }; + + template + struct node_builder, + std::tuple> { + using node_t = tbb::flow::resource_limited_node>; + + template + static node_t make(tbb::flow::graph& g, + std::size_t concurrency, + resource_catalog& resources, + Function ft, + NodeBody node_body) + { + return {g, + concurrency, + std::tie(resources.template limiter_for(), + resources.template limiter_for()...), + [ft = std::move(ft), node_body = std::move(node_body)]( + InputMessages const& messages, auto& ports, auto*... resource_tokens) mutable { + std::get<0>(ports).try_put( + std::invoke(node_body, ft, messages, resource_tokens...)); + }}; + } + + static tbb::flow::sender& output_port(node_t& node) + { + return tbb::flow::output_port<0>(node); + } + }; + + template + struct node_builder, no_resources_t> { + using node_t = tbb::flow::multifunction_node>; + + template + static node_t make(tbb::flow::graph& g, + std::size_t concurrency, + resource_catalog&, + Function ft, + NodeBody node_body) + { + return {g, + concurrency, + [ft = std::move(ft), node_body = std::move(node_body)](InputMessages const& messages, + auto& ports) mutable { + std::invoke(node_body, ft, messages, ports); + }}; + } + }; + + template + struct node_builder, + std::tuple> { + using node_t = tbb::flow::resource_limited_node>; + + template + static node_t make(tbb::flow::graph& g, + std::size_t concurrency, + resource_catalog& resources, + Function ft, + NodeBody node_body) + { + return {g, + concurrency, + std::tie(resources.template limiter_for(), + resources.template limiter_for()...), + [ft = std::move(ft), node_body = std::move(node_body)]( + InputMessages const& messages, auto& ports, auto*... resource_tokens) mutable { + std::invoke(node_body, ft, messages, ports, resource_tokens...); + }}; + } + }; +} + +#endif // PHLEX_CORE_NODE_BUILDER_HPP diff --git a/phlex/core/registration_api.hpp b/phlex/core/registration_api.hpp index 790f60e2d..f060c434e 100644 --- a/phlex/core/registration_api.hpp +++ b/phlex/core/registration_api.hpp @@ -8,15 +8,21 @@ #include "phlex/core/declared_fold.hpp" #include "phlex/core/detail/make_algorithm_name.hpp" #include "phlex/core/node_catalog.hpp" +#include "phlex/core/resource_api.hpp" #include "phlex/core/upstream_predicates.hpp" #include "phlex/metaprogramming/delegate.hpp" #include "phlex/metaprogramming/type_deduction.hpp" #include "phlex/model/algorithm_name.hpp" +#include #include +#include #include #include #include +#include +#include +#include namespace phlex { class configuration; @@ -24,6 +30,24 @@ namespace phlex { namespace phlex::detail { + template + concept selector_or_resource = + std::same_as, product_selector> || is_resource; + + template + auto partition_selectors_and_resources(Args&&... args) + { + auto all = std::forward_as_tuple(std::forward(args)...); + auto selectors = [&all](std::index_sequence) { + return std::array{std::move(std::get(all))...}; + }(std::make_index_sequence{}); + auto resources = [&all](std::index_sequence) { + return + typename Split::resources_type{std::move(std::get(all))...}; + }(std::make_index_sequence{}); + return std::pair{std::move(selectors), std::move(resources)}; + } + // ==================================================================================== // Registration API @@ -36,6 +60,8 @@ namespace phlex::detail { static constexpr auto num_inputs = AlgorithmBits::number_inputs; static constexpr auto num_outputs = hof_type::number_output_products; + static_assert(num_inputs > 0, "input_family requires at least one product selector."); + public: registration_api(configuration const* config, std::string_view name, @@ -43,51 +69,76 @@ namespace phlex::detail { concurrency c, tbb::flow::graph& g, node_catalog& nodes, - std::vector& errors) : + std::vector& errors, + resource_catalog& resources) : config_{config}, name_{phlex::experimental::internal::make_algorithm_name(config, name)}, alg_{std::move(alg)}, concurrency_{c}, graph_{g}, - registrar_{nodes.registrar_for(errors)} + registrar_{nodes.registrar_for(errors)}, + resources_{resources} { } - auto input_family(std::array input_args) + template + auto input_family(std::array input_args, std::tuple) { - populate_types(input_args); + static_assert((is_resource && ...), + "The second argument to input_family(...) must be a tuple of resources."); + static_assert(num_inputs == NProducts + sizeof...(Resources), + "The number of function parameters is not the same as the number of specified " + "input arguments."); + + populate_types>(input_args); if constexpr (num_outputs == 0ull) { registrar_.set_creator([this, inputs = std::move(input_args)]( auto predicates, auto const& /* output_product_suffixes */) { - return std::make_unique(std::move(name_), - concurrency_.value, - std::move(predicates), - graph_, - std::move(alg_), - std::vector(inputs.begin(), inputs.end())); + using node_type = HOF...>; + return std::make_unique(std::move(name_), + concurrency_.value, + std::move(predicates), + graph_, + std::move(alg_), + std::vector(std::from_range, std::move(inputs)), + resources_); }); } else { registrar_.set_creator( [this, inputs = std::move(input_args)](auto predicates, auto output_product_suffixes) { - return std::make_unique(std::move(name_), - concurrency_.value, - std::move(predicates), - graph_, - std::move(alg_), - std::vector(inputs.begin(), inputs.end()), - std::move(output_product_suffixes)); + using node_type = HOF...>; + return std::make_unique(std::move(name_), + concurrency_.value, + std::move(predicates), + graph_, + std::move(alg_), + std::vector(std::from_range, std::move(inputs)), + std::move(output_product_suffixes), + resources_); }); } return upstream_predicates{std::move(registrar_), config_}; } - auto input_family(std::same_as auto... input_args) + auto input_family(std::array input_args) + { + return input_family(std::move(input_args), std::tuple<>{}); + } + + auto input_family(product_selector first_selector, selector_or_resource auto... additional_args) { - static_assert(num_inputs == sizeof...(input_args), + using split = resource_split; + static_assert(split::resources_are_last, + "All resource arguments to input_family(...) must follow the " + "product_selector arguments."); + static_assert(num_inputs == split::number_selectors + split::number_resources, "The number of function parameters is not the same as the number of specified " "input arguments."); - return input_family({std::move(input_args)...}); + + auto [selectors, resources] = partition_selectors_and_resources( + std::move(first_selector), std::move(additional_args)...); + return input_family(std::move(selectors), std::move(resources)); } private: @@ -98,6 +149,7 @@ namespace phlex::detail { // Non-owning reference to the TBB graph; this class is a short-lived registration builder. tbb::flow::graph& graph_; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members) registrar registrar_; + resource_catalog& resources_; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members) }; template