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