diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 245a5c8c8..6f57713ca 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -209,6 +209,22 @@ cet_test( LIBRARIES phlex::core_internal ) +cet_test( + transform_node + USE_CATCH2_MAIN + SOURCE + transform_node_test.cpp + LIBRARIES + phlex::core_internal +) +cet_test( + multilayer_join + USE_CATCH2_MAIN + SOURCE + multilayer_join_test.cpp + LIBRARIES + phlex::core_internal +) cet_test( replicated USE_CATCH2_MAIN diff --git a/test/multilayer_join_test.cpp b/test/multilayer_join_test.cpp new file mode 100644 index 000000000..d8300f01e --- /dev/null +++ b/test/multilayer_join_test.cpp @@ -0,0 +1,99 @@ +#include "phlex/core/multilayer_join_node.hpp" +#include "phlex/model/data_cell_index.hpp" +#include "phlex/model/product_store.hpp" + +#include "catch2/catch_test_macros.hpp" +#include "oneapi/tbb/flow_graph.h" + +#include +#include +#include +#include +#include + +using namespace phlex; +using namespace phlex::detail; +using namespace phlex::experimental; + +namespace { + constexpr auto message_id = 42u; + + struct input_type_1 { + int value; + }; + + struct input_type_2 { + int value; + }; + + template + product_specification spec(char const* creator, char const* suffix) + { + return {algorithm_name{creator}, identifier{suffix}, make_type_id()}; + } + + template + product_store_ptr store_with_product(char const* creator, char const* suffix, T value) + { + auto store = product_store::base(creator); + store->add_product(spec(creator, suffix), std::move(value)); + return store; + } +} + +TEST_CASE("multilayer_join_node joins multiple input products", "[join]") +{ + oneapi::tbb::flow::graph graph; + auto left_store = store_with_product("left_input", "", input_type_1{17}); + auto right_store = store_with_product("right_input", "", input_type_2{25}); + + // Force repeaters by passing distinct layer names. + // The actual routing is performed by matching index hashes between data, index, and flush. + auto join = multilayer_join_node<2>{ + graph, + "multilayer_join_test", + std::vector{identifier{"left_layer"}, identifier{"right_layer"}}}; + + oneapi::tbb::flow::queue_node> sink{graph}; + make_edge(output_port<0>(join), sink); + + auto& left_port = receiver_for<0ull, 2>(join, 0u); + auto& right_port = receiver_for<0ull, 2>(join, 1u); + + REQUIRE(left_port.try_put({.store = left_store, .id = message_id})); + graph.wait_for_all(); + + message_tuple<2> output; + CHECK_FALSE(sink.try_get(output)); + + REQUIRE(right_port.try_put({.store = right_store, .id = message_id})); + graph.wait_for_all(); + CHECK_FALSE(sink.try_get(output)); + + auto index_ports = join.index_ports(); + REQUIRE(index_ports.size() == 2u); + REQUIRE(index_ports[0].index_port->try_put( + {.index = left_store->index(), .msg_id = message_id, .cache = true})); + graph.wait_for_all(); + CHECK_FALSE(sink.try_get(output)); + + REQUIRE(index_ports[1].index_port->try_put( + {.index = right_store->index(), .msg_id = message_id, .cache = true})); + graph.wait_for_all(); + + REQUIRE(sink.try_get(output)); + CHECK_FALSE(sink.try_get(output)); + + CHECK(std::get<0>(output).id == message_id); + CHECK(std::get<1>(output).id == message_id); + REQUIRE(std::get<0>(output).store); + REQUIRE(std::get<1>(output).store); + CHECK(std::get<0>(output).store->index() == left_store->index()); + CHECK(std::get<1>(output).store->index() == right_store->index()); + + // Do what is necessary to have the tokens flushed, so that the test does not generate + // warnings. + REQUIRE(index_ports[0].token_port->try_put({.index = left_store->index(), .count = 1})); + REQUIRE(index_ports[1].token_port->try_put({.index = right_store->index(), .count = 1})); + graph.wait_for_all(); +} diff --git a/test/transform_node_test.cpp b/test/transform_node_test.cpp new file mode 100644 index 000000000..fff626ba6 --- /dev/null +++ b/test/transform_node_test.cpp @@ -0,0 +1,153 @@ +#include "phlex/core/declared_transform.hpp" +#include "phlex/metaprogramming/delegate.hpp" +#include "phlex/model/data_cell_index.hpp" +#include "phlex/model/product_store.hpp" + +#include "catch2/catch_test_macros.hpp" +#include "oneapi/tbb/flow_graph.h" + +#include +#include +#include +#include +#include + +using namespace phlex; +using namespace phlex::detail; +using namespace phlex::experimental; + +namespace { + constexpr auto message_id = 42u; + + struct input_type_1 { + int value; + auto operator<=>(input_type_1 const&) const = default; + }; + + struct input_type_2 { + int value; + auto operator<=>(input_type_2 const&) const = default; + }; + + struct output_type_1 { + int value; + auto operator<=>(output_type_1 const&) const = default; + }; + + struct output_type_2 { + std::string value; + auto operator<=>(output_type_2 const&) const = default; + }; + + output_type_1 double_value(input_type_1 const& input) { return {input.value * 2}; } + + auto number_and_label(input_type_1 const& input) + { + return std::tuple{output_type_1{input.value}, output_type_2{std::to_string(input.value)}}; + } + + template + product_specification spec(char const* creator, char const* suffix) + { + return {algorithm_name{creator}, identifier{suffix}, make_type_id()}; + } + + template + product_selector selector(char const* creator, char const* suffix) + { + return {.creator = creator, .layer = "job", .suffix = suffix, .type = make_type_id()}; + } + + template + product_store_ptr store_with_product(char const* creator, char const* suffix, T value) + { + auto store = product_store::base(creator); + store->add_product(spec(creator, suffix), std::move(value)); + return store; + } + + template + auto algorithm_bits_for(F function) + { + return algorithm_bits{std::shared_ptr{}, std::move(function)}; + } +} + +TEST_CASE("transform_node directly transforms one input product", "[transform_node]") +{ + oneapi::tbb::flow::graph graph; + auto input_selector = selector("input", ""); + auto input_store = store_with_product("input", "", input_type_1{21}); + auto alg = algorithm_bits_for(double_value); + + transform_node node{ + algorithm_name{"double_value"}, 1u, {}, graph, std::move(alg), {input_selector}, {}}; + declared_transform& transform = node; + + auto const& output_specs = transform.output(); + REQUIRE(output_specs.size() == 1u); + CHECK(output_specs[0].creator() == algorithm_name{"double_value"}); + CHECK(output_specs[0].suffix() == identifier{""}); + CHECK(output_specs[0].type() == make_type_id()); + + oneapi::tbb::flow::queue_node sink{graph}; + make_edge(transform.output_port(), sink); + + REQUIRE(node.port(input_selector).try_put({.store = input_store, .id = message_id})); + graph.wait_for_all(); + + message output; + REQUIRE(sink.try_get(output)); + + CHECK(output.id == message_id); + REQUIRE(output.store); + CHECK(output.store->index() == input_store->index()); + CHECK(output.store->source() == algorithm_name{"double_value"}); + CHECK(output.store->get_product(output_specs[0]) == output_type_1{42}); + + CHECK(transform.num_calls() == 1u); + CHECK(transform.product_count() == 1u); +} + +TEST_CASE("transform_node stores multiple output products", "[transform_node]") +{ + oneapi::tbb::flow::graph graph; + auto input_selector = selector("input", ""); + auto input_store = store_with_product("input", "", input_type_1{7}); + auto alg = algorithm_bits_for(number_and_label); + + transform_node node{algorithm_name{"number_and_label"}, + 1u, + {}, + graph, + std::move(alg), + {input_selector}, + {"number", "label"}}; + declared_transform& transform = node; + + oneapi::tbb::flow::queue_node sink{graph}; + make_edge(transform.output_port(), sink); + + REQUIRE(node.port(input_selector).try_put({.store = input_store, .id = message_id})); + graph.wait_for_all(); + + message output; + REQUIRE(sink.try_get(output)); + CHECK_FALSE(sink.try_get(output)); + + auto const& output_specs = transform.output(); + REQUIRE(output_specs.size() == 2u); + CHECK(output_specs[0].creator() == algorithm_name{"number_and_label"}); + CHECK(output_specs[0].suffix() == identifier{"number"}); + CHECK(output_specs[0].type() == make_type_id()); + CHECK(output_specs[1].creator() == algorithm_name{"number_and_label"}); + CHECK(output_specs[1].suffix() == identifier{"label"}); + CHECK(output_specs[1].type() == make_type_id()); + + REQUIRE(output.store); + CHECK(output.store->get_product(output_specs[0]) == output_type_1{7}); + CHECK(output.store->get_product(output_specs[1]) == output_type_2{"7"}); + + CHECK(transform.num_calls() == 1u); + CHECK(transform.product_count() == 1u); +}