diff --git a/CMakeLists.txt b/CMakeLists.txt index 0a3d6e1b9..8bbb3f5b4 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -141,12 +141,10 @@ 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(TBB 2023.2.0 REQUIRED EXPORT) find_package(fmt REQUIRED EXPORT) find_package(spdlog REQUIRED EXPORT) diff --git a/Modules/private/PhlexTBB.cmake b/Modules/private/PhlexTBB.cmake deleted file mode 100644 index 8ec81b9ef..000000000 --- a/Modules/private/PhlexTBB.cmake +++ /dev/null @@ -1,39 +0,0 @@ -# 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 37d37dc75..d0cac2f6d 100644 --- a/phlex/core/CMakeLists.txt +++ b/phlex/core/CMakeLists.txt @@ -27,6 +27,7 @@ cet_make_library( provider_node.cpp registrar.cpp registration_api.cpp + resource/entries.cpp LIBRARIES PUBLIC TBB::tbb @@ -35,7 +36,9 @@ cet_make_library( phlex::model phlex::utilities PRIVATE + Boost::headers Boost::json + fmt::fmt spdlog::spdlog ) @@ -119,7 +122,7 @@ phlex_make_internal_library( phlex::metaprogramming phlex_model_internal phlex_utilities_internal - PRIVATE Boost::json spdlog::spdlog + PRIVATE Boost::headers Boost::json fmt::fmt spdlog::spdlog ) target_compile_definitions( phlex_core_internal diff --git a/phlex/core/framework_graph.hpp b/phlex/core/framework_graph.hpp index 1e68fb1a4..4cc87d309 100644 --- a/phlex/core/framework_graph.hpp +++ b/phlex/core/framework_graph.hpp @@ -163,6 +163,13 @@ namespace phlex::detail { resources_.template add_serialized(std::forward(args)...); } + template + requires pooled_resource_registration + void add_pooled_resource(Args&&... args) + { + resources_.template add_pooled(std::forward(args)...); + } + template glue make(Args&&... args) { diff --git a/phlex/core/node_builder.hpp b/phlex/core/node_builder.hpp index 886638136..785d6082f 100644 --- a/phlex/core/node_builder.hpp +++ b/phlex/core/node_builder.hpp @@ -15,7 +15,7 @@ namespace phlex::detail { // For each resource, in declaration order, the index it would occupy within just the - // unlimited or just the serialized subsequence, so a single pass can pick its argument + // unlimited or just the limited subsequence, so a single pass can pick its argument // from the right tuple. template inline constexpr auto resource_local_indices = [] { @@ -23,33 +23,33 @@ namespace phlex::detail { // The following variables are incremented in the fold expression below. // clang-tidy does not recognize this, so we suppress the warning about const correctness. // NOLINTBEGIN(misc-const-correctness) - std::size_t serialized = 0; + std::size_t limited = 0; std::size_t unlimited = 0; std::size_t i = 0; // NOLINTEND(misc-const-correctness) // gsl::at used to avoid cppcoreguidelines-pro-bounds-constant-array-index clang-tidy warning. - ((gsl::at(indices, i++) = unlimited_resource ? unlimited++ : serialized++), ...); + ((gsl::at(indices, i++) = unlimited_resource ? unlimited++ : limited++), ...); return indices; }(); - // Return either the unlimited resource (from the cached tuple) or the serialized-resource token + // Return either the unlimited resource (from the cached tuple) or a limited-resource token // (from the per-call tuple). template - decltype(auto) resource_argument(auto const& unlimited_resources, auto&& serialized_tokens) + decltype(auto) resource_argument(auto const& unlimited_resources, auto&& limited_tokens) { if constexpr (unlimited_resource) { return std::get(unlimited_resources); } else { - return std::get(serialized_tokens); + return std::get(limited_tokens); } } // Reassembles Resources... in declaration order for the node body, drawing each argument - // from either the per-call serialized_tokens (TBB-supplied) or the cached unlimited_resources. + // from either the per-call limited_tokens (TBB-supplied) or the cached unlimited_resources. template decltype(auto) invoke_with_resources(auto const& callable, auto&& arguments, - auto&& serialized_tokens, + auto&& limited_tokens, auto const& unlimited_resources) { return [&](std::index_sequence) -> decltype(auto) { @@ -59,28 +59,28 @@ namespace phlex::detail { return std::invoke(callable, std::forward(function_arguments)..., resource_argument[Is]>( - unlimited_resources, serialized_tokens)...); + unlimited_resources, limited_tokens)...); }, std::forward(arguments)); }(std::index_sequence_for{}); } using no_outputs_t = std::tuple<>; - using no_serialized_resources_t = std::tuple<>; + using no_limited_resources_t = std::tuple<>; template struct multifunction_outputs {}; // Selects and constructs the TBB node that matches an algorithm's output shape and resource // dependencies. Specializations use function_node or multifunction_node when all resources are - // unlimited, and resource_limited_node when any resource is serialized. They also adapt each + // unlimited, and resource_limited_node when any resource is limited. They also adapt each // node body's construction interface and provide output_port() only for node shapes with outputs. template struct node_builder; - // Specialization for an observer or a fold (no serialized resources) + // Specialization for an observer or a fold (no limited resources) template - requires(not has_serialized) + requires(not has_limited) struct node_builder> { using node_t = tbb::flow::function_node; @@ -99,15 +99,15 @@ namespace phlex::detail { unlimited = resource_dependencies::unlimited_resource_accesses(resources)]( InputMessages const& messages) mutable -> oneapi::tbb::flow::continue_msg { invoke_with_resources( - node_body, std::forward_as_tuple(ft, messages), no_serialized_resources_t{}, unlimited); + node_body, std::forward_as_tuple(ft, messages), no_limited_resources_t{}, unlimited); return {}; }}; } }; - // Specialization for an observer or a fold (serialized resources) + // Specialization for an observer or a fold (limited resources) template - requires(has_serialized) + requires(has_limited) struct node_builder> { using node_t = tbb::flow::resource_limited_node; @@ -121,7 +121,7 @@ namespace phlex::detail { return { g, concurrency, - resource_dependencies::serialized_resource_limiters(resources), + resource_dependencies::limited_resource_limiters(resources), [ft = std::move(ft), node_body = std::move(node_body), unlimited = resource_dependencies::unlimited_resource_accesses(resources)]( @@ -134,9 +134,9 @@ namespace phlex::detail { } }; - // Specialization for a transform or a predicate (no serialized resources) + // Specialization for a transform or a predicate (no limited resources) template - requires(not has_serialized) + requires(not has_limited) struct node_builder, std::tuple> { using node_t = tbb::flow::function_node; @@ -155,16 +155,16 @@ namespace phlex::detail { unlimited = resource_dependencies::unlimited_resource_accesses(resources)]( InputMessages const& messages) mutable { return invoke_with_resources( - node_body, std::forward_as_tuple(ft, messages), no_serialized_resources_t{}, unlimited); + node_body, std::forward_as_tuple(ft, messages), no_limited_resources_t{}, unlimited); }}; } static tbb::flow::sender& output_port(node_t& node) { return node; } }; - // Specialization for a transform or a predicate (serialized resources) + // Specialization for a transform or a predicate (limited resources) template - requires(has_serialized) + requires(has_limited) struct node_builder, std::tuple> { using node_t = tbb::flow::resource_limited_node>; @@ -178,7 +178,7 @@ namespace phlex::detail { return { g, concurrency, - resource_dependencies::serialized_resource_limiters(resources), + resource_dependencies::limited_resource_limiters(resources), [ft = std::move(ft), node_body = std::move(node_body), unlimited = resource_dependencies::unlimited_resource_accesses(resources)]( @@ -197,9 +197,9 @@ namespace phlex::detail { } }; - // Specialization for an unfold (no serialized resources) + // Specialization for an unfold (no limited resources) template - requires(not has_serialized) + requires(not has_limited) struct node_builder, std::tuple> { using node_t = tbb::flow::multifunction_node>; @@ -218,15 +218,15 @@ namespace phlex::detail { resources)](InputMessages const& messages, auto& ports) mutable { invoke_with_resources(node_body, std::forward_as_tuple(ft, messages, ports), - no_serialized_resources_t{}, + no_limited_resources_t{}, unlimited); }}; } }; - // Specialization for an unfold (serialized resources) + // Specialization for an unfold (limited resources) template - requires(has_serialized) + requires(has_limited) struct node_builder, std::tuple> { using node_t = tbb::flow::resource_limited_node>; @@ -240,7 +240,7 @@ namespace phlex::detail { return { g, concurrency, - resource_dependencies::serialized_resource_limiters(resources), + resource_dependencies::limited_resource_limiters(resources), [ft = std::move(ft), node_body = std::move(node_body), unlimited = resource_dependencies::unlimited_resource_accesses(resources)]( diff --git a/phlex/core/resource/catalog.hpp b/phlex/core/resource/catalog.hpp index b36065b2a..42bd6b86c 100644 --- a/phlex/core/resource/catalog.hpp +++ b/phlex/core/resource/catalog.hpp @@ -44,6 +44,19 @@ namespace phlex::detail { type, std::make_unique>(std::forward(args)...)); } + template + requires pooled_resource_registration + void add_pooled(Args&&... args) + { + auto const type = std::type_index(typeid(T)); + if (resources_.contains(type)) { + throw std::runtime_error(fmt::format("Resource of type '{}' has already been registered", + boost::core::demangle(typeid(T).name()))); + } + resources_.emplace(type, + std::make_unique>(std::forward(args)...)); + } + template gsl::not_null access_for() const { @@ -58,6 +71,13 @@ namespace phlex::detail { return entry->limiter(); } + template + auto& limiter_for() const + { + auto entry = entry_for>(); + return entry->limiter(); + } + private: template gsl::not_null entry_for() const diff --git a/phlex/core/resource/concepts.hpp b/phlex/core/resource/concepts.hpp index 24f78b215..3c1c627a7 100644 --- a/phlex/core/resource/concepts.hpp +++ b/phlex/core/resource/concepts.hpp @@ -2,7 +2,9 @@ #define PHLEX_CORE_RESOURCE_CONCEPTS_HPP #include +#include #include +#include namespace phlex::detail { namespace internal { @@ -10,11 +12,26 @@ namespace phlex::detail { concept has_token_type = requires { typename T::token_type; }; template - concept has_tokens = requires(T const& resource) { resource.tokens(); }; + concept has_tokens = requires(T& resource) { resource.tokens(); }; + + template + using pooled_token_range_t = decltype(std::declval().tokens()); + + template + using pooled_token_type_t = std::ranges::range_value_t>; + + template + concept has_valid_token_range = requires(T const& resource) { + { resource.tokens() } -> std::ranges::input_range; + requires std::copy_constructible>; + requires std::is_move_assignable_v>; + requires std::constructible_from, + std::ranges::range_reference_t>>; + }; } template - concept unlimited_resource = !internal::has_token_type; + concept unlimited_resource = !internal::has_token_type && !internal::has_tokens; template concept resource_token_constructible = @@ -28,10 +45,10 @@ namespace phlex::detail { concept serialized_resource = resource_token_constructible && !internal::has_tokens; template - concept pooled_resource = internal::has_token_type && internal::has_tokens; + concept pooled_resource = internal::has_valid_token_range; template - concept phase_1_supported_resource = unlimited_resource || serialized_resource; + concept limited_resource = serialized_resource || pooled_resource; template concept unlimited_resource_registration = @@ -41,6 +58,10 @@ namespace phlex::detail { concept serialized_resource_registration = serialized_resource && !std::is_const_v && std::constructible_from; + template + concept pooled_resource_registration = + pooled_resource && !std::is_const_v && std::constructible_from; + namespace internal { template struct resource_access_type; @@ -55,6 +76,11 @@ namespace phlex::detail { using type = T::token_type; }; + template + struct resource_access_type { + using type = pooled_token_type_t; + }; + template using resource_access_type_t = resource_access_type::type; @@ -62,7 +88,7 @@ namespace phlex::detail { using is_unlimited_resource = std::bool_constant>; template - using is_serialized_resource = std::bool_constant>; + using is_limited_resource = std::bool_constant>; } } diff --git a/phlex/core/resource/dependencies.hpp b/phlex/core/resource/dependencies.hpp index 47e25ab98..5b1a7b4d5 100644 --- a/phlex/core/resource/dependencies.hpp +++ b/phlex/core/resource/dependencies.hpp @@ -10,29 +10,26 @@ #include namespace phlex::detail { - // Retains the registration order while deriving the serialized and unlimited subsets needed - // by node construction. The index sequences let the invocation layer restore the original - // order after TBB has acquired only the serialized-resource tokens. + // Retains registration order while deriving the limited and unlimited subsets needed by node + // construction. The index sequences restore the original order after TBB acquires limited tokens. template struct resource_dependencies { - static_assert((phase_1_supported_resource && ...), - "resource dependencies currently support only unlimited or serialized " - "resources."); + static_assert(((unlimited_resource || limited_resource) && ...), + "resource dependencies must be unlimited, serialized, or pooled resources."); using all_resources = boost::mp11::mp_list; - using serialized_resources = - boost::mp11::mp_copy_if; + using limited_resources = boost::mp11::mp_copy_if; using unlimited_resources = boost::mp11::mp_copy_if; - using serialized_resource_indices = - internal::matching_indices; + using limited_resource_indices = + internal::matching_indices; using unlimited_resource_indices = internal::matching_indices; - static constexpr bool has_serialized_resources = serialized_resource_indices::size() != 0; + static constexpr bool has_limited_resources = limited_resource_indices::size() != 0; // Unlimited resources are cached once per node (access pointers are stable); serialized - // resources are looked up as the TBB resource_limiter references resource_limited_node needs. + // resources are looked up as the TBB resource_limiter references limited_resource_node needs. static auto unlimited_resource_accesses(resource_catalog const& resources) { return [&resources](boost::mp11::mp_list) { @@ -42,17 +39,16 @@ namespace phlex::detail { }(unlimited_resources{}); } - static auto serialized_resource_limiters(resource_catalog& resources) + static auto limited_resource_limiters(resource_catalog& resources) { - return [&resources](boost::mp11::mp_list) { - return std::tie(resources.template limiter_for()...); - }(serialized_resources{}); + return [&resources](boost::mp11::mp_list) { + return std::tie(resources.template limiter_for()...); + }(limited_resources{}); } }; template - inline constexpr bool has_serialized = - resource_dependencies::has_serialized_resources; + inline constexpr bool has_limited = resource_dependencies::has_limited_resources; } #endif // PHLEX_CORE_RESOURCE_DEPENDENCIES_HPP diff --git a/phlex/core/resource/entries.cpp b/phlex/core/resource/entries.cpp new file mode 100644 index 000000000..7166bcad9 --- /dev/null +++ b/phlex/core/resource/entries.cpp @@ -0,0 +1,15 @@ +#include "phlex/core/resource/entries.hpp" + +#include +#include + +#include +#include + +namespace phlex::detail { + [[noreturn]] void throw_empty_pooled_resource_error(std::type_info const& type) + { + throw std::runtime_error(fmt::format("Resource of type '{}' has an empty token pool", + boost::core::demangle(type.name()))); + } +} diff --git a/phlex/core/resource/entries.hpp b/phlex/core/resource/entries.hpp index 4c96617f6..dece9de48 100644 --- a/phlex/core/resource/entries.hpp +++ b/phlex/core/resource/entries.hpp @@ -2,14 +2,20 @@ #define PHLEX_CORE_RESOURCE_ENTRIES_HPP #include "phlex/core/resource/concepts.hpp" +#include "phlex/phlex_core_export.hpp" #include #include #include +#include +#include #include +#include namespace phlex::detail { + [[noreturn]] PHLEX_CORE_EXPORT void throw_empty_pooled_resource_error(std::type_info const& type); + class resource_base { public: virtual ~resource_base() = default; @@ -62,6 +68,40 @@ namespace phlex::detail { T resource_; tbb::flow::resource_limiter limiter_; }; + + template + class pooled_resource_entry : public resource_base { + public: + using resource_type = T; + using token_type = internal::pooled_token_type_t; + + template + explicit pooled_resource_entry(Args&&... args) : + resource_{std::forward(args)...}, limiter_{make_limiter(resource_)} + { + } + + tbb::flow::resource_limiter& limiter() noexcept { return limiter_; } + + private: + static tbb::flow::resource_limiter make_limiter(T const& resource) + { + std::vector tokens; + for (auto&& token : resource.tokens()) { + tokens.emplace_back(std::forward(token)); + } + + if (tokens.empty()) { + throw_empty_pooled_resource_error(typeid(T)); + } + return {tokens.begin(), tokens.end()}; + } + + // The limiter owns copies of token values. Declare the resource first so it outlives the + // limiter when tokens refer to resource-owned state. + T resource_; + tbb::flow::resource_limiter limiter_; + }; } #endif // PHLEX_CORE_RESOURCE_ENTRIES_HPP diff --git a/phlex/resource.hpp b/phlex/resource.hpp index 5645f4ca9..7e1f6f7dc 100644 --- a/phlex/resource.hpp +++ b/phlex/resource.hpp @@ -48,6 +48,13 @@ namespace phlex::detail { resources_.template add_serialized(std::forward(args)...); } + template + requires pooled_resource_registration + void add_pooled_resource(Args&&... args) const + { + resources_.template add_pooled(std::forward(args)...); + } + private: resource_catalog& resources_; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members) }; diff --git a/test/node_builder_test.cpp b/test/node_builder_test.cpp index fde2d038a..cd5a62833 100644 --- a/test/node_builder_test.cpp +++ b/test/node_builder_test.cpp @@ -5,15 +5,21 @@ #include #include +#include #include #include // Provides chrono literals used below but attributed to a transitive include by include-cleaner. +#include #include // IWYU pragma: keep #include +#include #include #include +#include +#include #include +#include using namespace phlex; using namespace phlex::detail; @@ -40,6 +46,20 @@ namespace { static int token() { return 42; } }; + struct pooled_resource_type { + static std::vector tokens() { return {42, 84}; } + }; + + struct one_token_pooled_resource_type { + static std::array tokens() { return {42}; } + }; + + struct span_pooled_resource_type { + std::array values{101, 202}; + + std::span tokens() const { return values; } + }; + // The production node bodies adapt product messages and output ports before invoking the // registered algorithm. These tests use an int input, so direct invocation isolates resource // argument assembly from node-family-specific behavior. @@ -107,6 +127,36 @@ namespace { std::atomic calls_; }; + // Releases two invocations together, but lets the test finish within a fixed bound if the + // pool or scheduler fails to provide overlap. + class bounded_overlap_gate { + public: + void arrive_and_wait() + { + std::unique_lock lock{mutex_}; + if (arrivals_ >= 2) { + return; + } + ++arrivals_; + condition_.notify_all(); + if (!condition_.wait_for(lock, 5s, [this] { return arrivals_ >= 2; })) { + timed_out_ = true; + } + } + + bool released_two_together() const + { + std::scoped_lock const lock{mutex_}; + return arrivals_ >= 2 && !timed_out_; + } + + private: + mutable std::mutex mutex_; + std::condition_variable condition_; + unsigned int arrivals_{}; + bool timed_out_{}; + }; + constexpr auto number_of_messages = 64uz; // Zero-resource node shapes. @@ -131,6 +181,9 @@ namespace { static_assert( std::same_as>::node_t, tbb::flow::resource_limited_node>); + static_assert( + std::same_as>::node_t, + tbb::flow::resource_limited_node>); } TEST_CASE("node builder supplies unlimited resources", "[node_builder][resource]") @@ -251,6 +304,151 @@ TEST_CASE("node builder supplies custom resource tokens", "[node_builder][resour CHECK(called); } +TEST_CASE("node builder supplies pooled resource tokens", "[node_builder][resource]") +{ + tbb::flow::graph graph; + resource_catalog resources; + resources.add_pooled(); + std::atomic received_token{}; + + using builder_t = node_builder>; + auto node = builder_t::make( + graph, + concurrency::serial.value, + resources, + [&received_token](int, int token) { received_token = token; }, + invoke_operation); + REQUIRE(node.try_put(43)); + graph.wait_for_all(); + auto const token = received_token.load(); + CHECK((token == 42 || token == 84)); +} + +TEST_CASE("pooled span values remain valid and distinct across dependent nodes", + "[node_builder][resource][concurrency]") +{ + tbb::global_control parallelism{tbb::global_control::max_allowed_parallelism, 4}; + tbb::flow::graph graph; + resource_catalog resources; + resources.add_pooled(); + concurrency_observations observations; + bounded_overlap_gate gate; + std::atomic token_values{}; + std::atomic accepted_messages{}; + + using builder_t = node_builder>; + auto function = [&observations, &gate, &token_values](int, int const token) { + concurrency_observations::scope const invocation{observations}; + if (token == 101) { + token_values.fetch_or(1u); + } else if (token == 202) { + token_values.fetch_or(2u); + } + gate.arrive_and_wait(); + }; + auto node1 = + builder_t::make(graph, concurrency::unlimited.value, resources, function, invoke_operation); + auto node2 = + builder_t::make(graph, concurrency::unlimited.value, resources, function, invoke_operation); + + tbb::parallel_for(0uz, 2uz, [&](std::size_t value) { + accepted_messages += node1.try_put(static_cast(value)); + accepted_messages += node2.try_put(static_cast(value)); + }); + graph.wait_for_all(); + + CHECK(accepted_messages == 4u); + CHECK(observations.calls() == 4u); + CHECK(observations.maximum_active() == 2u); + CHECK(gate.released_two_together()); + CHECK(token_values == 3u); +} + +TEST_CASE("one-token pooled resource limits dependent invocations", + "[node_builder][resource][concurrency]") +{ + tbb::flow::graph graph; + resource_catalog resources; + resources.add_pooled(); + concurrency_observations observations; + + using builder_t = node_builder>; + auto node = builder_t::make( + graph, + concurrency::unlimited.value, + resources, + [&observations](int, int token) { + concurrency_observations::scope const invocation{observations}; + if (token != 42) { + return; + } + spin_for(1ms); + }, + invoke_operation); + + tbb::parallel_for( + 0uz, number_of_messages, [&](std::size_t value) { node.try_put(static_cast(value)); }); + graph.wait_for_all(); + + CHECK(observations.calls() == number_of_messages); + CHECK(observations.maximum_active() == 1u); +} + +TEST_CASE("node concurrency remains below pooled capacity", "[node_builder][resource][concurrency]") +{ + tbb::flow::graph graph; + resource_catalog resources; + resources.add_pooled(); + concurrency_observations observations; + + using builder_t = node_builder>; + auto node = builder_t::make( + graph, + 1, + resources, + [&observations](int, int) { + concurrency_observations::scope const invocation{observations}; + spin_for(1ms); + }, + invoke_operation); + + tbb::parallel_for( + 0uz, number_of_messages, [&](std::size_t value) { node.try_put(static_cast(value)); }); + graph.wait_for_all(); + + CHECK(observations.calls() == number_of_messages); + CHECK(observations.maximum_active() == 1u); +} + +TEST_CASE("node builder preserves mixed unlimited serialized and pooled argument order", + "[node_builder][resource]") +{ + tbb::flow::graph graph; + resource_catalog resources; + resources.add_unlimited(); + resources.add_serialized(); + resources.add_pooled(); + std::atomic received_expected_arguments{}; + + using builder_t = node_builder< + int, + no_outputs_t, + std::tuple>; + auto node = builder_t::make( + graph, + concurrency::serial.value, + resources, + [&received_expected_arguments]( + int, int pooled_token, unlimited_resource_type const*, serialized_resource_type const*) { + received_expected_arguments = pooled_token == 42 || pooled_token == 84; + }, + invoke_operation); + REQUIRE(node.try_put(43)); + graph.wait_for_all(); + + CHECK(received_expected_arguments); +} + TEST_CASE("a shared serialized resource limits concurrency", "[node_builder][resource][concurrency]") { diff --git a/test/plugins/register_resources_for_testing.cpp b/test/plugins/register_resources_for_testing.cpp index 533c40e6a..fb5b43647 100644 --- a/test/plugins/register_resources_for_testing.cpp +++ b/test/plugins/register_resources_for_testing.cpp @@ -15,6 +15,7 @@ namespace { static_assert(requires(proxy const& r) { r.template add_unlimited_resource(1); r.template add_serialized_resource(); + r.template add_pooled_resource(); }); static_assert(not exposes_observe); } @@ -23,4 +24,5 @@ PHLEX_REGISTER_RESOURCES(r, config) { r.add_unlimited_resource(config.get("value")); r.add_serialized_resource(); + r.add_pooled_resource(); } diff --git a/test/plugins/resources_for_testing.hpp b/test/plugins/resources_for_testing.hpp index 914cc6f0f..27dcc32df 100644 --- a/test/plugins/resources_for_testing.hpp +++ b/test/plugins/resources_for_testing.hpp @@ -1,6 +1,8 @@ #ifndef TEST_PLUGINS_RESOURCES_FOR_TESTING_HPP #define TEST_PLUGINS_RESOURCES_FOR_TESTING_HPP +#include + namespace test::plugins { struct configured_resource { explicit configured_resource(int value) : value{value} {} @@ -11,6 +13,10 @@ namespace test::plugins { struct serialized_resource { using token_type = serialized_resource const*; }; + + struct pooled_resource { + static std::array tokens() { return {1, 2}; } + }; } #endif // TEST_PLUGINS_RESOURCES_FOR_TESTING_HPP diff --git a/test/plugins/verify_resources.cpp b/test/plugins/verify_resources.cpp index 2a434c66f..c1d384805 100644 --- a/test/plugins/verify_resources.cpp +++ b/test/plugins/verify_resources.cpp @@ -12,13 +12,18 @@ PHLEX_REGISTER_ALGORITHMS(m) { m.observe( "verify_resources", - [](int const, configured_resource const* configured, serialized_resource const* serialized) { + [](int const, + configured_resource const* configured, + serialized_resource const* serialized, + int pooled_token) { assert(configured != nullptr); assert(configured->value == 42); assert(serialized != nullptr); + assert(pooled_token == 1 || pooled_token == 2); }, concurrency::unlimited) .input_family(product_selector{.creator = "input", .layer = "event", .suffix = "i"}, resource{}, - resource{}); + resource{}, + resource{}); } diff --git a/test/resources_test.cpp b/test/resources_test.cpp index 8a525ae67..fb020ae4a 100644 --- a/test/resources_test.cpp +++ b/test/resources_test.cpp @@ -21,7 +21,13 @@ // Provides chrono literals used below but attributed to a transitive include by include-cleaner. #include // IWYU pragma: keep #include +#include +#include +#include +#include +#include #include +#include using namespace phlex; using namespace phlex::detail; @@ -70,12 +76,12 @@ namespace { int token() const { return 24; } }; - // A mutable-only tokens() member does not affect the unlimited classification. + // A malformed pooled declaration must not fall back to unlimited registration. struct mutable_tokens_unlimited_resource { void tokens() {} }; - // A mutable-only tokens() member does not make a token-bearing resource bounded. + // A mutable-only tokens() declaration makes this resource invalid. struct mutable_tokens_serialized_resource { using token_type = mutable_tokens_serialized_resource const*; @@ -83,14 +89,88 @@ namespace { }; struct pooled_token_resource { - using token_type = int*; - static auto tokens() { return std::array{}; } }; + struct pooled_view_resource { + static auto tokens() { return std::span{}; } + }; + + struct pooled_object_view_resource { + std::array values{11, 29}; + + auto tokens() const + { + return values | std::views::transform([](int const& value) { return &value; }); + } + }; + + struct pooled_object_reference_resource { + std::vector values{11, 29}; + + std::vector const& tokens() const { return values; } + }; + + struct temporary_pooled_objects_resource { + static std::vector tokens() { return {11, 29}; } + }; + + struct filtered_pooled_resource { + std::array values{11, 0, 29}; + + auto tokens() const + { + return values | std::views::filter([](int value) { return value != 0; }); + } + }; + + struct proxy_pooled_objects_resource { + std::vector values{true, false}; + + auto tokens() const { return std::views::all(values); } + }; + + struct prvalue_pooled_objects_resource { + static auto tokens() { return std::views::iota(11) | std::views::take(2); } + }; + + struct counting_pooled_resource { + counting_pooled_resource(int& calls, std::vector values) : + calls{calls}, values{std::move(values)} + { + } + + std::vector tokens() const + { + ++calls; + return values; + } + + int& calls; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members) + std::vector values; + }; + + struct invalid_pooled_range_resource { + static int tokens() { return 42; } + }; + + struct noncopyable_pooled_token_resource { + static auto tokens() { return std::array, 1>{}; } + }; + + struct nonassignable_pooled_token_resource { + struct token { + // NOLINTNEXTLINE(readability-identifier-naming, cppcoreguidelines-avoid-const-or-ref-data-members) + int const value; + }; + + static auto tokens() { return std::array{{{42}}}; } + }; + using mixed_dependencies = resource_dependencies; using reversed_dependencies = resource_dependencies; using serialized_dependencies = resource_dependencies; + using pooled_dependencies = resource_dependencies; template concept graph_accepts_unlimited = @@ -108,6 +188,14 @@ namespace { concept catalog_accepts_serialized = requires(resource_catalog& catalog) { catalog.template add_serialized(); }; + template + concept graph_accepts_pooled = + requires(framework_graph& graph) { graph.template add_pooled_resource(); }; + + template + concept catalog_accepts_pooled = + requires(resource_catalog& catalog) { catalog.template add_pooled(); }; + static_assert(graph_accepts_unlimited); static_assert(not graph_accepts_serialized); static_assert(catalog_accepts_unlimited); @@ -128,20 +216,22 @@ namespace { static_assert(not catalog_accepts_unlimited); static_assert(catalog_accepts_serialized); - static_assert(graph_accepts_unlimited); + static_assert(not graph_accepts_unlimited); static_assert(not graph_accepts_serialized); - static_assert(catalog_accepts_unlimited); + static_assert(not catalog_accepts_unlimited); static_assert(not catalog_accepts_serialized); static_assert(not graph_accepts_unlimited); - static_assert(graph_accepts_serialized); + static_assert(not graph_accepts_serialized); static_assert(not catalog_accepts_unlimited); - static_assert(catalog_accepts_serialized); + static_assert(not catalog_accepts_serialized); static_assert(not graph_accepts_unlimited); static_assert(not graph_accepts_serialized); + static_assert(graph_accepts_pooled); static_assert(not catalog_accepts_unlimited); static_assert(not catalog_accepts_serialized); + static_assert(catalog_accepts_pooled); static_assert(not graph_accepts_unlimited); static_assert(not graph_accepts_serialized); @@ -180,37 +270,103 @@ namespace { // Invalid token types. static_assert(not resource_token_constructible); static_assert(not serialized_resource); - static_assert(not phase_1_supported_resource); - // Pooled resources, deferred in Phase 1. + // Pooled handles are inferred from range values, not range references. static_assert(pooled_resource); + static_assert(pooled_resource_registration); static_assert(not serialized_resource); - static_assert(not phase_1_supported_resource); + static_assert(pooled_resource); + static_assert(pooled_resource_registration); + static_assert(not pooled_resource_registration); + static_assert(not pooled_resource_registration); + static_assert(not serialized_resource); + + static_assert(pooled_resource); + static_assert(graph_accepts_pooled); + static_assert(catalog_accepts_pooled); + static_assert(pooled_resource); + static_assert(pooled_resource); + static_assert(graph_accepts_pooled); + static_assert(catalog_accepts_pooled); + static_assert(not serialized_resource); + static_assert(pooled_resource); + static_assert(pooled_resource); + static_assert(pooled_resource); + static_assert( + std::same_as, int const*>); + static_assert( + std::same_as, int>); + static_assert( + std::same_as, bool>); + + // Malformed token ranges are neither pooled nor eligible for serialized fallback. + static_assert(not pooled_resource); + static_assert(not unlimited_resource); + static_assert(not graph_accepts_unlimited); + static_assert(not serialized_resource); + static_assert(not graph_accepts_serialized); + static_assert(not catalog_accepts_serialized); + static_assert(not pooled_resource); + static_assert(not serialized_resource); + static_assert(not pooled_resource); + static_assert(not serialized_resource); + static_assert(not pooled_resource); + static_assert(not serialized_resource); // Resource dependency partitioning. - static_assert(mixed_dependencies::has_serialized_resources); + static_assert(mixed_dependencies::has_limited_resources); static_assert( - std::same_as>); + std::same_as>); static_assert(std::same_as>); - static_assert( - std::same_as>); + static_assert(std::same_as>); static_assert( std::same_as>); - static_assert(reversed_dependencies::has_serialized_resources); + static_assert(reversed_dependencies::has_limited_resources); static_assert( - std::same_as>); + std::same_as>); static_assert( std::same_as>); - static_assert(serialized_dependencies::has_serialized_resources); + static_assert(serialized_dependencies::has_limited_resources); static_assert( - std::same_as>); + std::same_as>); static_assert( std::same_as>); - static_assert(not resource_dependencies::has_serialized_resources); - static_assert(not resource_dependencies<>::has_serialized_resources); + static_assert(not resource_dependencies::has_limited_resources); + static_assert(not resource_dependencies<>::has_limited_resources); + + static_assert(pooled_dependencies::has_limited_resources); + static_assert(std::same_as>); + static_assert( + std::same_as>); + static_assert(std::same_as, int*>); + + template + void verify_pooled_tokens(resource_catalog& catalog, + IsValid is_valid, + unsigned int number_of_messages = 2) + { + catalog.add_pooled(); + tbb::flow::graph graph; + std::atomic valid_tokens{}; + tbb::flow::resource_limited_node> node{ + graph, + tbb::flow::serial, + std::tie(catalog.limiter_for()), + [&valid_tokens, is_valid](int, auto&, internal::resource_access_type_t token) { + if (is_valid(token)) { + ++valid_tokens; + } + }}; + for (unsigned int i = 0; i < number_of_messages; ++i) { + REQUIRE(node.try_put(static_cast(i))); + } + graph.wait_for_all(); + CHECK(valid_tokens == number_of_messages); + } } TEST_CASE("registering a pointer resource with the graph", "[graph][resource]") @@ -274,6 +430,72 @@ TEST_CASE("resource catalog", "[resource]") CHECK_NOTHROW(catalog.limiter_for()); } + SECTION("pooled resources materialize a temporary token range once") + { + int calls{}; + catalog.add_pooled(calls, std::vector{11, 29, 47}); + CHECK(calls == 1); + CHECK_NOTHROW(catalog.limiter_for()); + CHECK(calls == 1); + } + + SECTION("explicit pointer handles refer to resource-owned elements") + { + verify_pooled_tokens( + catalog, [](int const* token) { return token && (*token == 11 || *token == 29); }); + } + + SECTION("owning ranges and views supply copied value handles") + { + auto is_expected_value = [](int token) { return token == 11 || token == 29; }; + verify_pooled_tokens(catalog, is_expected_value); + verify_pooled_tokens(catalog, is_expected_value); + verify_pooled_tokens(catalog, is_expected_value); + } + + SECTION("prvalue input ranges with distinct sentinels supply value handles") + { + verify_pooled_tokens( + catalog, [](int token) { return token == 11 || token == 12; }, 1); + } + + SECTION("pointer ranges supply pointer values rather than their addresses") + { + catalog.add_pooled(); + tbb::flow::graph graph; + std::atomic received_null_token{}; + tbb::flow::resource_limited_node> node{ + graph, + tbb::flow::serial, + std::tie(catalog.limiter_for()), + [&received_null_token](int, auto&, int const* token) { + received_null_token = token == nullptr; + }}; + REQUIRE(node.try_put(0)); + graph.wait_for_all(); + CHECK(received_null_token); + } + + SECTION("empty pooled resources throw during registration") + { + int calls{}; + CHECK_THROWS_WITH(catalog.add_pooled(calls, std::vector{}), + Catch::Matchers::ContainsSubstring("counting_pooled_resource") && + Catch::Matchers::ContainsSubstring("empty token pool")); + CHECK(calls == 1); + CHECK_THROWS_WITH(catalog.limiter_for(), + Catch::Matchers::ContainsSubstring("has not been registered")); + } + + SECTION("duplicate pooled registrations throw") + { + int calls{}; + catalog.add_pooled(calls, std::vector{11}); + CHECK_THROWS_WITH(catalog.add_pooled(calls, std::vector{29}), + Catch::Matchers::ContainsSubstring("has already been registered")); + CHECK(calls == 1); + } + SECTION("duplicate serialized registrations throw") { catalog.add_serialized(); @@ -298,5 +520,8 @@ TEST_CASE("resource catalog", "[resource]") CHECK_THROWS_WITH(catalog.access_for(), Catch::Matchers::ContainsSubstring("Resource of type '") && Catch::Matchers::ContainsSubstring("' has not been registered")); + CHECK_THROWS_WITH(catalog.limiter_for(), + Catch::Matchers::ContainsSubstring("Resource of type '") && + Catch::Matchers::ContainsSubstring("' has not been registered")); } } diff --git a/test/tbb-preview/resource_limiting_test.cpp b/test/tbb-preview/resource_limiting_test.cpp index 0ead0219c..7a9e76a86 100644 --- a/test/tbb-preview/resource_limiting_test.cpp +++ b/test/tbb-preview/resource_limiting_test.cpp @@ -8,6 +8,7 @@ #include // Provides chrono literals used below but attributed to a transitive include by include-cleaner. +#include #include // IWYU pragma: keep #include #include @@ -200,7 +201,8 @@ TEST_CASE("Test based on oneTBB PR 1677 (RFC)", "[multithreading]") database const db1{1}; database const db13{13}; - flow::resource_limiter db_limiter{&db1, &db13}; + std::array databases{&db1, &db13}; + flow::resource_limiter db_limiter{databases.begin(), databases.end()}; auto fill_histo = [&root_counter](unsigned int const i, auto& outputs, root) { thread_counter const c{root_counter};