1 /* 2 Copyright (c) 2020-2021 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 #ifndef __TBB_test_conformance_conformance_flowgraph_H 18 #define __TBB_test_conformance_conformance_flowgraph_H 19 20 #include "common/test.h" 21 #include "common/utils.h" 22 #include "common/graph_utils.h" 23 #include "common/concurrency_tracker.h" 24 25 #include "oneapi/tbb/flow_graph.h" 26 #include "oneapi/tbb/task_arena.h" 27 #include "oneapi/tbb/global_control.h" 28 29 namespace conformance { 30 31 constexpr int expected = 5; 32 33 template<typename V> 34 using test_push_receiver = oneapi::tbb::flow::queue_node<V>; 35 36 template<typename Input, typename Output = Input> 37 using multifunc_ports_t = 38 typename oneapi::tbb::flow::multifunction_node<Input, std::tuple<Output>>::output_ports_type; 39 40 template<typename Input, typename Output = Input> 41 using async_ports_t = 42 typename oneapi::tbb::flow::async_node<Input, Output>::gateway_type; 43 44 template<bool DefaultConstructible, bool CopyConstructible, bool CopyAssignable> 45 struct message { 46 int data; 47 48 message(int _data) : data(_data) {}; 49 50 template<bool T = DefaultConstructible, typename = typename std::enable_if<T>::type> 51 message(){}; 52 53 template<bool T = CopyConstructible, typename = typename std::enable_if<T>::type> 54 message(const message& msg) : data(msg.data) {}; 55 56 template<bool T = CopyAssignable, typename = typename std::enable_if<T>::type> 57 message& operator=(const message& msg) { 58 this->data = msg.data; 59 return *this; 60 }; 61 62 bool operator==(const int expected_data) const { 63 return data == expected_data; 64 } 65 66 bool operator==(const message& msg) const { 67 return data == msg.data; 68 } 69 70 operator std::size_t() const { 71 return static_cast<std::size_t>(data); 72 } 73 74 operator int() const { 75 return data; 76 } 77 }; 78 79 template<typename V> 80 typename std::enable_if<!std::is_default_constructible<V>::value, std::vector<V>>::type get_values( test_push_receiver<V>& rr ) { 81 std::vector<V> messages; 82 int val = 0; 83 for(V tmp(0); rr.try_get(tmp); ++val) { 84 messages.push_back(tmp); 85 } 86 return messages; 87 } 88 89 template<typename V> 90 typename std::enable_if<std::is_default_constructible<V>::value, std::vector<V>>::type get_values( test_push_receiver<V>& rr ) { 91 std::vector<V> messages; 92 int val = 0; 93 for(V tmp; rr.try_get(tmp); ++val) { 94 messages.push_back(tmp); 95 } 96 return messages; 97 } 98 99 template<typename Node, typename InputType = void> 100 bool produce_messages(Node& node, int arg = 1) { 101 utils::suppress_unused_warning(arg); 102 #if defined CONFORMANCE_INPUT_NODE 103 node.activate(); 104 return true; 105 #elif defined CONFORMANCE_CONTINUE_NODE 106 return node.try_put(InputType()); 107 #else 108 return node.try_put(InputType(arg)); 109 #endif 110 } 111 112 template<typename T, typename U> 113 typename std::enable_if<std::is_same<T, U>::value, bool>::type check_output_type(){ 114 return true; 115 } 116 117 template<typename T, typename U> 118 typename std::enable_if<!std::is_same<T, U>::value, bool>::type check_output_type(){ 119 return false; 120 } 121 122 template<typename T> 123 struct sequencer_functor { 124 struct seq_message { 125 std::size_t id; 126 T data; 127 }; 128 129 using input_type = T; 130 131 std::size_t operator()(T v) { 132 return v; 133 } 134 135 std::size_t operator()(seq_message msg) { 136 return msg.id; 137 } 138 }; 139 140 template<typename OutputType> 141 struct track_first_id_functor { 142 int my_id; 143 static std::atomic<int> first_id; 144 145 track_first_id_functor(int id) : my_id(id) {} 146 147 OutputType operator()( OutputType argument ) { 148 int old_value = -1; 149 while(first_id == -1 && 150 !first_id.compare_exchange_strong(old_value, my_id)); 151 return argument; 152 } 153 154 template<typename InputType> 155 OutputType operator()( InputType& ) { 156 return operator()(OutputType(0)); 157 } 158 159 template<typename InputType> 160 void operator()( InputType, async_ports_t<InputType, OutputType>& g ) { 161 g.try_put(operator()(OutputType(0))); 162 } 163 164 template<typename InputType> 165 void operator()( InputType, multifunc_ports_t<InputType, OutputType>& op ) { 166 std::get<0>(op).try_put(operator()(OutputType(0))); 167 } 168 }; 169 170 template<typename OutputType> 171 std::atomic<int> track_first_id_functor<OutputType>::first_id = {-1}; 172 173 template<typename OutputType> 174 struct counting_functor { 175 OutputType return_value; 176 177 static std::atomic<std::size_t> execute_count; 178 179 counting_functor( OutputType value = OutputType(0) ) : return_value(value) { 180 execute_count = 0; 181 } 182 183 counting_functor( const counting_functor & c ) : return_value(static_cast<int>(c.return_value)) { 184 execute_count = 0; 185 } 186 187 template<typename InputType> 188 OutputType operator()( InputType ) { 189 ++execute_count; 190 return return_value; 191 } 192 193 template<typename InputType> 194 void operator()( InputType, multifunc_ports_t<InputType, OutputType>& op ) { 195 ++execute_count; 196 std::get<0>(op).try_put(return_value); 197 } 198 199 OutputType operator()( oneapi::tbb::flow_control& fc ) { 200 ++execute_count; 201 if(execute_count > std::size_t(return_value)) { 202 fc.stop(); 203 return return_value; 204 } 205 return return_value; 206 } 207 208 template<typename InputType> 209 void operator()( InputType, async_ports_t<InputType, OutputType>& g ) { 210 ++execute_count; 211 g.try_put(return_value); 212 } 213 }; 214 215 template<typename OutputType> 216 std::atomic<std::size_t> counting_functor<OutputType>::execute_count = {0}; 217 218 template<typename OutputType> 219 struct dummy_functor { 220 template<typename InputType> 221 OutputType operator()( InputType ) { 222 #ifdef CONFORMANCE_CONTINUE_NODE 223 return OutputType(); 224 #else 225 return OutputType(0); 226 #endif 227 } 228 229 template<typename InputType> 230 void operator()( InputType, multifunc_ports_t<InputType, OutputType>& op ) { 231 std::get<0>(op).try_put(OutputType(0)); 232 } 233 234 template<typename InputType> 235 void operator()( InputType, async_ports_t<InputType, OutputType>& g ) { 236 g.try_put(OutputType(0)); 237 } 238 239 template<typename InputType, typename T> 240 void operator()( InputType, std::tuple<T, T>& ) {} 241 242 OutputType operator()( oneapi::tbb::flow_control & fc ) { 243 static bool check = false; 244 if(check) { 245 check = false; 246 fc.stop(); 247 return OutputType(1); 248 } 249 check = true; 250 return OutputType(1); 251 } 252 }; 253 254 struct wait_flag_body { 255 static std::atomic<bool> flag; 256 257 wait_flag_body() { 258 flag.store(false); 259 } 260 261 template<typename InputType> 262 InputType operator()( InputType ) { 263 while(!flag.load()) { utils::yield(); }; 264 #ifdef CONFORMANCE_CONTINUE_NODE 265 return InputType(); 266 #else 267 return InputType(0); 268 #endif 269 } 270 271 template<typename InputType> 272 void operator()( InputType argument, multifunc_ports_t<InputType>& op ) { 273 while(!flag.load()) { }; 274 std::get<0>(op).try_put(argument); 275 } 276 277 template<typename InputType> 278 void operator()( InputType argument, async_ports_t<InputType>& g ) { 279 while(!flag.load()) { }; 280 g.try_put(argument); 281 } 282 }; 283 284 std::atomic<bool> wait_flag_body::flag{false}; 285 286 struct concurrency_peak_checker_body { 287 std::size_t required_max_concurrency = 0; 288 289 concurrency_peak_checker_body( std::size_t req_max_concurrency = 0 ) : 290 required_max_concurrency(req_max_concurrency) {} 291 292 concurrency_peak_checker_body( const concurrency_peak_checker_body & ) = default; 293 294 int operator()( oneapi::tbb::flow_control & fc ) { 295 static int counter = 0; 296 utils::ConcurrencyTracker ct; 297 if(++counter > 500) { 298 counter = 0; 299 fc.stop(); 300 return 1; 301 } 302 utils::doDummyWork(1000); 303 CHECK_MESSAGE((int)utils::ConcurrencyTracker::PeakParallelism() <= required_max_concurrency, 304 "Input node is serial and its body never invoked concurrently"); 305 return 1; 306 } 307 308 int operator()( int ) { 309 utils::ConcurrencyTracker ct; 310 utils::doDummyWork(1000); 311 CHECK_MESSAGE((int)utils::ConcurrencyTracker::PeakParallelism() <= required_max_concurrency, 312 "Measured parallelism is not expected"); 313 return 1; 314 } 315 316 void operator()( const int& argument, multifunc_ports_t<int>& op ) { 317 utils::ConcurrencyTracker ct; 318 utils::doDummyWork(1000); 319 CHECK_MESSAGE((int)utils::ConcurrencyTracker::PeakParallelism() <= required_max_concurrency, 320 "Measured parallelism is not expected"); 321 std::get<0>(op).try_put(argument); 322 } 323 324 void operator()( const int& argument , async_ports_t<int>& g ) { 325 utils::ConcurrencyTracker ct; 326 utils::doDummyWork(1000); 327 CHECK_MESSAGE((int)utils::ConcurrencyTracker::PeakParallelism() <= required_max_concurrency, 328 "Measured parallelism is not expected"); 329 g.try_put(argument); 330 } 331 }; 332 333 template<typename OutputType, typename InputType = int> 334 struct copy_counting_object { 335 std::size_t copy_count;/*increases on every new copied object*/ 336 mutable std::size_t copies_count;/*count number of objects copied from this object*/ 337 std::size_t assign_count; 338 bool is_copy; 339 340 copy_counting_object(): 341 copy_count(0), copies_count(0), assign_count(0), is_copy(false) {} 342 343 copy_counting_object(int): 344 copy_count(0), copies_count(0), assign_count(0), is_copy(false) {} 345 346 copy_counting_object( const copy_counting_object<OutputType, InputType>& other ): 347 copy_count(other.copy_count + 1), is_copy(true) { 348 ++other.copies_count; 349 } 350 351 copy_counting_object& operator=( const copy_counting_object<OutputType, InputType>& other ) { 352 assign_count = other.assign_count + 1; 353 is_copy = true; 354 return *this; 355 } 356 357 OutputType operator()( InputType ) { 358 return OutputType(1); 359 } 360 361 void operator()( InputType, multifunc_ports_t<InputType,OutputType>& op ) { 362 std::get<0>(op).try_put(OutputType(1)); 363 } 364 365 void operator()( InputType , async_ports_t<InputType, OutputType>& g) { 366 g.try_put(OutputType(1)); 367 } 368 369 OutputType operator()( oneapi::tbb::flow_control & fc ) { 370 static bool check = false; 371 if(check) { 372 check = false; 373 fc.stop(); 374 return OutputType(1); 375 } 376 check = true; 377 return OutputType(1); 378 } 379 }; 380 381 template <typename OutputType = int> 382 struct passthru_body { 383 OutputType operator()( const oneapi::tbb::flow::continue_msg& ) { 384 return OutputType(0); 385 } 386 387 OutputType operator()( const OutputType& i ) { 388 return i; 389 } 390 391 OutputType operator()( oneapi::tbb::flow_control & fc ) { 392 static bool check = false; 393 if(check) { 394 check = false; 395 fc.stop(); 396 return OutputType(0); 397 } 398 check = true; 399 return OutputType(0); 400 } 401 402 void operator()( OutputType argument, multifunc_ports_t<OutputType>& op ) { 403 std::get<0>(op).try_put(argument); 404 } 405 406 void operator()( OutputType argument, async_ports_t<OutputType>& g ) { 407 g.try_put(argument); 408 } 409 }; 410 411 template<typename Node, typename InputType, typename OutputType, typename ...Args> 412 void test_body_exec(Args... node_args) { 413 oneapi::tbb::flow::graph g; 414 counting_functor<OutputType> counting_body; 415 counting_body.execute_count = 0; 416 417 Node testing_node(g, node_args..., counting_body); 418 419 constexpr std::size_t n = 10; 420 for(std::size_t i = 0; i < n; ++i) { 421 CHECK_MESSAGE((produce_messages<Node, InputType>(testing_node) == true), 422 "try_put of first node should return true"); 423 } 424 g.wait_for_all(); 425 426 CHECK_MESSAGE((counting_body.execute_count == n), "Body of the first node needs to be executed N times"); 427 } 428 429 template<typename Node, typename Body, typename ...Args> 430 void test_copy_body_function(Args... node_args) { 431 using namespace oneapi::tbb::flow; 432 433 Body base_body; 434 435 graph g; 436 437 Node testing_node(g, node_args..., base_body); 438 439 Body b2 = copy_body<Body, Node>(testing_node); 440 441 CHECK_MESSAGE((base_body.copy_count + 1 < b2.copy_count), "copy_body and constructor should copy bodies"); 442 } 443 444 template<typename Node, typename InputType, typename ...Args> 445 void test_buffering(Args... node_args) { 446 oneapi::tbb::flow::graph g; 447 448 Node testing_node(g, node_args...); 449 oneapi::tbb::flow::limiter_node<int> rejecter(g, 0); 450 451 oneapi::tbb::flow::make_edge(testing_node, rejecter); 452 453 int tmp = -1; 454 produce_messages<Node, InputType>(testing_node); 455 g.wait_for_all(); 456 457 458 #if defined CONFORMANCE_BUFFERING_NODES || defined CONFORMANCE_INPUT_NODE 459 CHECK_MESSAGE((testing_node.try_get(tmp) == true), "try_get after rejection should succeed"); 460 CHECK_MESSAGE((tmp == 1), "try_get after rejection should set value"); 461 #else 462 #ifdef CONFORMANCE_MULTIFUNCTION_NODE 463 CHECK_MESSAGE((std::get<0>(testing_node.output_ports()).try_get(tmp) == false), "try_get after rejection should not succeed"); 464 #else 465 CHECK_MESSAGE((testing_node.try_get(tmp) == false), "try_get after rejection should not succeed"); 466 #endif 467 CHECK_MESSAGE((tmp == -1), "try_get after rejection should not alter passed value"); 468 #endif 469 } 470 471 472 template<typename Node, typename InputType, typename OutputType = InputType, typename ...Args> 473 void test_forwarding(std::size_t messages_received, Args... node_args) { 474 oneapi::tbb::flow::graph g; 475 476 Node testing_node(g, node_args...); 477 std::vector<std::unique_ptr<test_push_receiver<OutputType>>> receiver_nodes; 478 479 for(std::size_t i = 0; i < 10; ++i) { 480 receiver_nodes.emplace_back(new test_push_receiver<OutputType>(g)); 481 oneapi::tbb::flow::make_edge(testing_node, *receiver_nodes.back()); 482 } 483 484 produce_messages<Node, InputType>(testing_node, expected); 485 486 #ifdef CONFORMANCE_INPUT_NODE 487 CHECK_MESSAGE(expected == messages_received, "For correct execution of test"); 488 #endif 489 490 g.wait_for_all(); 491 for(auto& receiver : receiver_nodes) { 492 auto values = get_values(*receiver); 493 CHECK_MESSAGE((values.size() == messages_received), std::string("Descendant of the node must receive " + std::to_string(messages_received) + " message.")); 494 CHECK_MESSAGE((values[0] == expected), "Value passed is the actual one received."); 495 } 496 } 497 498 template<typename Node, typename ...Args> 499 void test_forwarding_single_push(Args... node_args) { 500 oneapi::tbb::flow::graph g; 501 502 Node testing_node(g, node_args...); 503 test_push_receiver<int> suc_node1(g); 504 test_push_receiver<int> suc_node2(g); 505 506 oneapi::tbb::flow::make_edge(testing_node, suc_node1); 507 oneapi::tbb::flow::make_edge(testing_node, suc_node2); 508 509 testing_node.try_put(0); 510 g.wait_for_all(); 511 512 auto values1 = get_values(suc_node1); 513 auto values2 = get_values(suc_node2); 514 CHECK_MESSAGE((values1.size() != values2.size()), "Only one descendant the node needs to receive"); 515 CHECK_MESSAGE((values1.size() + values2.size() == 1), "All messages need to be received"); 516 517 testing_node.try_put(1); 518 g.wait_for_all(); 519 520 auto values3 = get_values(suc_node1); 521 auto values4 = get_values(suc_node2); 522 CHECK_MESSAGE((values3.size() != values4.size()), "Only one descendant the node needs to receive"); 523 CHECK_MESSAGE((values3.size() + values4.size() == 1), "All messages need to be received"); 524 525 #ifdef CONFORMANCE_QUEUE_NODE 526 CHECK_MESSAGE((values1[0] == 0), "Value passed is the actual one received"); 527 CHECK_MESSAGE((values3[0] == 1), "Value passed is the actual one received"); 528 #else 529 if(values1.size() == 1) { 530 CHECK_MESSAGE((values1[0] == 0), "Value passed is the actual one received"); 531 }else{ 532 CHECK_MESSAGE((values2[0] == 0), "Value passed is the actual one received"); 533 } 534 #endif 535 } 536 537 template<typename Node, typename InputType, typename OutputType> 538 void test_inheritance() { 539 using namespace oneapi::tbb::flow; 540 541 CHECK_MESSAGE((std::is_base_of<graph_node, Node>::value), "Node should be derived from graph_node"); 542 CHECK_MESSAGE((std::is_base_of<receiver<InputType>, Node>::value), "Node should be derived from receiver<Input>"); 543 CHECK_MESSAGE((std::is_base_of<sender<OutputType>, Node>::value), "Node should be derived from sender<Output>"); 544 } 545 546 template<typename Node> 547 void test_copy_ctor() { 548 using namespace oneapi::tbb::flow; 549 graph g; 550 551 dummy_functor<int> fun1; 552 conformance::copy_counting_object<int> fun2; 553 554 Node node0(g, unlimited, fun1); 555 Node node1(g, unlimited, fun2); 556 test_push_receiver<int> suc_node1(g); 557 test_push_receiver<int> suc_node2(g); 558 559 oneapi::tbb::flow::make_edge(node0, node1); 560 oneapi::tbb::flow::make_edge(node1, suc_node1); 561 562 Node node_copy(node1); 563 564 conformance::copy_counting_object<int> b2 = copy_body<conformance::copy_counting_object<int>, Node>(node_copy); 565 566 CHECK_MESSAGE((fun2.copy_count + 1 < b2.copy_count), "constructor should copy bodies"); 567 568 oneapi::tbb::flow::make_edge(node_copy, suc_node2); 569 570 node_copy.try_put(1); 571 g.wait_for_all(); 572 573 CHECK_MESSAGE((get_values(suc_node1).size() == 0 && get_values(suc_node2).size() == 1), "Copied node doesn`t copy successor"); 574 575 node0.try_put(1); 576 g.wait_for_all(); 577 578 CHECK_MESSAGE((get_values(suc_node1).size() == 1 && get_values(suc_node2).size() == 0), "Copied node doesn`t copy predecessor"); 579 } 580 581 template<typename Node, typename ...Args> 582 void test_copy_ctor_for_buffering_nodes(Args... node_args) { 583 oneapi::tbb::flow::graph g; 584 585 dummy_functor<int> fun; 586 587 Node testing_node(g, node_args...); 588 oneapi::tbb::flow::continue_node<int> pred_node(g, fun); 589 test_push_receiver<int> suc_node1(g); 590 test_push_receiver<int> suc_node2(g); 591 592 oneapi::tbb::flow::make_edge(pred_node, testing_node); 593 oneapi::tbb::flow::make_edge(testing_node, suc_node1); 594 595 #ifdef CONFORMANCE_OVERWRITE_NODE 596 testing_node.try_put(1); 597 #endif 598 599 Node node_copy(testing_node); 600 601 #ifdef CONFORMANCE_OVERWRITE_NODE 602 int tmp; 603 CHECK_MESSAGE((!node_copy.is_valid() && !node_copy.try_get(tmp)), "The buffered value is not copied from src"); 604 get_values(suc_node1); 605 #endif 606 607 oneapi::tbb::flow::make_edge(node_copy, suc_node2); 608 609 node_copy.try_put(0); 610 g.wait_for_all(); 611 612 CHECK_MESSAGE((get_values(suc_node1).size() == 0 && get_values(suc_node2).size() == 1), "Copied node doesn`t copy successor"); 613 614 #ifdef CONFORMANCE_OVERWRITE_NODE 615 node_copy.clear(); 616 testing_node.clear(); 617 #endif 618 619 pred_node.try_put(oneapi::tbb::flow::continue_msg()); 620 g.wait_for_all(); 621 622 CHECK_MESSAGE((get_values(suc_node1).size() == 1 && get_values(suc_node2).size() == 0), "Copied node doesn`t copy predecessor"); 623 } 624 625 template<typename Node, typename InputType, typename ...Args> 626 void test_priority(Args... node_args) { 627 std::size_t concurrency_limit = 1; 628 oneapi::tbb::global_control control(oneapi::tbb::global_control::max_allowed_parallelism, concurrency_limit); 629 630 oneapi::tbb::flow::graph g; 631 632 oneapi::tbb::flow::continue_node<InputType> source(g, dummy_functor<InputType>()); 633 634 track_first_id_functor<int>::first_id = -1; 635 track_first_id_functor<int> low_functor(1); 636 track_first_id_functor<int> high_functor(2); 637 638 Node high(g, node_args..., high_functor, oneapi::tbb::flow::node_priority_t(1)); 639 Node low(g, node_args..., low_functor); 640 641 make_edge(source, low); 642 make_edge(source, high); 643 644 source.try_put(oneapi::tbb::flow::continue_msg()); 645 646 g.wait_for_all(); 647 648 CHECK_MESSAGE((track_first_id_functor<int>::first_id == 2), "High priority node should execute first"); 649 } 650 651 template<typename Node> 652 void test_concurrency() { 653 auto max_num_threads = oneapi::tbb::this_task_arena::max_concurrency(); 654 655 oneapi::tbb::global_control c(oneapi::tbb::global_control::max_allowed_parallelism, 656 max_num_threads); 657 658 std::vector<int> threads_count = {1, oneapi::tbb::flow::serial, max_num_threads, oneapi::tbb::flow::unlimited}; 659 660 if(max_num_threads > 2) { 661 threads_count.push_back(max_num_threads / 2); 662 } 663 664 for(auto num_threads : threads_count) { 665 utils::ConcurrencyTracker::Reset(); 666 int expected_threads = num_threads; 667 if(num_threads == oneapi::tbb::flow::unlimited) { 668 expected_threads = max_num_threads; 669 } 670 if(num_threads == oneapi::tbb::flow::serial) { 671 expected_threads = 1; 672 } 673 oneapi::tbb::flow::graph g; 674 concurrency_peak_checker_body counter(expected_threads); 675 Node fnode(g, num_threads, counter); 676 677 test_push_receiver<int> suc_node(g); 678 679 make_edge(fnode, suc_node); 680 681 for(int i = 0; i < 500; ++i) { 682 fnode.try_put(i); 683 } 684 g.wait_for_all(); 685 } 686 } 687 688 template<typename Node> 689 void test_rejecting() { 690 oneapi::tbb::flow::graph g; 691 692 wait_flag_body body; 693 Node fnode(g, oneapi::tbb::flow::serial, body); 694 695 test_push_receiver<int> suc_node(g); 696 697 make_edge(fnode, suc_node); 698 699 fnode.try_put(0); 700 701 CHECK_MESSAGE((!fnode.try_put(1)), "Messages should be rejected while the first is being processed"); 702 703 wait_flag_body::flag = true; 704 705 g.wait_for_all(); 706 CHECK_MESSAGE((get_values(suc_node).size() == 1), "Messages should be rejected while the first is being processed"); 707 } 708 709 template<typename Node, typename CountingBody> 710 void test_output_input_class() { 711 using namespace oneapi::tbb::flow; 712 713 passthru_body<CountingBody> fun; 714 715 graph g; 716 Node node1(g, unlimited, fun); 717 test_push_receiver<CountingBody> suc_node(g); 718 make_edge(node1, suc_node); 719 CountingBody b1; 720 CountingBody b2; 721 node1.try_put(b1); 722 g.wait_for_all(); 723 suc_node.try_get(b2); 724 DOCTEST_WARN_MESSAGE((b1.copies_count > 0), "The type Input must meet the DefaultConstructible and CopyConstructible requirements"); 725 DOCTEST_WARN_MESSAGE((b2.is_copy), "The type Output must meet the CopyConstructible requirements"); 726 } 727 728 template<typename Node, typename Output = copy_counting_object<int>> 729 void test_output_class() { 730 using namespace oneapi::tbb::flow; 731 732 passthru_body<Output> fun; 733 734 graph g; 735 Node node1(g, fun); 736 test_push_receiver<Output> suc_node(g); 737 make_edge(node1, suc_node); 738 739 #ifdef CONFORMANCE_INPUT_NODE 740 node1.activate(); 741 #else 742 node1.try_put(oneapi::tbb::flow::continue_msg()); 743 #endif 744 745 g.wait_for_all(); 746 Output b; 747 suc_node.try_get(b); 748 DOCTEST_WARN_MESSAGE((b.is_copy), "The type Output must meet the CopyConstructible requirements"); 749 } 750 751 template<typename Node> 752 void test_with_reserving_join_node_class() { 753 using namespace oneapi::tbb::flow; 754 755 graph g; 756 757 function_node<int, int> static_result_computer_n( 758 g, serial, 759 [&](const int& msg) { 760 // compute the result using incoming message and pass it further, e.g.: 761 int result = int((msg >> 2) / 4); 762 return result; 763 }); 764 Node testing_node(g); // for buffering once computed value 765 766 buffer_node<int> buffer_n(g); 767 join_node<std::tuple<int, int>, reserving> join_n(g); 768 769 std::atomic<int> number{2}; 770 std::atomic<int> counter{0}; 771 function_node<std::tuple<int, int>> consumer_n( 772 g, unlimited, 773 [&](const std::tuple<int, int>& arg) { 774 // use the precomputed static result along with dynamic data 775 ++counter; 776 #ifdef CONFORMANCE_OVERWRITE_NODE 777 CHECK_MESSAGE((std::get<0>(arg) == int((number >> 2) / 4)), "A overwrite_node store a single item that can be overwritten"); 778 #else 779 CHECK_MESSAGE((std::get<0>(arg) == int((number >> 2) / 4)), "A write_once_node store a single item that cannot be overwritten"); 780 #endif 781 }); 782 783 make_edge(static_result_computer_n, testing_node); 784 make_edge(testing_node, input_port<0>(join_n)); 785 make_edge(buffer_n, input_port<1>(join_n)); 786 make_edge(join_n, consumer_n); 787 788 // do one-time calculation that will be reused many times further in the graph 789 static_result_computer_n.try_put(number); 790 791 constexpr int put_count = 50; 792 for (int i = 0; i < put_count / 2; i++) { 793 buffer_n.try_put(i); 794 } 795 #ifdef CONFORMANCE_OVERWRITE_NODE 796 number = 3; 797 #endif 798 static_result_computer_n.try_put(number); 799 for (int i = 0; i < put_count / 2; i++) { 800 buffer_n.try_put(i); 801 } 802 803 g.wait_for_all(); 804 CHECK_MESSAGE((counter == put_count), "join_node with reserving policy \ 805 if at least one successor accepts the tuple must consume messages"); 806 } 807 } 808 #endif // __TBB_test_conformance_conformance_flowgraph_H 809