Add test for join_node, polish the implementation

Author: kboyarinov

include/oneapi/tbb/detail/_flow_graph_indexer_impl.h

@@ -1,5 +1,6 @@
 /*
-    Copyright (c) 2005-2024 Intel Corporation
+    Copyright (c) 2005-2025 Intel Corporation
+    Copyright (c) 2025 UXL Foundation Contributors
 
     Licensed under the Apache License, Version 2.0 (the "License");
     you may not use this file except in compliance with the License.
@@ -227,18 +228,18 @@
     struct indexer_types {
         using output_type = tagged_msg<std::size_t, T0, TN...>;
         using input_ports_type = std::tuple<indexer_input_port<T0>, indexer_input_port<TN>...>;
-        using indexer_FE_type = indexer_node_FE<input_ports_type, output_type, std::tuple<T0, TN...>>;
         using indexer_base_type = indexer_node_base<input_ports_type, output_type, std::tuple<T0, TN...>>;
     };
 
     template<typename T0, typename... TN>
     class unfolded_indexer_node : public indexer_types<T0, TN...>::indexer_base_type {
     public:
-        typedef std::tuple<T0, TN...> tuple_types;
-        typedef typename indexer_types<T0, TN...>::input_ports_type input_ports_type;
-        typedef typename indexer_types<T0, TN...>::output_type output_type;
+        using input_ports_type = typename indexer_types<T0, TN...>::input_ports_type;
+        using output_type = typename indexer_types<T0, TN...>::output_type;
+        using tuple_types = std::tuple<T0, TN...>;
+
     private:
-        typedef typename indexer_types<T0, TN...>::indexer_base_type base_type;
+        using base_type = typename indexer_types<T0, TN...>::indexer_base_type;
     public:
         unfolded_indexer_node(graph& g) : base_type(g) {}
         unfolded_indexer_node(const unfolded_indexer_node &other) : base_type(other) {}

include/oneapi/tbb/detail/_flow_graph_join_impl.h

@@ -1,5 +1,6 @@
 /*
-    Copyright (c) 2005-2024 Intel Corporation
+    Copyright (c) 2005-2025 Intel Corporation
+    Copyright (c) 2025 UXL Foundation Contributors
 
     Licensed under the Apache License, Version 2.0 (the "License");
     you may not use this file except in compliance with the License.
@@ -1458,34 +1459,19 @@
     // join base class type generator
     template<template<class> class PT, typename OutputTuple, typename JP>
     struct join_base {
-        typedef join_node_base<JP, typename wrap_tuple_elements<PT,OutputTuple>::type, OutputTuple> type;
+        using type = join_node_base<JP, typename wrap_tuple_elements<PT, OutputTuple>::type, OutputTuple>;
     };
 
     template<typename OutputTuple, typename K, typename KHash>
     struct join_base<key_matching_port, OutputTuple, key_matching<K,KHash> > {
-        typedef key_matching<K, KHash> key_traits_type;
-        typedef K key_type;
-        typedef KHash key_hash_compare;
-        typedef join_node_base< key_traits_type,
-                // ports type
-                typename wrap_key_tuple_elements<key_matching_port,key_traits_type,OutputTuple>::type,
-                OutputTuple > type;
-    };
+        using key_type = K;
+        using key_hash_compare = KHash;
+        using key_traits_type = key_matching<key_type, key_hash_compare>;
 
-    //! unfolded_join_node : passes input_ports_type to join_node_base.  We build the input port type
-    //  using tuple_element.  The class PT is the port type (reserving_port, queueing_port, key_matching_port)
-    //  and should match the typename.
-
-    template<int M, template<class> class PT, typename OutputTuple, typename JP>
-    class unfolded_join_node : public join_base<M,PT,OutputTuple,JP>::type {
-    public:
-        typedef typename wrap_tuple_elements<M, PT, OutputTuple>::type input_ports_type;
-        typedef OutputTuple output_type;
-    private:
-        typedef join_node_base<JP, input_ports_type, output_type > base_type;
-    public:
-        unfolded_join_node(graph &g) : base_type(g) {}
-        unfolded_join_node(const unfolded_join_node &other) : base_type(other) {}
+        using type = join_node_base<key_traits_type,
+                // ports type
+                typename wrap_key_tuple_elements<key_matching_port, key_traits_type, OutputTuple>::type,
+                OutputTuple>;
     };
 
 #if __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING
@@ -1504,25 +1490,23 @@
     };
 #endif /* __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING */
 
-    //! unfolded_join_node : passes input_ports_type to join_node_base.  We build the input port type
-    //  using tuple_element.  The class PT is the port type (reserving_port, queueing_port, key_matching_port)
+    //! unfolded_join_node : passes input_ports_type to join_node_base. We build the input port type
+    //  using tuple_element.  The class PortType is the port type (reserving_port, queueing_port, key_matching_port)
     //  and should match the typename.
-
-    template<template<class> class PT, typename OutputTuple, typename JP>
-    class unfolded_join_node : public join_base<PT, OutputTuple, JP>::type {
+    template<template<class> class PortType, typename OutputTuple, typename JoinPolicy>
+    class unfolded_join_node : public join_base<PortType, OutputTuple, JoinPolicy>::type {
     public:
-        using input_ports_type = typename wrap_tuple_elements<PT, OutputTuple>::type;
+        using input_ports_type = typename wrap_tuple_elements<PortType, OutputTuple>::type;
         using output_type = OutputTuple;
     private:
-        using base_type = join_node_base<JP, input_ports_type, output_type>;
+        using base_type = join_node_base<JoinPolicy, input_ports_type, output_type>;
     public:
         unfolded_join_node(graph& g) : base_type(g) {}
         unfolded_join_node(const unfolded_join_node &other) : base_type(other) {}
     };
 
     // key_matching unfolded_join_node.  This must be a separate specialization because the constructors
     // differ.
-
     template<typename K, typename KHash, typename...Types>
     class unfolded_join_node<key_matching_port, std::tuple<Types...>, key_matching<K, KHash>>
         : public join_base<key_matching_port, std::tuple<Types...>, key_matching<K, KHash>>::type

include/oneapi/tbb/detail/_flow_graph_types_impl.h

@@ -1,5 +1,6 @@
 /*
-    Copyright (c) 2005-2024 Intel Corporation
+    Copyright (c) 2005-2025 Intel Corporation
+    Copyright (c) 2025 UXL Foundation Contributors
 
     Licensed under the Apache License, Version 2.0 (the "License");
     you may not use this file except in compliance with the License.
@@ -43,24 +44,24 @@ struct KeyTrait {
 };
 
 // wrap each element of a tuple in a template, and make a tuple of the result.
-template<template<class> class PT, typename TypeTuple>
+template<template<class> class PortType, typename TypeTuple>
 struct wrap_tuple_elements;
 
 // A wrapper that generates the traits needed for each port of a key-matching join,
 // and the type of the tuple of input ports.
-template<template<class> class PT, typename KeyTraits, typename TypeTuple>
+template<template<class> class PortType, typename KeyTraits, typename TypeTuple>
 struct wrap_key_tuple_elements;
 
-template<template<class> class PT,  typename... Args>
-struct wrap_tuple_elements<PT, std::tuple<Args...> >{
-    typedef typename std::tuple<PT<Args>... > type;
+template<template<class> class PortType,  typename... Args>
+struct wrap_tuple_elements<PortType, std::tuple<Args...> >{
+    using type = std::tuple<PortType<Args>...>;
 };
 
-template<template<class> class PT, typename KeyTraits, typename... Args>
-struct wrap_key_tuple_elements<PT, KeyTraits, std::tuple<Args...> > {
-    typedef typename KeyTraits::key_type K;
-    typedef typename KeyTraits::hash_compare_type KHash;
-    typedef typename std::tuple<PT<KeyTrait<K, KHash, Args> >... > type;
+template<template<class> class PortType, typename KeyTraits, typename... Args>
+struct wrap_key_tuple_elements<PortType, KeyTraits, std::tuple<Args...> > {
+    using key_type = typename KeyTraits::key_type;
+    using hash_compare_type = typename KeyTraits::hash_compare_type;
+    using type = std::tuple<PortType<KeyTrait<key_type, hash_compare_type, Args>>...>;
 };
 
 template< int... S > class sequence {};
@@ -309,9 +310,9 @@ struct do_if<T, false> {
 
 using tbb::detail::punned_cast;
 
-template<typename TagType, typename T0, typename... TN>
+template<typename TagType, typename... TN>
 class tagged_msg {
-    using Tuple = std::tuple<T0, TN...>;
+    using Tuple = std::tuple<TN...>;
 
     class variant {
         static const size_t N = std::tuple_size<Tuple>::value;
@@ -387,7 +388,7 @@ class tagged_msg {
     bool is_a() const {return my_msg.template variant_is_a<V>();}
 
     bool is_default_constructed() const {return my_msg.variant_is_default_constructed();}
-};
+}; // class tagged_msg
 
 // template to simplify cast and test for tagged_msg in template contexts
 template<typename V, typename T>

include/oneapi/tbb/flow_graph.h

@@ -1,5 +1,6 @@
 /*
     Copyright (c) 2005-2025 Intel Corporation
+    Copyright (c) 2025 UXL Foundation Contributors
 
     Licensed under the Apache License, Version 2.0 (the "License");
     you may not use this file except in compliance with the License.
@@ -2495,9 +2496,9 @@ class join_node<OutputTuple, key_matching<K, KHash> > : public unfolded_join_nod
     join_node(graph &g) : unfolded_type(g) {}
 #endif  /* __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING */
 
-    template<typename... Bodies, typename = typename std::enable_if<sizeof...(Bodies) == N>>
-        __TBB_requires(join_node_functions<OutputTuple, K, Bodies...>)
-    __TBB_NOINLINE_SYM join_node(graph& g, Bodies... bodies) : unfolded_type(g, bodies...) {
+    template <typename Body, typename... Bodies, typename = typename std::enable_if<1 + sizeof...(Bodies) == N>>
+        __TBB_requires(join_node_functions<OutputTuple, K, Body, Bodies...>)
+    __TBB_NOINLINE_SYM join_node(graph& g, Body body, Bodies... bodies) : unfolded_type(g, body, bodies...) {
         fgt_multiinput_node<N>( CODEPTR(), FLOW_JOIN_NODE_TAG_MATCHING, &this->my_graph,
                                                            this->input_ports(), static_cast< sender< output_type > *>(this) );
     }

test/conformance/conformance_flowgraph.h

@@ -1,5 +1,6 @@
 /*
     Copyright (c) 2020-2025 Intel Corporation
+    Copyright (c) 2025 UXL Foundation Contributors
 
     Licensed under the Apache License, Version 2.0 (the "License");
     you may not use this file except in compliance with the License.
@@ -811,4 +812,25 @@ void test_with_reserving_join_node_class() {
         if at least one successor accepts the tuple must consume messages");
 }
 }
+
+template <std::size_t N>
+struct edge_maker {
+    template <typename Sender, typename NodeType>
+    static void make(Sender& sender, NodeType& node) {
+        oneapi::tbb::flow::make_edge(sender, oneapi::tbb::flow::input_port<N - 1>(node));
+        edge_maker<N - 1>::make(sender, node);
+    }
+
+    template <typename Sender, typename NodeType>
+    static void make(std::vector<Sender>& senders, NodeType& node) {
+        oneapi::tbb::flow::make_edge(senders[N - 1], oneapi::tbb::flow::input_port<N - 1>(node));
+        edge_maker<N - 1>::make(senders, node);
+    }
+};
+
+template <>
+struct edge_maker<0> {
+    template <typename Sender, typename NodeType>
+    static void make(Sender&, NodeType&) {}
+};
 #endif // __TBB_test_conformance_conformance_flowgraph_H

test/conformance/conformance_indexer_node.cpp

@@ -1,5 +1,6 @@
 /*
-    Copyright (c) 2020-2021 Intel Corporation
+    Copyright (c) 2020-2025 Intel Corporation
+    Copyright (c) 2025 UXL Foundation Contributors
 
     Licensed under the Apache License, Version 2.0 (the "License");
     you may not use this file except in compliance with the License.
@@ -19,6 +20,7 @@
 #endif
 
 #include "conformance_flowgraph.h"
+#include <unordered_set>
 
 //! \file conformance_indexer_node.cpp
 //! \brief Test for [flow_graph.indexer_node] specification
@@ -162,8 +164,6 @@ TEST_CASE("indexer_node output_type") {
     CHECK_MESSAGE((conformance::check_output_type<my_output_type, oneapi::tbb::flow::tagged_msg<size_t, int, float, input_msg>>()), "indexer_node output_type should returns a tagged_msg");
 }
 
-#include <unordered_set>
-
 template <std::size_t N, typename... Args>
 struct indexer_node_type_generator_impl {
     using type = typename indexer_node_type_generator_impl<N - 1, Args..., int>::type;
@@ -178,66 +178,40 @@ struct indexer_node_type_generator_impl<0, Args...> {
 template <std::size_t N>
 using indexer_node_type_generator_t = typename indexer_node_type_generator_impl<N>::type;
 
-template <std::size_t N>
-struct edge_maker {
-    template <typename SendersVector, typename IndexerNodeType>
-    static void make(SendersVector& senders, IndexerNodeType& indexer) {
-        oneapi::tbb::flow::make_edge(senders[N - 1], oneapi::tbb::flow::input_port<N - 1>(indexer));
-        edge_maker<N - 1>::make(senders, indexer);
-    }
-};
-
-template <>
-struct edge_maker<1> {
-    template <typename SendersVector, typename IndexerNodeType>
-    static void make(SendersVector& senders, IndexerNodeType& indexer) {
-        oneapi::tbb::flow::make_edge(senders[0], oneapi::tbb::flow::input_port<0>(indexer));
-    }  
-};
-
 template <std::size_t NInputs>
 void test_indexer_node_with_n_inputs() {
     using namespace oneapi::tbb::flow;
-    graph g;
-    int message = 42;
-
-    using submitter_type = function_node<int, int>;
-
-    std::vector<submitter_type> submitters;
-    submitters.reserve(NInputs);
-    for (std::size_t i = 0; i < NInputs; ++i) {
-        submitters.emplace_back(g, unlimited, [](int obj) { return obj; });
-    }
-
     using indexer_type = indexer_node_type_generator_t<NInputs>;
+    int message = 42;
 
+    graph g;
+    
+    broadcast_node<int> submitter(g);
     indexer_type indexer(g);
 
-    edge_maker<NInputs>::make(submitters, indexer);
-
     using output_type = typename indexer_type::output_type;
 
-    std::unordered_set<std::size_t> indices;
-    indices.reserve(NInputs);
+    std::unordered_set<std::size_t> tags;
 
     function_node<output_type> receiver(g, serial, [&](const output_type& indexer_output) {
-        indices.emplace(indexer_output.tag());
-        CHECK_MESSAGE(cast_to<int>(indexer_output) == message, "invalid message returned from indexer_node");
+        auto result = tags.emplace(indexer_output.tag());
+        CHECK_MESSAGE(result.second, "Duplicated tags returned from the indexer_node");
+        CHECK_MESSAGE(cast_to<int>(indexer_output) == message, "Invalid message returned from indexer node");
     });
 
+    edge_maker<NInputs>::make(submitter, indexer);
     make_edge(indexer, receiver);
 
-    for (auto& submitter : submitters) {
-        submitter.try_put(message);
-    }
+    submitter.try_put(message);
     g.wait_for_all();
 
-    CHECK_MESSAGE(indices.size() == NInputs, "Message from some port lost");
+    CHECK_MESSAGE(tags.size() == NInputs, "Incorrect number of tags returned from the indexer_node");
     for (std::size_t i = 0; i < NInputs; ++i) {
-        CHECK(indices.find(i) != indices.end());
+        CHECK_MESSAGE(tags.count(i) == 1, "Some tag was not returned from indexer_node");
     }
 }
 
-TEST_CASE("indexer_node with large number of inputs") {
+//! \brief \ref interface \ref requirement
+TEST_CASE("indexer_node with large number of input ports") {
     test_indexer_node_with_n_inputs<50>();
-}
+}