talloc: use the system pytalloc-util for python3 as well
[sfrench/samba-autobuild/.git] / lib / tevent / tevent_queue.c
1 /*
2    Unix SMB/CIFS implementation.
3    Infrastructure for async requests
4    Copyright (C) Volker Lendecke 2008
5    Copyright (C) Stefan Metzmacher 2009
6
7      ** NOTE! The following LGPL license applies to the tevent
8      ** library. This does NOT imply that all of Samba is released
9      ** under the LGPL
10
11    This library is free software; you can redistribute it and/or
12    modify it under the terms of the GNU Lesser General Public
13    License as published by the Free Software Foundation; either
14    version 3 of the License, or (at your option) any later version.
15
16    This library is distributed in the hope that it will be useful,
17    but WITHOUT ANY WARRANTY; without even the implied warranty of
18    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
19    Lesser General Public License for more details.
20
21    You should have received a copy of the GNU Lesser General Public
22    License along with this library; if not, see <http://www.gnu.org/licenses/>.
23 */
24
25 #include "replace.h"
26 #include "tevent.h"
27 #include "tevent_internal.h"
28 #include "tevent_util.h"
29
30 struct tevent_queue_entry {
31         struct tevent_queue_entry *prev, *next;
32         struct tevent_queue *queue;
33
34         bool triggered;
35
36         struct tevent_req *req;
37         struct tevent_context *ev;
38
39         tevent_queue_trigger_fn_t trigger;
40         void *private_data;
41 };
42
43 struct tevent_queue {
44         const char *name;
45         const char *location;
46
47         bool running;
48         struct tevent_immediate *immediate;
49
50         size_t length;
51         struct tevent_queue_entry *list;
52 };
53
54 static void tevent_queue_immediate_trigger(struct tevent_context *ev,
55                                            struct tevent_immediate *im,
56                                            void *private_data);
57
58 static int tevent_queue_entry_destructor(struct tevent_queue_entry *e)
59 {
60         struct tevent_queue *q = e->queue;
61
62         if (!q) {
63                 return 0;
64         }
65
66         DLIST_REMOVE(q->list, e);
67         q->length--;
68
69         if (!q->running) {
70                 return 0;
71         }
72
73         if (!q->list) {
74                 return 0;
75         }
76
77         if (q->list->triggered) {
78                 return 0;
79         }
80
81         tevent_schedule_immediate(q->immediate,
82                                   q->list->ev,
83                                   tevent_queue_immediate_trigger,
84                                   q);
85
86         return 0;
87 }
88
89 static int tevent_queue_destructor(struct tevent_queue *q)
90 {
91         q->running = false;
92
93         while (q->list) {
94                 struct tevent_queue_entry *e = q->list;
95                 talloc_free(e);
96         }
97
98         return 0;
99 }
100
101 struct tevent_queue *_tevent_queue_create(TALLOC_CTX *mem_ctx,
102                                           const char *name,
103                                           const char *location)
104 {
105         struct tevent_queue *queue;
106
107         queue = talloc_zero(mem_ctx, struct tevent_queue);
108         if (!queue) {
109                 return NULL;
110         }
111
112         queue->name = talloc_strdup(queue, name);
113         if (!queue->name) {
114                 talloc_free(queue);
115                 return NULL;
116         }
117         queue->immediate = tevent_create_immediate(queue);
118         if (!queue->immediate) {
119                 talloc_free(queue);
120                 return NULL;
121         }
122
123         queue->location = location;
124
125         /* queue is running by default */
126         queue->running = true;
127
128         talloc_set_destructor(queue, tevent_queue_destructor);
129         return queue;
130 }
131
132 static void tevent_queue_immediate_trigger(struct tevent_context *ev,
133                                            struct tevent_immediate *im,
134                                            void *private_data)
135 {
136         struct tevent_queue *q =
137                 talloc_get_type_abort(private_data,
138                 struct tevent_queue);
139
140         if (!q->running) {
141                 return;
142         }
143
144         if (!q->list) {
145                 return;
146         }
147
148         q->list->triggered = true;
149         q->list->trigger(q->list->req, q->list->private_data);
150 }
151
152 static struct tevent_queue_entry *tevent_queue_add_internal(
153                                         struct tevent_queue *queue,
154                                         struct tevent_context *ev,
155                                         struct tevent_req *req,
156                                         tevent_queue_trigger_fn_t trigger,
157                                         void *private_data,
158                                         bool allow_direct)
159 {
160         struct tevent_queue_entry *e;
161
162         e = talloc_zero(req, struct tevent_queue_entry);
163         if (e == NULL) {
164                 return NULL;
165         }
166
167         e->queue = queue;
168         e->req = req;
169         e->ev = ev;
170         e->trigger = trigger;
171         e->private_data = private_data;
172
173         /*
174          * if there is no trigger, it is just a blocker
175          */
176         if (trigger == NULL) {
177                 e->triggered = true;
178         }
179
180         if (queue->length > 0) {
181                 /*
182                  * if there are already entries in the
183                  * queue do not optimize.
184                  */
185                 allow_direct = false;
186         }
187
188         if (req->async.fn != NULL) {
189                 /*
190                  * If the caller wants to optimize for the
191                  * empty queue case, call the trigger only
192                  * if there is no callback defined for the
193                  * request yet.
194                  */
195                 allow_direct = false;
196         }
197
198         DLIST_ADD_END(queue->list, e);
199         queue->length++;
200         talloc_set_destructor(e, tevent_queue_entry_destructor);
201
202         if (!queue->running) {
203                 return e;
204         }
205
206         if (queue->list->triggered) {
207                 return e;
208         }
209
210         /*
211          * If allowed we directly call the trigger
212          * avoiding possible delays caused by
213          * an immediate event.
214          */
215         if (allow_direct) {
216                 queue->list->triggered = true;
217                 queue->list->trigger(queue->list->req,
218                                      queue->list->private_data);
219                 return e;
220         }
221
222         tevent_schedule_immediate(queue->immediate,
223                                   queue->list->ev,
224                                   tevent_queue_immediate_trigger,
225                                   queue);
226
227         return e;
228 }
229
230 bool tevent_queue_add(struct tevent_queue *queue,
231                       struct tevent_context *ev,
232                       struct tevent_req *req,
233                       tevent_queue_trigger_fn_t trigger,
234                       void *private_data)
235 {
236         struct tevent_queue_entry *e;
237
238         e = tevent_queue_add_internal(queue, ev, req,
239                                       trigger, private_data, false);
240         if (e == NULL) {
241                 return false;
242         }
243
244         return true;
245 }
246
247 struct tevent_queue_entry *tevent_queue_add_entry(
248                                         struct tevent_queue *queue,
249                                         struct tevent_context *ev,
250                                         struct tevent_req *req,
251                                         tevent_queue_trigger_fn_t trigger,
252                                         void *private_data)
253 {
254         return tevent_queue_add_internal(queue, ev, req,
255                                          trigger, private_data, false);
256 }
257
258 struct tevent_queue_entry *tevent_queue_add_optimize_empty(
259                                         struct tevent_queue *queue,
260                                         struct tevent_context *ev,
261                                         struct tevent_req *req,
262                                         tevent_queue_trigger_fn_t trigger,
263                                         void *private_data)
264 {
265         return tevent_queue_add_internal(queue, ev, req,
266                                          trigger, private_data, true);
267 }
268
269 void tevent_queue_start(struct tevent_queue *queue)
270 {
271         if (queue->running) {
272                 /* already started */
273                 return;
274         }
275
276         queue->running = true;
277
278         if (!queue->list) {
279                 return;
280         }
281
282         if (queue->list->triggered) {
283                 return;
284         }
285
286         tevent_schedule_immediate(queue->immediate,
287                                   queue->list->ev,
288                                   tevent_queue_immediate_trigger,
289                                   queue);
290 }
291
292 void tevent_queue_stop(struct tevent_queue *queue)
293 {
294         queue->running = false;
295 }
296
297 size_t tevent_queue_length(struct tevent_queue *queue)
298 {
299         return queue->length;
300 }
301
302 bool tevent_queue_running(struct tevent_queue *queue)
303 {
304         return queue->running;
305 }
306
307 struct tevent_queue_wait_state {
308         uint8_t dummy;
309 };
310
311 static void tevent_queue_wait_trigger(struct tevent_req *req,
312                                       void *private_data);
313
314 struct tevent_req *tevent_queue_wait_send(TALLOC_CTX *mem_ctx,
315                                           struct tevent_context *ev,
316                                           struct tevent_queue *queue)
317 {
318         struct tevent_req *req;
319         struct tevent_queue_wait_state *state;
320         bool ok;
321
322         req = tevent_req_create(mem_ctx, &state,
323                                 struct tevent_queue_wait_state);
324         if (req == NULL) {
325                 return NULL;
326         }
327
328         ok = tevent_queue_add(queue, ev, req,
329                               tevent_queue_wait_trigger,
330                               NULL);
331         if (!ok) {
332                 tevent_req_oom(req);
333                 return tevent_req_post(req, ev);
334         }
335
336         return req;
337 }
338
339 static void tevent_queue_wait_trigger(struct tevent_req *req,
340                                          void *private_data)
341 {
342         tevent_req_done(req);
343 }
344
345 bool tevent_queue_wait_recv(struct tevent_req *req)
346 {
347         enum tevent_req_state state;
348         uint64_t err;
349
350         if (tevent_req_is_error(req, &state, &err)) {
351                 tevent_req_received(req);
352                 return false;
353         }
354
355         tevent_req_received(req);
356         return true;
357 }