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
16 completion of a task, e.g. for a large quantity of data that are about to be
17 written from a buffer into a socket. The tevent library has its own queue
18 structure that is ready to use after it has been initialized and started up
21 @subsection cr_queue Creation of Queues
23 The first and most important step is the creation of the tevent queue
24 (represented by struct tevent_queue), which will then be in running mode.
27 struct tevent_queue* tevent_queue_create (TALLOC_CTX *mem_ctx, const char *name)
30 When the program returns from this function, the allocated memory, set
31 destructor and labeled queue as running has been done and the structure is
32 ready to be filled with entries. Stopping and starting queues on the run. If
33 you need to stop a queue from processing its entries, and then turn it on
34 again, a couple of functions which serve this purpose are:
36 - bool tevent_queue_stop()
37 - bool tevent_queue_start()
39 These functions actually only provide for the simple setting of a variable,
40 which indicates that the queue has been stopped/started. Returned value
43 @subsection add_queue Adding Requests to a Queue
45 Tevent in fact offers 3 possible ways of inserting a request into a queue.
46 There are no vast differences between them, but still there might be situations
47 where one of them is more suitable and desired than another.
50 bool tevent_queue_add(struct tevent_queue *queue,
51 struct tevent_context *ev,
52 struct tevent_req *req,
53 tevent_queue_trigger_fn_t trigger,
57 This call is the simplest of all three. It offers only boolean verification of
58 whether the operation of adding the request into a queue was successful or not.
59 No additional deletion of an item from the queue is possible, i.e. it is only
60 possible to deallocate the whole tevent request, which would cause triggering
61 of destructor handling and also dropping the request from the queue.
63 <strong>Extended Options</strong>
65 Both of the following functions have a feature in common - they return tevent
66 queue entry structure representing the item in a queue. There is no further
67 possible handling with this structure except the use of the structure’s pointer
68 for its deallocation (which leads also its removal from the queue). The
69 difference lies in the possibility that with the following functions it is
70 possible to remove the tevent request from a queue without its deallocation.
71 The previous function can only deallocate the tevent request as it was from
72 memory, and thereby logically cause its removal from the queue as well. There
73 is no other utilization of this structure via API at this stage of tevent
74 library. The possibility of easier debugging while developing with tevent could
75 be considered to be an advantage of this returned pointer.
78 struct tevent_queue_entry *tevent_queue_add_entry(struct tevent_queue *queue,
79 struct tevent_context *ev,
80 struct tevent_req *req,
81 tevent_queue_trigger_fn_t trigger,
85 The feature that allows for the optimized addition of entries to a queue is
86 that a check for an empty queue with no items is first of all carried out. If
87 it is found that the queue is empty, then the request for inserting the entry
88 into a queue will be omitted and directly triggered.
91 struct tevent_queue_entry *tevent_queue_add_optimize_empty(struct tevent_queue *queue,
92 struct tevent_context *ev,
93 struct tevent_req *req,
94 tevent_queue_trigger_fn_t trigger,
98 When calling any of the functions serving for inserting an item into a queue,
99 it is possible to leave out the fourth argument (trigger) and instead of a
100 function pass a NULL pointer. This usage sets so-called blocking entries.
101 These entries, since they do not have any trigger operation to be activated,
102 just sit in their position until they are labeled as a done by another
103 function. Their purpose is to block other items in the queue from being
106 @subsection example_q Example of tevent queue
120 struct tevent_context *ev;
126 static void timer_handler(struct tevent_context *ev, struct tevent_timer *te,
127 struct timeval current_time, void *private_data)
129 // time event which after all sets request as done. Following item from
130 // the queue may be invoked.
131 struct tevent_req *req = private_data;
132 struct foo_state *stateX = tevent_req_data(req, struct foo_state);
134 // processing some stuff
136 printf("time_handler\n");
138 tevent_req_done(req);
141 printf("Request #%d set as done.\n", stateX->x);
144 static void trigger(struct tevent_req *req, void *private_data)
146 struct juststruct *priv = tevent_req_callback_data (req, struct juststruct);
147 struct foo_state *in = tevent_req_data(req, struct foo_state);
148 struct timeval schedule;
149 struct tevent_timer *tim;
150 schedule = tevent_timeval_current_ofs(1, 0);
151 printf("Processing request #%d\n", in->x);
153 if (in->x % 3 == 0) { // just example; third request does not contain
154 // any further operation and will be finished right
158 tim = tevent_add_timer(priv->ev, req, schedule, timer_handler, req);
162 tevent_req_done(req);
164 printf("Request #%d set as done.\n", in->x);
168 struct tevent_req *foo_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
169 const char *name, int num)
171 struct tevent_req *req;
172 struct foo_state *state;
173 struct foo_state *in;
174 struct tevent_timer *tim;
176 printf("foo_send\n");
177 req = tevent_req_create(mem_ctx, &state, struct foo_state);
178 if (req == NULL) { // check for appropriate allocation
179 tevent_req_error(req, 1);
183 // exemplary filling of variables
184 state->local_var = 1;
190 static void foo_done(struct tevent_req *req) {
192 enum tevent_req_state state;
195 if (tevent_req_is_error(req, &state, &err)) {
196 printf("ERROR WAS SET %d\n", state);
199 // processing some stuff
200 printf("Callback is done...\n");
204 int main (int argc, char **argv)
207 struct tevent_req* req[6];
208 struct tevent_req* tmp;
209 struct tevent_context *ev;
210 struct tevent_queue *fronta = NULL;
211 struct juststruct *data;
215 const char * const names[] = {
216 "first", "second", "third", "fourth", "fifth"
221 mem_ctx = talloc_new(NULL); //parent
222 talloc_parent(mem_ctx);
223 ev = tevent_context_init(mem_ctx);
225 fprintf(stderr, "MEMORY ERROR\n");
230 fronta = tevent_queue_create(mem_ctx, "test_queue");
231 tevent_queue_stop(fronta);
232 tevent_queue_start(fronta);
233 if (tevent_queue_running(fronta)) {
234 printf ("Queue is running (length: %d)\n", tevent_queue_length(fronta));
236 printf ("Queue is not running\n");
239 data = talloc(ev, struct juststruct);
245 for (i = 1; i < 5; i++) {
246 req[i] = foo_send(mem_ctx, ev, names[i], i);
248 if (req[i] == NULL) {
249 fprintf(stderr, "Request error! %d \n", ret);
252 tevent_req_set_callback(req[i], foo_done, data);
256 // add item to a queue
257 tevent_queue_add(fronta, ev, req[1], trigger, data);
258 tevent_queue_add(fronta, ev, req[2], trigger, data);
259 tevent_queue_add(fronta, ev, req[3], trigger, data);
260 tevent_queue_add(fronta, ev, req[4], trigger, data);
262 printf("Queue length: %d\n", tevent_queue_length(fronta));
263 while(tevent_queue_length(fronta) > 0) {
264 tevent_loop_once(ev);
265 printf("Queue: %d items left\n", tevent_queue_length(fronta));
268 talloc_free(mem_ctx);