Intel® oneAPI Threading Building Blocks Developer Guide and API Reference
A newer version of this document is available. Customers should click here to go to the newest version.
Visible to Intel only — GUID: GUID-0C40D33C-235A-417D-B4DE-0B4F12293802
Visible to Intel only — GUID: GUID-0C40D33C-235A-417D-B4DE-0B4F12293802
Create a Token-Based System
A more flexible solution to limit the number of messages in a flow graph is to use tokens. In a token-based system, a limited number of tokens are available in the graph and a message will not be allowed to enter the graph until it can be paired with an available token. When a message is retired from the graph, its token is released, and can be paired with a new message that will then be allowed to enter.
The oneapi::tbb::parallel_pipeline algorithm relies on a token-based system. In the flow graph interface, there is no explicit support for tokens, but join_node``s can be used to create an analogous system. A ``join_node has two template arguments, the tuple that describes the types of its inputs and a buffer policy:
template<typename OutputTuple, graph_buffer_policy JP = queueing> class join_node;
The buffer policy can be one of the following:
queueing. This type of policy causes inputs to be matched first-in-first-out; that is, the inputs are joined together to form a tuple in the order they are received.
tag_matching. This type of policy joins inputs together that have matching tags.
reserving. This type of policy causes the join_node to do no internally buffering, but instead to consume inputs only when it can first reserve an input on each port from an upstream source. If it can reserve an input at each port, it gets those inputs and joins those together to form an output tuple.
A token-based system can be created by using reserving join_nodes.
In the example below, there is an input_node that generates M big objects and a buffer_node that is pre-filled with three tokens. The token_t can be anything, for example it could be typedef int token_t;. The input_node and buffer_node are connected to a reserving join_node. The input_node will only generate an input when one is pulled from it by the reserving join_node, and the reserving join_node will only pull the input from the input_node when it knows there is also an item to pull from the buffer_node.
graph g;
int src_count = 0;
int number_of_objects = 0;
int max_objects = 3;
input_node< big_object * > s( g, [&]( oneapi::tbb::flow_control& fc ) -> big_object* {
if ( src_count < M ) {
big_object* v = new big_object();
++src_count;
return v;
} else {
fc.stop();
return nullptr;
}
} );
s.activate();
join_node< tuple_t, reserving > j(g);
buffer_node< token_t > b(g);
function_node< tuple_t, token_t > f( g, unlimited,
[]( const tuple_t &t ) -> token_t {
spin_for(1);
cout << get<1>(t) << "\n";
delete get<0>(t);
return get<1>(t);
} );
make_edge( s, input_port<0>(j) );
make_edge( b, input_port<1>(j) );
make_edge( j, f );
make_edge( f, b );
b.try_put( 1 );
b.try_put( 2 );
b.try_put( 3 );
g.wait_for_all();
In the above code, you can see that the function_node returns the token back to the buffer_node. This cycle in the flow graph allows the token to be recycled and paired with another input from the input_node. So like in the previous sections, there will be at most four big objects in the graph. There could be three big objects in the function_node and one buffered in the input_node, awaiting a token to be paired with.
Since there is no specific token_t defined for the flow graph, you can use any type for a token, including objects or pointers to arrays. Therefore, unlike in the example above, the token_t doesn’t need to be a dummy type; it could for example be a buffer or other object that is essential to the computation. We could, for example, modify the example above to use the big objects themselves as the tokens, removing the need to repeatedly allocate and deallocate them, and essentially create a free list of big objects using a cycle back to the buffer_node.
Also, in our example above, the buffer_node was prefilled by a fixed number of explicit calls to try_put, but there are other options. For example, an input_node could be attached to the input of the buffer_node, and it could generate the tokens. In addition, our function_node could be replaced by a multifunction_node that can optionally put 0 or more outputs to each of its output ports. Using a multifunction_node, you can choose to recycle or not recycle a token, or even generate more tokens, thereby increasing or decreasing the allowed concurrency in the graph.
A token based system is therefore very flexible. You are free to declare the token to be of any type and to inject or remove tokens from the system as it is executing, thereby having dynamic control of the allowed concurrency in the system. Since you can pair the token with an input at the source, this approach enables you to limit resource consumption across the entire graph.