2 @page tevent_queue Chapter 5: Tevent queue
3 @section queue Tevent queue
5 There is a possibility that the dispatcher and its handlers may not be able to
6 handle all the incoming events as quickly as they arrive. One way to deal with
7 this situation is to buffer the received events by introducing an event queue
8 into the events stream, between the events generator and the dispatcher. Events
9 are added to the queue as they arrive, and the dispatcher pops them off the
10 beginning of the queue as fast as possible. In tevent library it is
11 similar, but the queue is not automatically set for any event. The queue has to
12 be created on purpose, and events which should follow the order of the FIFO
13 queue have to be explicitly pinpointed. Creating such a queue is crucial in
14 situations when sequential processing is absolutely essential for the succesful
15 completion of a task, e.g. for a large quantity of data that are about to be
16 written from a buffer into a socket. The tevent library has its own queue
17 structure that is ready to use after it has been initialized and started up
20 @subsection cr_queue Creation of Queues
22 The first and most important step is the creation of the tevent queue
23 (represented by struct tevent queue), which will then be in running mode.
26 struct tevent_queue* tevent_queue_create (TALLOC_CTX *mem_ctx, const char *name)
29 When the program returns from this function, the allocated memory, set
30 destructor and labeled queue as running has been done and the structure is
31 ready to be filled with entries. Stopping and starting queues on the run. If
32 you need to stop a queue from processing its entries, and then turn it on
33 again, a couple of functions which serve this purpose are:
35 - bool tevent_queue_stop()
36 - bool tevent_queue_start()
38 These functions actually only provide for the simple setting of a variable,
39 which indicates that the queue has been stopped/started. Returned value
42 @subsection add_queue Adding Requests to a Queue
44 Tevent in fact offers 3 possible ways of inserting a request into a queue.
45 There are no vast differences between them, but still there might be situations
46 where one of them is more suitable and desired than another.
49 bool tevent_queue_add(struct tevent_queue *queue,
50 struct tevent_context *ev,
51 struct tevent_req *req,
52 tevent_queue_trigger_fn_t trigger,
56 This call is the simplest of all three. It offers only boolean verification of
57 whether the operation of adding the request into a queue was successful or not.
58 No additional deletion of an item from the queue is possible, i.e. it is only
59 possible to deallocate the whole tevent request, which would cause triggering
60 of destructor handling and also dropping the request from the queue.
62 <strong>Extended Options</strong>
64 Both of the following functions have a feature in common - they return tevent
65 queue entry structure representing the item in a queue. There is no further
66 possible handling with this structure except the use of the structure’s pointer
67 for its deallocation (which leads also its removal from the queue). The
68 difference lies in the possibility that with the following functions it is
69 possible to remove the tevent request from a queue without its deallocation.
70 The previous function can only deallocate the tevent request as it was from
71 memory, and thereby logically cause its removal from the queue as well. There
72 is no other utilization of this structure via API at this stage of tevent
73 library. The possibility of easier debugging while developing with tevent could
74 be considered to be an advantage of this returned pointer.
77 struct tevent_queue_entry *tevent_queue_add_entry(struct tevent_queue *queue,
78 struct tevent_context *ev,
79 struct tevent_req *req,
80 tevent_queue_trigger_fn_t trigger,
84 The feature that allows for the optimized addition of entries to a queue is
85 that a check for an empty queue with no items is first of all carried out. If
86 it is found that the queue is empty, then the request for inserting the entry
87 into a queue will be omitted and directly triggered.
90 struct tevent_queue_entry *tevent_queue_add_optimize_empty(struct tevent_queue *queue,
91 struct tevent_context *ev,
92 struct tevent_req *req,
93 tevent_queue_trigger_fn_t trigger,
97 When calling any of the functions serving for inserting an item into a queue,
98 it is possible to leave out the fourth argument (trigger) and instead of a
99 function pass a NULL pointer. This usage sets so-called blocking entries.
100 These entries, since they do not have any trigger operation to be activated,
101 just sit in their position until they are labeled as a done by another
102 function. Their purpose is to block other items in the queue from being
105 @subsection example_q Example of tevent queue
119 struct tevent_context *ev;
125 static void timer_handler(struct tevent_context *ev, struct tevent_timer *te,
126 struct timeval current_time, void *private_data)
128 // time event which after all sets request as done. Following item from
129 // the queue may be invoked.
130 struct tevent_req *req = private_data;
131 struct foo_state *stateX = tevent_req_data(req, struct foo_state);
133 // processing some stuff
135 printf("time_handler\n");
137 tevent_req_done(req);
140 printf("Request #%d set as done.\n", stateX->x);
143 static void trigger(struct tevent_req *req, void *private_data)
145 struct juststruct *priv = tevent_req_callback_data (req, struct juststruct);
146 struct foo_state *in = tevent_req_data(req, struct foo_state);
147 struct timeval schedule;
148 struct tevent_timer *tim;
149 schedule = tevent_timeval_current_ofs(1, 0);
150 printf("Processing request #%d\n", in->x);
152 if (in->x % 3 == 0) { // just example; third request does not contain
153 // any further operation and will be finished right
157 tim = tevent_add_timer(priv->ev, req, schedule, timer_handler, req);
161 tevent_req_done(req);
163 printf("Request #%d set as done.\n", in->x);
167 struct tevent_req *foo_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
168 const char *name, int num)
170 struct tevent_req *req;
171 struct foo_state *state;
172 struct foo_state *in;
173 struct tevent_timer *tim;
175 printf("foo_send\n");
176 req = tevent_req_create(mem_ctx, &state, struct foo_state);
177 if (req == NULL) { // check for appropriate allocation
178 tevent_req_error(req, 1);
182 // exemplary filling of variables
183 state->local_var = 1;
189 static void foo_done(struct tevent_req *req) {
191 enum tevent_req_state state;
194 if (tevent_req_is_error(req, &state, &err)) {
195 printf("ERROR WAS SET %d\n", state);
198 // processing some stuff
199 printf("Callback is done...\n");
203 int main (int argc, char **argv)
206 struct tevent_req* req[6];
207 struct tevent_req* tmp;
208 struct tevent_context *ev;
209 struct tevent_queue *fronta = NULL;
210 struct juststruct *data;
214 const char * const names[] = {
215 "first", "second", "third", "fourth", "fifth"
220 mem_ctx = talloc_new(NULL); //parent
221 talloc_parent(mem_ctx);
222 ev = tevent_context_init(mem_ctx);
224 fprintf(stderr, "MEMORY ERROR\n");
229 fronta = tevent_queue_create(mem_ctx, "test_queue");
230 tevent_queue_stop(fronta);
231 tevent_queue_start(fronta);
232 if (tevent_queue_running(fronta)) {
233 printf ("Queue is runnning (length: %d)\n", tevent_queue_length(fronta));
235 printf ("Queue is not runnning\n");
238 data = talloc(ev, struct juststruct);
244 for (i = 1; i < 5; i++) {
245 req[i] = foo_send(mem_ctx, ev, names[i], i);
247 if (req[i] == NULL) {
248 fprintf(stderr, "Request error! %d \n", ret);
251 tevent_req_set_callback(req[i], foo_done, data);
255 // add item to a queue
256 tevent_queue_add(fronta, ev, req[1], trigger, data);
257 tevent_queue_add(fronta, ev, req[2], trigger, data);
258 tevent_queue_add(fronta, ev, req[3], trigger, data);
259 tevent_queue_add(fronta, ev, req[4], trigger, data);
261 printf("Queue length: %d\n", tevent_queue_length(fronta));
262 while(tevent_queue_length(fronta) > 0) {
263 tevent_loop_once(ev);
264 printf("Queue: %d items left\n", tevent_queue_length(fronta));
267 talloc_free(mem_ctx);