/* Copyright (c) 2005-2021 Intel Corporation Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. */ #if __INTEL_COMPILER && _MSC_VER #pragma warning(disable : 2586) // decorated name length exceeded, name was truncated #endif #include "common/config.h" #include "tbb/flow_graph.h" #include "common/test.h" #include "common/utils.h" #include "common/graph_utils.h" #include #include #include //! \file test_composite_node.cpp //! \brief Test for [flow_graph.composite_node] specification struct passthru_body { int operator()( int i ) { return i; } }; class my_input_body{ int start; int finish; int step; public: my_input_body(int f, int s) : start(1), finish(f), step(s) {} int operator()(tbb::flow_control& fc) { int a = start; if (start <= finish) { a = start; start+=step; return a; } else { fc.stop(); return int(); }; } }; struct m_fxn_body{ void operator()(int, tbb::flow::multifunction_node >::output_ports_type ) {} }; struct ct_body { ct_body(){} void operator()(tbb::flow::continue_msg){} }; struct seq_body { std::size_t operator()(int i) { return i; } }; template struct compare { static void compare_refs(T1 tuple1, T2 tuple2) { CHECK_MESSAGE( ( &std::get(tuple1) == &std::get(tuple2)), "ports not set correctly"); compare::compare_refs(tuple1, tuple2); } }; template struct compare<1, T1, T2> { static void compare_refs(T1 tuple1, T2 tuple2) { CHECK_MESSAGE( (&std::get<0>(tuple1) == &std::get<0>(tuple2)), "port 0 not correctly set"); } }; struct tiny_node : public tbb::flow::composite_node< std::tuple< int >, std::tuple< int > > { tbb::flow::function_node< int, int > f1; tbb::flow::function_node< int, int > f2; typedef tbb::flow::composite_node< std::tuple< int >, std::tuple< int > > base_type; public: tiny_node(tbb::flow::graph &g, bool hidden = false) : base_type(g), f1(g, tbb::flow::unlimited, passthru_body() ), f2(g, tbb::flow::unlimited, passthru_body() ) { tbb::flow::make_edge(f1, f2); std::tuple& > input_tuple(f1); std::tuple& > output_tuple(f2); base_type::set_external_ports(input_tuple, output_tuple); if(hidden) base_type::add_nodes(f1, f2); else base_type::add_visible_nodes(f1, f2); } }; int test_tiny(bool hidden = false) { tbb::flow::graph g; tbb::flow::function_node< int, int > f0( g, tbb::flow::unlimited, passthru_body() ); tiny_node t(g, hidden); CHECK_MESSAGE( (&tbb::flow::input_port<0>(t) == &t.f1), "f1 not bound to input port 0 in composite_node t"); CHECK_MESSAGE( (&tbb::flow::output_port<0>(t) == &t.f2), "f2 not bound to output port 0 in composite_node t"); tiny_node t1(g, hidden); CHECK_MESSAGE( (&std::get<0>(t1.input_ports()) == &t1.f1), "f1 not bound to input port 0 in composite_node t1"); CHECK_MESSAGE( (&std::get<0>(t1.output_ports()) == &t1.f2), "f2 not bound to output port 0 in composite_node t1"); test_input_ports_return_ref(t1); test_output_ports_return_ref(t1); tiny_node t2(g, hidden); CHECK_MESSAGE( (&tbb::flow::input_port<0>(t2) == &t2.f1), "f1 not bound to input port 0 in composite_node t2"); CHECK_MESSAGE( (&tbb::flow::output_port<0>(t2) == &t2.f2), "f2 not bound to output port 0 in composite_node t2"); tbb::flow::function_node< int, int > f3( g, tbb::flow::unlimited, passthru_body() ); tbb::flow::make_edge( f0, t ); tbb::flow::make_edge( t, t1 ); tbb::flow::make_edge( t1, t2 ); tbb::flow::make_edge( t2 , f3 ); tbb::flow::queue_node q(g); tbb::flow::make_edge(f3, q); f0.try_put(1); g.wait_for_all(); int i, j =0; q.try_get(i); CHECK_MESSAGE( ( i == 1), "item did not go through graph"); q.try_get(j); CHECK_MESSAGE( ( !j), "unexpected item in graph"); g.wait_for_all(); tbb::flow::remove_edge(f3, q); tbb::flow::remove_edge(t2, f3); tbb::flow::remove_edge(t1, t2); tbb::flow::make_edge( t1 , f3 ); tbb::flow::make_edge(f3, q); f0.try_put(2); g.wait_for_all(); q.try_get(i); CHECK_MESSAGE( ( i == 2), "item did not go through graph after removal of edge"); q.try_get(j); CHECK_MESSAGE( ( !j), "unexpected item in graph after removal of edge"); return 0; } class adder_node : public tbb::flow::composite_node< std::tuple< int, int >, std::tuple< int > > { public: tbb::flow::join_node< std::tuple< int, int >, tbb::flow::queueing > j; tbb::flow::function_node< std::tuple< int, int >, int > f; private: typedef tbb::flow::composite_node< std::tuple< int, int >, std::tuple< int > > base_type; struct f_body { int operator()( const std::tuple< int, int > &t ) { return std::get<0>(t) + std::get<1>(t); } }; public: adder_node(tbb::flow::graph &g, bool hidden = false) : base_type(g), j(g), f(g, tbb::flow::unlimited, f_body() ) { tbb::flow::make_edge( j, f ); base_type::set_external_ports(base_type::input_ports_type(tbb::flow::input_port<0>(j), tbb::flow::input_port<1>(j)), base_type::output_ports_type(f)); if (hidden) base_type::add_nodes(j, f); else base_type::add_visible_nodes(j, f); } }; struct square_body { int operator()(int v) { return v*v; } }; struct cube_body { int operator()(int v) { return v*v*v; } }; int adder_sum(int i) { return (int)(pow(3*pow(i,3) + pow(i, 2),2)); } int test_adder(bool hidden = false) { tbb::flow::graph g; tbb::flow::function_node s(g, tbb::flow::unlimited, square_body()); tbb::flow::function_node c(g, tbb::flow::unlimited, cube_body()); tbb::flow::function_node p(g, tbb::flow::unlimited, passthru_body()); adder_node a0(g, hidden); CHECK_MESSAGE( (&tbb::flow::input_port<0>(a0) == &tbb::flow::input_port<0>(a0.j)), "input_port 0 of j not bound to input port 0 in composite_node a0"); CHECK_MESSAGE( (&tbb::flow::input_port<1>(a0) == &tbb::flow::input_port<1>(a0.j)), "input_port 1 of j not bound to input port 1 in composite_node a0"); CHECK_MESSAGE( (&tbb::flow::output_port<0>(a0) == &a0.f), "f not bound to output port 0 in composite_node a0"); adder_node a1(g, hidden); CHECK_MESSAGE( (&std::get<0>(a0.input_ports()) == &tbb::flow::input_port<0>(a0.j)), "input_port 0 of j not bound to input port 0 in composite_node a1"); CHECK_MESSAGE( (&std::get<1>(a0.input_ports()) == &tbb::flow::input_port<1>(a0.j)), "input_port1 of j not bound to input port 1 in composite_node a1"); CHECK_MESSAGE( (&std::get<0>(a0.output_ports()) == &a0.f), "f not bound to output port 0 in composite_node a1"); adder_node a2(g, hidden); CHECK_MESSAGE( (&tbb::flow::input_port<0>(a2) == &tbb::flow::input_port<0>(a2.j)), "input_port 0 of j not bound to input port 0 in composite_node a2"); CHECK_MESSAGE( (&tbb::flow::input_port<1>(a2) == &tbb::flow::input_port<1>(a2.j)), "input_port 1 of j not bound to input port 1 in composite_node a2"); CHECK_MESSAGE( (&tbb::flow::output_port<0>(a2) == &a2.f), "f not bound to output port 0 in composite_node a2"); adder_node a3(g, hidden); CHECK_MESSAGE( (&std::get<0>(a3.input_ports()) == &tbb::flow::input_port<0>(a3.j)), "input_port 0 of j not bound to input port 0 in composite_node a3"); CHECK_MESSAGE( (&std::get<1>(a3.input_ports()) == &tbb::flow::input_port<1>(a3.j)), "input_port1 of j not bound to input port 1 in composite_node a3"); CHECK_MESSAGE( (&std::get<0>(a3.output_ports()) == &a3.f), "f not bound to output port 0 in composite_node a3"); tbb::flow::function_node s2(g, tbb::flow::unlimited, square_body()); tbb::flow::queue_node q(g); tbb::flow::make_edge( s, tbb::flow::input_port<0>(a0) ); tbb::flow::make_edge( c, tbb::flow::input_port<1>(a0) ); tbb::flow::make_edge( c, tbb::flow::input_port<0>(a1) ); tbb::flow::make_edge( c, tbb::flow::input_port<1>(a1) ); tbb::flow::make_edge( tbb::flow::output_port<0>(a0), tbb::flow::input_port<0>(a2) ); tbb::flow::make_edge( tbb::flow::output_port<0>(a1), tbb::flow::input_port<1>(a2) ); tbb::flow::make_edge( tbb::flow::output_port<0>(a2), s2 ); tbb::flow::make_edge( s2, q ); int sum_total=0; int result=0; for ( int i = 1; i < 4; ++i ) { s.try_put(i); c.try_put(i); sum_total += adder_sum(i); g.wait_for_all(); } int j; for ( int i = 1; i < 4; ++i ) { q.try_get(j); result += j; } g.wait_for_all(); CHECK_MESSAGE( (result == sum_total), "the sum from the graph does not match the calculated value"); tbb::flow::remove_edge(s2, q); tbb::flow::remove_edge( a2, s2 ); tbb::flow::make_edge( a0, a3 ); tbb::flow::make_edge( a1, tbb::flow::input_port<1>(a3) ); tbb::flow::make_edge( a3, s2 ); tbb::flow::make_edge( s2, q ); sum_total=0; result=0; for ( int i = 10; i < 20; ++i ) { s.try_put(i); c.try_put(i); sum_total += adder_sum(i); g.wait_for_all(); } for ( int i = 10; i < 20; ++i ) { q.try_get(j); result += j; } g.wait_for_all(); CHECK_MESSAGE( (result == sum_total), "the new sum after the replacement of the nodes does not match the calculated value"); return 0; } /* outer composite node (outer_node) |-------------------------------------------------------------------| | | | |------------------| |------------------| |------------------| | |---------------------| |--| inner composite | /| inner composite | /| inner composite | | |-------------------| |broadcast node(input)|/| | node |/ | node |/ | node |-+-| queue node(output)| |---------------------|\| |(inner_node1) |\ | (inner_node2) |\ | (inner_node3) | | |-------------------| |--| | \| | \| | | | |------------------| |------------------| |------------------| | | | |-------------------------------------------------------------------| */ int test_nested_adder(bool hidden=false) { tbb::flow::graph g; tbb::flow::composite_node, std::tuple > outer_node(g); typedef tbb::flow::composite_node, std::tuple > base_type; tbb::flow::broadcast_node input(g); tbb::flow::queue_node output(g); adder_node inner_node1(g, hidden); adder_node inner_node2(g, hidden); adder_node inner_node3(g, hidden); outer_node.set_external_ports(base_type::input_ports_type(tbb::flow::input_port<0>(inner_node1), tbb::flow::input_port<1>(inner_node1)), base_type::output_ports_type(tbb::flow::output_port<0>(inner_node3))); CHECK_MESSAGE( (&tbb::flow::input_port<0>(outer_node) == &tbb::flow::input_port<0>(inner_node1)), "input port 0 of inner_node1 not bound to input port 0 in outer_node"); CHECK_MESSAGE( (&tbb::flow::input_port<1>(outer_node) == &tbb::flow::input_port<1>(inner_node1)), "input port 1 of inner_node1 not bound to input port 1 in outer_node"); CHECK_MESSAGE( (&tbb::flow::output_port<0>(outer_node) == &tbb::flow::output_port<0>(inner_node3)), "output port 0 of inner_node3 not bound to output port 0 in outer_node"); tbb::flow::make_edge(input, tbb::flow::input_port<0>(outer_node)/*inner_node1*/); tbb::flow::make_edge(input, tbb::flow::input_port<1>(outer_node)/*inner_node1*/); tbb::flow::make_edge(inner_node1, tbb::flow::input_port<0>(inner_node2)); tbb::flow::make_edge(inner_node1, tbb::flow::input_port<1>(inner_node2)); tbb::flow::make_edge(inner_node2, tbb::flow::input_port<0>(inner_node3)); tbb::flow::make_edge(inner_node2, tbb::flow::input_port<1>(inner_node3)); tbb::flow::make_edge(outer_node/*inner_node3*/, output); if(hidden) outer_node.add_nodes(inner_node1, inner_node2, inner_node3); else outer_node.add_visible_nodes(inner_node1, inner_node2, inner_node3); int out; for (int i = 1; i < 200000; ++i) { input.try_put(i); g.wait_for_all(); output.try_get(out); CHECK_MESSAGE( (tbb::flow::output_port<0>(outer_node).try_get(out) == output.try_get(out)), "output from outer_node does not match output from graph"); CHECK_MESSAGE( (out == 8*i), "output from outer_node not correct"); } g.wait_for_all(); return 0; } template< typename T > class prefix_node : public tbb::flow::composite_node< std::tuple< T, T, T, T, T >, std::tuple< T, T, T, T, T > > { typedef std::tuple< T, T, T, T, T > my_tuple_t; public: tbb::flow::join_node< my_tuple_t, tbb::flow::queueing > j; tbb::flow::split_node< my_tuple_t > s; private: tbb::flow::function_node< my_tuple_t, my_tuple_t > f; typedef tbb::flow::composite_node< my_tuple_t, my_tuple_t > base_type; struct f_body { my_tuple_t operator()( const my_tuple_t &t ) { return my_tuple_t( std::get<0>(t), std::get<0>(t) + std::get<1>(t), std::get<0>(t) + std::get<1>(t) + std::get<2>(t), std::get<0>(t) + std::get<1>(t) + std::get<2>(t) + std::get<3>(t), std::get<0>(t) + std::get<1>(t) + std::get<2>(t) + std::get<3>(t) + std::get<4>(t) ); } }; public: prefix_node(tbb::flow::graph &g, bool hidden = false ) : base_type(g), j(g), s(g), f(g, tbb::flow::serial, f_body() ) { tbb::flow::make_edge( j, f ); tbb::flow::make_edge( f, s ); typename base_type::input_ports_type input_tuple(tbb::flow::input_port<0>(j), tbb::flow::input_port<1>(j), tbb::flow::input_port<2>(j), tbb::flow::input_port<3>(j), tbb::flow::input_port<4>(j)); typename base_type::output_ports_type output_tuple(tbb::flow::output_port<0>(s), tbb::flow::output_port<1>(s), tbb::flow::output_port<2>(s), tbb::flow::output_port<3>(s), tbb::flow::output_port<4>(s)); base_type::set_external_ports(input_tuple, output_tuple); if(hidden) base_type::add_nodes(j,s,f); else base_type::add_visible_nodes(j,s,f); } }; int test_prefix(bool hidden = false) { tbb::flow::graph g; prefix_node p(g, hidden); CHECK_MESSAGE( (&std::get<0>(p.input_ports()) == &tbb::flow::input_port<0>(p.j)), "input port 0 of j is not bound to input port 0 of composite node p"); CHECK_MESSAGE( (&tbb::flow::input_port<1>(p.j) == &tbb::flow::input_port<1>(p.j)), "input port 1 of j is not bound to input port 1 of composite node p"); CHECK_MESSAGE( (&std::get<2>(p.input_ports()) == &tbb::flow::input_port<2>(p.j)), "input port 2 of j is not bound to input port 2 of composite node p"); CHECK_MESSAGE( (&tbb::flow::input_port<3>(p.j) == &tbb::flow::input_port<3>(p.j)), "input port 3 of j is not bound to input port 3 of composite node p"); CHECK_MESSAGE( (&std::get<4>(p.input_ports()) == &tbb::flow::input_port<4>(p.j)), "input port 4 of j is not bound to input port 4 of composite node p"); CHECK_MESSAGE( (&std::get<0>(p.output_ports()) == &tbb::flow::output_port<0>(p.s)), "output port 0 of s is not bound to output port 0 of composite node p"); CHECK_MESSAGE( (&tbb::flow::output_port<1>(p.s) == &tbb::flow::output_port<1>(p.s)), "output port 1 of s is not bound to output port 1 of composite node p"); CHECK_MESSAGE( (&std::get<2>(p.output_ports()) == &tbb::flow::output_port<2>(p.s)), "output port 2 of s is not bound to output port 2 of composite node p"); CHECK_MESSAGE( (&tbb::flow::output_port<3>(p.s) == &tbb::flow::output_port<3>(p.s)), "output port 3 of s is not bound to output port 3 of composite node p"); CHECK_MESSAGE( (&std::get<4>(p.output_ports()) == &tbb::flow::output_port<4>(p.s)), "output port 4 of s is not bound to output port 4 of composite node p"); std::vector< tbb::flow::queue_node > v( 5, tbb::flow::queue_node(g) ); tbb::flow::make_edge( tbb::flow::output_port<0>(p), v[0] ); tbb::flow::make_edge( tbb::flow::output_port<1>(p), v[1] ); tbb::flow::make_edge( tbb::flow::output_port<2>(p), v[2] ); tbb::flow::make_edge( tbb::flow::output_port<3>(p), v[3] ); tbb::flow::make_edge( tbb::flow::output_port<4>(p), v[4] ); for( double offset = 1; offset < 10000; offset *= 10 ) { tbb::flow::input_port<0>(p).try_put( offset ); tbb::flow::input_port<1>(p).try_put( offset + 1 ); tbb::flow::input_port<2>(p).try_put( offset + 2 ); tbb::flow::input_port<3>(p).try_put( offset + 3 ); tbb::flow::input_port<4>(p).try_put( offset + 4 ); } g.wait_for_all(); double x; while ( v[0].try_get(x) ) { g.wait_for_all(); for ( int i = 1; i < 5; ++i ) { v[i].try_get(x); g.wait_for_all(); } } return 0; } struct input_only_output_only_seq { std::size_t operator()(int i) { CHECK(i > 0); return std::size_t((i + 3) / 4 - 1); } }; void input_only_output_only_composite(bool hidden) { tbb::flow::graph g; tbb::flow::composite_node, std::tuple > input_output(g); typedef tbb::flow::composite_node, std::tuple<> > input_only_composite; typedef tbb::flow::composite_node, std::tuple > output_only_composite; typedef tbb::flow::input_node src_type; typedef tbb::flow::queue_node q_type; typedef tbb::flow::function_node f_type; typedef tbb::flow::sequencer_node sequencer_type; int num = 0; int finish=1000; int step = 4; input_only_composite a_in(g); output_only_composite a_out(g); src_type src(g, my_input_body(finish, step)); q_type que(g); f_type f(g, 1, passthru_body()); // Sequencer_node is needed, because serial function_node guarantees only serial body execution, // not a sequential order of messages dispatch sequencer_type seq(g, input_only_output_only_seq()); std::tuple input_tuple(f); a_in.set_external_ports(input_tuple); CHECK_MESSAGE( (&std::get<0>(a_in.input_ports()) == &f), "f not bound to input port 0 in composite_node a_in"); std::tuple output_tuple(src); a_out.set_external_ports(output_tuple); CHECK_MESSAGE( (&std::get<0>(a_out.output_ports()) == &src), "src not bound to output port 0 in composite_node a_out"); if(hidden) { a_in.add_nodes(f, seq, que); a_out.add_nodes(src); } else { a_in.add_visible_nodes(f, seq, que); a_out.add_visible_nodes(src); } tbb::flow::make_edge(a_out, a_in); tbb::flow::make_edge(f, seq); tbb::flow::make_edge(seq, que); src.activate(); g.wait_for_all(); for(int i = 1; i