diff --git a/phlex/core/registrar.hpp b/phlex/core/registrar.hpp index e906ce1fc..cf7ce3c62 100644 --- a/phlex/core/registrar.hpp +++ b/phlex/core/registrar.hpp @@ -80,8 +80,25 @@ namespace phlex::detail { registrar(registrar const&) = delete; registrar& operator=(registrar const&) = delete; - registrar(registrar&&) = default; - registrar& operator=(registrar&&) = default; + // Moving must clear the source creator so its destructor cannot register the node again. + registrar(registrar&& other) noexcept : + nodes_{other.nodes_}, + errors_{other.errors_}, + creator_{std::exchange(other.creator_, node_creator{})}, + predicates_{std::exchange(other.predicates_, std::nullopt)}, + output_product_suffixes_{std::move(other.output_product_suffixes_)} + { + } + + registrar& operator=(registrar&& other) noexcept + { + nodes_ = other.nodes_; + errors_ = other.errors_; + creator_ = std::exchange(other.creator_, node_creator{}); + predicates_ = std::exchange(other.predicates_, std::nullopt); + output_product_suffixes_ = std::move(other.output_product_suffixes_); + return *this; + } bool has_predicates() const { return predicates_.has_value(); } diff --git a/test/core_misc_test.cpp b/test/core_misc_test.cpp index 36c2cb75f..d90c13ad0 100644 --- a/test/core_misc_test.cpp +++ b/test/core_misc_test.cpp @@ -1,4 +1,5 @@ #include "phlex/configuration.hpp" +#include "phlex/core/declared_output.hpp" #include "phlex/core/glue.hpp" #include "phlex/core/registrar.hpp" #include "phlex/model/algorithm_name.hpp" @@ -7,9 +8,13 @@ #include #include +#include +#include +#include #include #include +#include #include TEST_CASE("algorithm_name tests", "[model]") @@ -95,3 +100,65 @@ TEST_CASE("add_to_error_messages tests", "[core]") REQUIRE(errors.size() == 1); CHECK(errors[0].contains("duplicate_node")); } + +TEST_CASE("A throwing registrar creator is not retried by the destructor", "[core]") +{ + using namespace phlex::detail; + + declared_outputs outputs; + std::vector errors; + int creations = 0; + { + registrar reg{outputs, errors}; + reg.set_creator([&](auto const&, auto const&) -> declared_output_ptr { + ++creations; + throw std::runtime_error{"creator failed"}; + }); + CHECK_THROWS_WITH(reg.set_output_product_suffixes({}), "creator failed"); + } + CHECK(creations == 1); + CHECK(errors.empty()); +} + +TEST_CASE("Moving a registrar transfers node creation", "[core]") +{ + using namespace phlex::detail; + + tbb::flow::graph graph; + declared_outputs outputs; + std::vector errors; + int creations = 0; + auto creator = [&](auto const&, auto const&) -> declared_output_ptr { + ++creations; + return std::make_unique(phlex::experimental::algorithm_name::create("output"), + 1, + std::vector{}, + graph, + [](auto const&) {}); + }; + + SECTION("Move construction") + { + { + registrar original{outputs, errors}; + original.set_creator(creator); + auto moved = std::move(original); + } + CHECK(creations == 1); + CHECK(outputs.get("output") != nullptr); + } + + SECTION("Move assignment") + { + { + registrar original{outputs, errors}; + original.set_creator(creator); + registrar moved{outputs, errors}; + moved = std::move(original); + } + CHECK(creations == 1); + CHECK(outputs.get("output") != nullptr); + } + + CHECK(errors.empty()); +} diff --git a/test/multilayer_join_test.cpp b/test/multilayer_join_test.cpp index 379678191..62df127e5 100644 --- a/test/multilayer_join_test.cpp +++ b/test/multilayer_join_test.cpp @@ -1,5 +1,7 @@ +#include "phlex/core/fold_join_node.hpp" #include "phlex/core/message.hpp" #include "phlex/core/multilayer_join_node.hpp" +#include "phlex/core/product_selector.hpp" #include "phlex/model/algorithm_name.hpp" #include "phlex/model/fwd.hpp" #include "phlex/model/identifier.hpp" @@ -10,7 +12,13 @@ #include #include +#include +#include +#include +#include +#include #include +#include #include #include @@ -73,6 +81,8 @@ TEST_CASE("multilayer_join_node joins multiple input products", "[join]") auto& left_port = receiver_for<0ull, 2>(join, 0u); auto& right_port = receiver_for<0ull, 2>(join, 1u); + CHECK_THROWS_AS(receiver_for<0ull>(join, 2u), std::runtime_error); + REQUIRE(left_port.try_put({.store = left_store, .id = message_id})); graph.wait_for_all(); @@ -85,6 +95,10 @@ TEST_CASE("multilayer_join_node joins multiple input products", "[join]") auto index_ports = join.index_ports(); REQUIRE(index_ports.size() == 2u); + CHECK(index_ports[0].layer == "left_layer"_id); + CHECK(index_ports[1].layer == "right_layer"_id); + CHECK_FALSE(index_ports[0].counting_layer.has_value()); + CHECK_FALSE(index_ports[1].counting_layer.has_value()); REQUIRE(index_ports[0].index_port->try_put( {.index = left_store->index(), .msg_id = message_id, .cache = true})); graph.wait_for_all(); @@ -110,3 +124,75 @@ TEST_CASE("multilayer_join_node joins multiple input products", "[join]") REQUIRE(index_ports[1].token_port->try_put({.index = right_store->index(), .count = 1})); graph.wait_for_all(); } + +TEST_CASE("Same-layer joins wait for all inputs and preserve selector order", "[join]") +{ + tbb::flow::graph graph; + multilayer_join_node<3> join{graph, "same_layer_join", {"event"_id, "event"_id, "event"_id}}; + CHECK(join.index_ports().empty()); + + auto const left_creator = algorithm_name::create("left"); + auto const middle_creator = algorithm_name::create("middle"); + auto const right_creator = algorithm_name::create("right"); + + product_selectors const selectors{{.creator = left_creator.to_string()}, + {.creator = middle_creator.to_string()}, + {.creator = right_creator.to_string()}}; + auto const stage = "test"_id; + std::array const stores{product_store::base(&left_creator, &stage), + product_store::base(&middle_creator, &stage), + product_store::base(&right_creator, &stage)}; + tbb::flow::queue_node> sink{graph}; + + auto try_put_for_receiver = [&join, &selectors, &stores](std::size_t const index) { + return internal::receiver_for<3>(join, selectors, selectors[index]) + .try_put({.store = stores.at(index), .id = message_id}); + }; + + using internal::receiver_for; + make_edge(output_port<0>(join), sink); + REQUIRE(try_put_for_receiver(2)); + REQUIRE(try_put_for_receiver(0)); + graph.wait_for_all(); + message_tuple<3> output; + CHECK_FALSE(sink.try_get(output)); + REQUIRE(try_put_for_receiver(1)); + graph.wait_for_all(); + REQUIRE(sink.try_get(output)); + CHECK(std::get<0>(output).id == message_id); + CHECK(std::get<1>(output).id == message_id); + CHECK(std::get<2>(output).id == message_id); + CHECK(std::get<0>(output).store == stores[0]); + CHECK(std::get<1>(output).store == stores[1]); + CHECK(std::get<2>(output).store == stores[2]); + CHECK_FALSE(sink.try_get(output)); + CHECK_THROWS_AS(receiver_for<3>(join, selectors, {.creator = "missing"_id}), std::runtime_error); +} + +TEST_CASE("Fold join data receivers exclude the partition port", "[join]") +{ + tbb::flow::graph graph; + auto const name = algorithm_name::create("fold_join_test"); + fold_join_node join{graph, + name, + "test_stage"_id, + "job"_id, + {"event"_id, "run"_id}, + product_specifications(1), + [](data_cell_index const&) { return std::make_unique(0); }}; + CHECK_THROWS_AS(receiver_for<1ull>(join, 2u), std::runtime_error); + CHECK_THROWS_AS(receiver_for<1ull>(join, std::numeric_limits::max()), + std::runtime_error); + product_selectors const selectors{{.creator = "left"_id}, {.creator = "right"_id}}; + CHECK(&receiver_for(join, selectors, selectors[0]) == &input_port<1>(join)); + CHECK(&receiver_for(join, selectors, selectors[1]) == &input_port<2>(join)); + + auto const ports = join.index_ports(); + REQUIRE(ports.size() == 3); + CHECK(ports[0].layer == "job"_id); + CHECK(ports[0].counting_layer == "event"_id); + CHECK(ports[1].layer == "event"_id); + CHECK(ports[1].counting_layer == "event"_id); + CHECK(ports[2].layer == "run"_id); + CHECK(ports[2].counting_layer == "run"_id); +} diff --git a/test/resources_test.cpp b/test/resources_test.cpp index 8a525ae67..65298e185 100644 --- a/test/resources_test.cpp +++ b/test/resources_test.cpp @@ -44,6 +44,16 @@ namespace { struct default_unlimited_resource {}; + struct counted_unlimited_resource { + explicit counted_unlimited_resource(int& constructions) { ++constructions; } + }; + + struct counted_serialized_resource { + using token_type = counted_serialized_resource const*; + + explicit counted_serialized_resource(int& constructions) { ++constructions; } + }; + struct invalid_serialized_resource { using token_type = int*; }; @@ -276,18 +286,28 @@ TEST_CASE("resource catalog", "[resource]") SECTION("duplicate serialized registrations throw") { - catalog.add_serialized(); - CHECK_THROWS_WITH(catalog.add_serialized(), + int constructions = 0; + catalog.add_serialized(constructions); + auto& limiter = catalog.limiter_for(); + CHECK(constructions == 1); + CHECK_THROWS_WITH(catalog.add_serialized(constructions), Catch::Matchers::ContainsSubstring("Resource of type '") && Catch::Matchers::ContainsSubstring("' has already been registered")); + CHECK(constructions == 1); + CHECK(&catalog.limiter_for() == &limiter); } SECTION("duplicate unlimited registrations throw") { - catalog.add_unlimited(42); - CHECK_THROWS_WITH(catalog.add_unlimited(43), + int constructions = 0; + catalog.add_unlimited(constructions); + auto const resource = catalog.access_for(); + CHECK(constructions == 1); + CHECK_THROWS_WITH(catalog.add_unlimited(constructions), Catch::Matchers::ContainsSubstring("Resource of type '") && Catch::Matchers::ContainsSubstring("' has already been registered")); + CHECK(constructions == 1); + CHECK(catalog.access_for() == resource); } SECTION("missing resources throw")