1 /* 2 Copyright (c) 2020-2023 Intel Corporation 3 4 Licensed under the Apache License, Version 2.0 (the "License"); 5 you may not use this file except in compliance with the License. 6 You may obtain a copy of the License at 7 8 http://www.apache.org/licenses/LICENSE-2.0 9 10 Unless required by applicable law or agreed to in writing, software 11 distributed under the License is distributed on an "AS IS" BASIS, 12 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 13 See the License for the specific language governing permissions and 14 limitations under the License. 15 */ 16 17 #if __INTEL_COMPILER && _MSC_VER 18 #pragma warning(disable : 2586) // decorated name length exceeded, name was truncated 19 #endif 20 21 #include "conformance_flowgraph.h" 22 #include "common/test_invoke.h" 23 24 //! \file conformance_async_node.cpp 25 //! \brief Test for [flow_graph.async_node] specification 26 27 using input_msg = conformance::message</*default_ctor*/true, /*copy_ctor*/true, /*copy_assign*/false>; 28 using output_msg = conformance::message</*default_ctor*/false, /*copy_ctor*/false, /*copy_assign*/false>; 29 30 //! Test async_node constructors 31 //! \brief \ref requirement 32 TEST_CASE("async_node constructors"){ 33 using namespace oneapi::tbb::flow; 34 graph g; 35 36 conformance::dummy_functor<int> fun; 37 38 async_node<int, int> fn1(g, unlimited, fun); 39 async_node<int, int> fn2(g, unlimited, fun, oneapi::tbb::flow::node_priority_t(1)); 40 41 async_node<int, int, lightweight> lw_node1(g, serial, fun, lightweight()); 42 async_node<int, int, lightweight> lw_node2(g, serial, fun, lightweight(), oneapi::tbb::flow::node_priority_t(1)); 43 } 44 45 //! Test buffering property 46 //! \brief \ref requirement 47 TEST_CASE("async_node buffering") { 48 conformance::dummy_functor<int> fun; 49 conformance::test_buffering<oneapi::tbb::flow::async_node<input_msg, int>, input_msg>(oneapi::tbb::flow::unlimited, fun); 50 } 51 52 //! Test priorities work in single-threaded configuration 53 //! \brief \ref requirement 54 TEST_CASE("async_node priority support"){ 55 conformance::test_priority<oneapi::tbb::flow::async_node<input_msg, int>, input_msg>(oneapi::tbb::flow::unlimited); 56 } 57 58 //! The node that is constructed has a reference to the same graph object as src, has a copy of the initial body used by src, and has the same concurrency threshold as src. 59 //! The predecessors and successors of src are not copied. 60 //! \brief \ref requirement 61 TEST_CASE("async_node copy constructor"){ 62 conformance::test_copy_ctor<oneapi::tbb::flow::async_node<int, int>>(); 63 } 64 65 //! Test calling async body 66 //! \brief \ref interface \ref requirement 67 TEST_CASE("Test async_node body") { 68 conformance::test_body_exec<oneapi::tbb::flow::async_node<input_msg, output_msg>, input_msg, output_msg>(oneapi::tbb::flow::unlimited); 69 } 70 71 //! Test async_node inheritance relations 72 //! \brief \ref interface 73 TEST_CASE("async_node superclasses"){ 74 conformance::test_inheritance<oneapi::tbb::flow::async_node<int, int>, int, int>(); 75 conformance::test_inheritance<oneapi::tbb::flow::async_node<void*, float>, void*, float>(); 76 conformance::test_inheritance<oneapi::tbb::flow::async_node<input_msg, output_msg>, input_msg, output_msg>(); 77 } 78 79 //! Test node broadcast messages to successors 80 //! \brief \ref requirement 81 TEST_CASE("async_node broadcast"){ 82 conformance::counting_functor<int> fun(conformance::expected); 83 conformance::test_forwarding<oneapi::tbb::flow::async_node<input_msg, int>, input_msg, int>(1, oneapi::tbb::flow::unlimited, fun); 84 } 85 86 //! Test async_node has a user-settable concurrency limit. It can be set to one of predefined values. 87 //! The user can also provide a value of type std::size_t to limit concurrency. 88 //! Test that not more than limited threads works in parallel. 89 //! \brief \ref requirement 90 TEST_CASE("concurrency follows set limits"){ 91 conformance::test_concurrency<oneapi::tbb::flow::async_node<int, int>>(); 92 } 93 94 //! Test body copying and copy_body logic 95 //! Test the body object passed to a node is copied 96 //! \brief \ref interface 97 TEST_CASE("async_node body copying"){ 98 conformance::test_copy_body_function<oneapi::tbb::flow::async_node<int, int>, conformance::copy_counting_object<int>>(oneapi::tbb::flow::unlimited); 99 } 100 101 //! Test node reject the incoming message if the concurrency limit achieved. 102 //! \brief \ref interface 103 TEST_CASE("async_node with rejecting policy"){ 104 conformance::test_rejecting<oneapi::tbb::flow::async_node<int, int, oneapi::tbb::flow::rejecting>>(); 105 } 106 107 //! Test node Input class meet the DefaultConstructible and CopyConstructible requirements and Output class meet the CopyConstructible requirements. 108 //! \brief \ref interface \ref requirement 109 TEST_CASE("Test async_node Output and Input class") { 110 using Body = conformance::copy_counting_object<int>; 111 conformance::test_output_input_class<oneapi::tbb::flow::async_node<Body, Body>, Body>(); 112 } 113 114 //! Test the body of assync_node typically submits the messages to an external activity for processing outside of the graph. 115 //! \brief \ref interface 116 TEST_CASE("async_node with rejecting policy"){ 117 using async_node_type = tbb::flow::async_node<int, int>; 118 using gateway_type = async_node_type::gateway_type; 119 120 oneapi::tbb::flow::graph g; 121 std::atomic<bool> flag{false}; 122 std::thread thr; 123 async_node_type testing_node{ 124 g, tbb::flow::unlimited, 125 [&](const int& input, gateway_type& gateway) { 126 gateway.reserve_wait(); 127 thr = std::thread{[&]{ 128 flag = true; 129 gateway.try_put(input); 130 gateway.release_wait(); 131 }}; 132 } 133 }; 134 135 testing_node.try_put(1); 136 g.wait_for_all(); 137 CHECK_MESSAGE((flag.load()), "The body of assync_node must submits the messages to an external activity for processing outside of the graph"); 138 thr.join(); 139 } 140 141 #if __TBB_CPP17_INVOKE_PRESENT 142 //! Test that async_node uses std::invoke to run the body 143 //! \brief \ref requirement 144 TEST_CASE("async_node and std::invoke") { 145 using namespace oneapi::tbb::flow; 146 147 using start_node_type = function_node<std::size_t, test_invoke::SmartID<std::size_t>>; 148 using async_node_type = async_node<test_invoke::SmartID<std::size_t>, std::size_t>; 149 150 auto async_body = &test_invoke::SmartID<std::size_t>::template send_id_to_gateway<typename async_node_type::gateway_type>; 151 152 graph g; 153 start_node_type starter(g, serial, [](std::size_t i) -> test_invoke::SmartID<std::size_t> { return {i}; }); 154 async_node_type activity_submitter(g, serial, async_body); 155 buffer_node<std::size_t> buf(g); 156 157 make_edge(starter, activity_submitter); 158 make_edge(activity_submitter, buf); 159 160 starter.try_put(1); 161 162 g.wait_for_all(); 163 std::size_t result = 0; 164 CHECK(buf.try_get(result)); 165 CHECK(result == 1); 166 CHECK(!buf.try_get(result)); 167 } 168 #endif 169