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 #if __INTEL_COMPILER && _MSC_VER
18 #pragma warning(disable : 2586) // decorated name length exceeded, name was truncated
19 #endif
20 
21 #define SEQUENCER_NODE
22 
23 #include "conformance_flowgraph.h"
24 
25 //! \file conformance_sequencer_node.cpp
26 //! \brief Test for [flow_graph.sequencer_node] specification
27 
28 
29 #if __TBB_CPP17_DEDUCTION_GUIDES_PRESENT
30 template <typename Body>
31 void test_deduction_guides_common(Body body) {
32     using namespace tbb::flow;
33     graph g;
34     broadcast_node<int> br(g);
35 
36     sequencer_node s1(g, body);
37     static_assert(std::is_same_v<decltype(s1), sequencer_node<int>>);
38 
39 #if __TBB_PREVIEW_FLOW_GRAPH_NODE_SET
40     sequencer_node s2(follows(br), body);
41     static_assert(std::is_same_v<decltype(s2), sequencer_node<int>>);
42 #endif
43 
44     sequencer_node s3(s1);
45     static_assert(std::is_same_v<decltype(s3), sequencer_node<int>>);
46 }
47 
48 std::size_t sequencer_body_f(const int&) { return 1; }
49 
50 void test_deduction_guides() {
51     test_deduction_guides_common([](const int&)->std::size_t { return 1; });
52     test_deduction_guides_common([](const int&) mutable ->std::size_t { return 1; });
53     test_deduction_guides_common(sequencer_body_f);
54 }
55 #endif
56 
57 //! Test deduction guides
58 //! \brief \ref interface \ref requirement
59 TEST_CASE("Deduction guides"){
60 #if __TBB_CPP17_DEDUCTION_GUIDES_PRESENT
61     test_deduction_guides();
62 #endif
63 }
64 
65 //! Test sequencer_node single_push
66 //! \brief \ref requirement
67 TEST_CASE("sequencer_node single_push"){
68     conformance::sequencer_functor<int> sequencer;
69     conformance::test_forwarding_single_push<oneapi::tbb::flow::sequencer_node<int>>(sequencer);
70 }
71 
72 //! Test function_node buffering
73 //! \brief \ref requirement
74 TEST_CASE("sequencer_node buffering"){
75     conformance::sequencer_functor<int> sequencer;
76     conformance::test_buffering<oneapi::tbb::flow::sequencer_node<int>, int>(sequencer);
77 }
78 
79 //! Constructs an empty sequencer_node that belongs to the same graph g as src.
80 //! Any intermediate state of src, including its links to predecessors and successors, is not copied.
81 //! \brief \ref requirement
82 TEST_CASE("sequencer_node copy constructor"){
83     conformance::sequencer_functor<int> sequencer;
84     conformance::test_copy_ctor_for_buffering_nodes<oneapi::tbb::flow::sequencer_node<int>>(sequencer);
85 }
86 
87 //! Test inheritance relations
88 //! \brief \ref interface
89 TEST_CASE("sequencer_node superclasses"){
90     conformance::test_inheritance<oneapi::tbb::flow::sequencer_node<int>, int, int>();
91     conformance::test_inheritance<oneapi::tbb::flow::sequencer_node<void*>, void*, void*>();
92 }
93 
94 //! Test the sequencer_node rejects duplicate sequencer numbers
95 //! \brief \ref interface
96 TEST_CASE("sequencer_node rejects duplicate"){
97     oneapi::tbb::flow::graph g;
98     conformance::sequencer_functor<int> sequencer;
99 
100     oneapi::tbb::flow::sequencer_node<int> node(g, sequencer);
101 
102     node.try_put(1);
103 
104     CHECK_MESSAGE((node.try_put(1) == false), "sequencer_node must rejects duplicate sequencer numbers");
105     g.wait_for_all();
106 }
107 
108 //! Test queue_node node `try_put()` and `try_get()`
109 //! \brief \ref requirement
110 TEST_CASE("queue_node methods"){
111     oneapi::tbb::flow::graph g;
112     conformance::sequencer_functor<int> sequencer;
113 
114     oneapi::tbb::flow::sequencer_node<int> node(g, sequencer);
115 
116     node.try_put(1);
117     node.try_put(0);
118     node.try_put(1);
119     g.wait_for_all();
120 
121     int tmp = -1;
122     CHECK_MESSAGE((node.try_get(tmp) == true), "Getting from sequencer should succeed");
123     CHECK_MESSAGE((tmp == 0), "Received value should be correct");
124 
125     tmp = -1;
126     CHECK_MESSAGE((node.try_get(tmp) == true), "Getting from sequencer should succeed");
127     CHECK_MESSAGE((tmp == 1), "Received value should be correct");
128 
129     tmp = -1;
130     CHECK_MESSAGE((node.try_get(tmp) == false), "Getting from sequencer should not succeed");
131 }
132 
133 //! The example demonstrates ordering capabilities of the sequencer_node.
134 //! While being processed in parallel, the data is passed to the successor node in the exact same order it was read.
135 //! \brief \ref requirement
136 TEST_CASE("sequencer_node ordering"){
137     using namespace oneapi::tbb::flow;
138     using message = conformance::sequencer_functor<int>::seq_message;
139     graph g;
140 
141     // Due to parallelism the node can push messages to its successors in any order
142     function_node<message, message> process(g, unlimited, [] (message msg) {
143         msg.data++;
144         return msg;
145     });
146 
147     sequencer_node<message> ordering(g, conformance::sequencer_functor<int>());
148 
149     std::atomic<std::size_t> counter{0};
150     function_node<message> writer(g, tbb::flow::serial, [&] (const message& msg) {
151         CHECK_MESSAGE((msg.id == counter++), "The data is passed to the successor node in the exact same order it was read");
152     });
153 
154     tbb::flow::make_edge(process, ordering);
155     tbb::flow::make_edge(ordering, writer);
156 
157     for (std::size_t i = 0; i < 100; ++i) {
158         message msg = {i, 0};
159         process.try_put(msg);
160     }
161 
162     g.wait_for_all();
163 }
164