1.. _create_token_based_system:
2
3Create a Token-Based System
4===========================
5
6
7A more flexible solution to limit the number of messages in a flow graph
8is to use tokens. In a token-based system, a limited number of tokens
9are available in the graph and a message will not be allowed to enter
10the graph until it can be paired with an available token. When a message
11is retired from the graph, its token is released, and can be paired with
12a new message that will then be allowed to enter.
13
14
15The ``oneapi::tbb::parallel_pipeline`` algorithm relies on a token-based system. In
16the flow graph interface, there is no explicit support for tokens, but
17``join_node``s can be used to create an analogous system. A ``join_node`` has
18two template arguments, the tuple that describes the types of its inputs
19and a buffer policy:
20
21
22::
23
24
25   template<typename OutputTuple, graph_buffer_policy JP = queueing>
26   class join_node;
27
28
29The buffer policy can be one of the following:
30
31
32-  ``queueing``. This type of policy causes inputs to be matched
33   first-in-first-out; that is, the inputs are joined together to form a
34   tuple in the order they are received.
35-  ``tag_matching``. This type of policy joins inputs together that have
36   matching tags.
37-  ``reserving``. This type of policy causes the ``join_node`` to do no
38   internally buffering, but instead to consume inputs only when it can
39   first reserve an input on each port from an upstream source. If it
40   can reserve an input at each port, it gets those inputs and joins
41   those together to form an output tuple.
42
43
44A token-based system can be created by using reserving join_nodes.
45
46
47In the example below, there is an ``input_node`` that generates ``M`` big
48objects and a ``buffer_node`` that is pre-filled with three tokens. The
49``token_t`` can be anything, for example it could be ``typedef int token_t;``.
50The ``input_node`` and ``buffer_node`` are connected to a reserving ``join_node``.
51The ``input_node`` will only generate an input when one is pulled from it
52by the reserving ``join_node``, and the reserving ``join_node`` will only pull
53the input from the ``input_node`` when it knows there is also an item to
54pull from the ``buffer_node``.
55
56
57::
58
59
60     graph g;
61
62
63     int src_count = 0;
64     int number_of_objects = 0;
65     int max_objects = 3;
66
67
68     input_node< big_object * > s( g, [&]( oneapi::tbb::flow_control& fc ) -> big_object* {
69         if ( src_count < M ) {
70           big_object* v = new big_object();
71           ++src_count;
72           return v;
73         } else {
74           fc.stop();
75           return nullptr;
76         }
77     } );
78     s.activate();
79
80     join_node< tuple_t, reserving > j(g);
81
82
83     buffer_node< token_t > b(g);
84
85
86     function_node< tuple_t, token_t > f( g, unlimited,
87       []( const tuple_t &t ) -> token_t {
88           spin_for(1);
89        cout << get<1>(t) << "\n";
90           delete get<0>(t);
91        return get<1>(t);
92     } );
93
94
95     make_edge( s, input_port<0>(j) );
96     make_edge( b, input_port<1>(j) );
97     make_edge( j, f );
98     make_edge( f, b );
99
100
101     b.try_put( 1 );
102     b.try_put( 2 );
103     b.try_put( 3 );
104
105
106     g.wait_for_all();
107
108
109In the above code, you can see that the ``function_node`` returns the token
110back to the ``buffer_node``. This cycle in the flow graph allows the token
111to be recycled and paired with another input from the ``input_node``. So
112like in the previous sections, there will be at most four big objects in
113the graph. There could be three big objects in the ``function_node`` and one
114buffered in the ``input_node``, awaiting a token to be paired with.
115
116
117Since there is no specific ``token_t`` defined for the flow graph, you can
118use any type for a token, including objects or pointers to arrays.
119Therefore, unlike in the example above, the ``token_t`` doesn't need to be a
120dummy type; it could for example be a buffer or other object that is
121essential to the computation. We could, for example, modify the example
122above to use the big objects themselves as the tokens, removing the need
123to repeatedly allocate and deallocate them, and essentially create a
124free list of big objects using a cycle back to the ``buffer_node``.
125
126
127Also, in our example above, the ``buffer_node`` was prefilled by a fixed
128number of explicit calls to ``try_put``, but there are other options. For
129example, an ``input_node`` could be attached to the input of the
130``buffer_node``, and it could generate the tokens. In addition, our
131``function_node`` could be replaced by a ``multifunction_node`` that can
132optionally put 0 or more outputs to each of its output ports. Using a
133``multifunction_node``, you can choose to recycle or not recycle a token, or
134even generate more tokens, thereby increasing or decreasing the allowed
135concurrency in the graph.
136
137
138A token based system is therefore very flexible. You are free to declare
139the token to be of any type and to inject or remove tokens from the
140system as it is executing, thereby having dynamic control of the allowed
141concurrency in the system. Since you can pair the token with an input at
142the source, this approach enables you to limit resource consumption
143across the entire graph.
144
145