0e890cb7ce52a2568710688707547bf6fa96916e
[samba.git] / lib / pthreadpool / pthreadpool_tevent.c
1 /*
2  * Unix SMB/CIFS implementation.
3  * threadpool implementation based on pthreads
4  * Copyright (C) Volker Lendecke 2009,2011
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License as published by
8  * the Free Software Foundation; either version 3 of the License, or
9  * (at your option) any later version.
10  *
11  * This program is distributed in the hope that it will be useful,
12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14  * GNU General Public License for more details.
15  *
16  * You should have received a copy of the GNU General Public License
17  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
18  */
19
20 #include "replace.h"
21 #include "pthreadpool_tevent.h"
22 #include "pthreadpool.h"
23 #include "lib/util/tevent_unix.h"
24 #include "lib/util/dlinklist.h"
25
26 struct pthreadpool_tevent_job_state;
27
28 /*
29  * We need one pthreadpool_tevent_glue object per unique combintaion of tevent
30  * contexts and pthreadpool_tevent objects. Maintain a list of used tevent
31  * contexts in a pthreadpool_tevent.
32  */
33 struct pthreadpool_tevent_glue {
34         struct pthreadpool_tevent_glue *prev, *next;
35         struct pthreadpool_tevent *pool; /* back-pointer to owning object. */
36         /* Tuple we are keeping track of in this list. */
37         struct tevent_context *ev;
38         struct tevent_threaded_context *tctx;
39         /* Pointer to link object owned by *ev. */
40         struct pthreadpool_tevent_glue_ev_link *ev_link;
41 };
42
43 /*
44  * The pthreadpool_tevent_glue_ev_link and its destructor ensure we remove the
45  * tevent context from our list of active event contexts if the event context
46  * is destroyed.
47  * This structure is talloc()'ed from the struct tevent_context *, and is a
48  * back-pointer allowing the related struct pthreadpool_tevent_glue object
49  * to be removed from the struct pthreadpool_tevent glue list if the owning
50  * tevent_context is talloc_free()'ed.
51  */
52 struct pthreadpool_tevent_glue_ev_link {
53         struct pthreadpool_tevent_glue *glue;
54 };
55
56 struct pthreadpool_tevent {
57         struct pthreadpool *pool;
58         struct pthreadpool_tevent_glue *glue_list;
59
60         struct pthreadpool_tevent_job_state *jobs;
61 };
62
63 struct pthreadpool_tevent_job_state {
64         struct pthreadpool_tevent_job_state *prev, *next;
65         struct pthreadpool_tevent *pool;
66         struct tevent_context *ev;
67         struct tevent_immediate *im;
68         struct tevent_req *req;
69
70         void (*fn)(void *private_data);
71         void *private_data;
72 };
73
74 static int pthreadpool_tevent_destructor(struct pthreadpool_tevent *pool);
75
76 static int pthreadpool_tevent_job_signal(int jobid,
77                                          void (*job_fn)(void *private_data),
78                                          void *job_private_data,
79                                          void *private_data);
80
81 int pthreadpool_tevent_init(TALLOC_CTX *mem_ctx, unsigned max_threads,
82                             struct pthreadpool_tevent **presult)
83 {
84         struct pthreadpool_tevent *pool;
85         int ret;
86
87         pool = talloc_zero(mem_ctx, struct pthreadpool_tevent);
88         if (pool == NULL) {
89                 return ENOMEM;
90         }
91
92         ret = pthreadpool_init(max_threads, &pool->pool,
93                                pthreadpool_tevent_job_signal, pool);
94         if (ret != 0) {
95                 TALLOC_FREE(pool);
96                 return ret;
97         }
98
99         talloc_set_destructor(pool, pthreadpool_tevent_destructor);
100
101         *presult = pool;
102         return 0;
103 }
104
105 size_t pthreadpool_tevent_max_threads(struct pthreadpool_tevent *pool)
106 {
107         if (pool->pool == NULL) {
108                 return 0;
109         }
110
111         return pthreadpool_max_threads(pool->pool);
112 }
113
114 size_t pthreadpool_tevent_queued_jobs(struct pthreadpool_tevent *pool)
115 {
116         if (pool->pool == NULL) {
117                 return 0;
118         }
119
120         return pthreadpool_queued_jobs(pool->pool);
121 }
122
123 static int pthreadpool_tevent_destructor(struct pthreadpool_tevent *pool)
124 {
125         struct pthreadpool_tevent_job_state *state, *next;
126         struct pthreadpool_tevent_glue *glue = NULL;
127         int ret;
128
129         ret = pthreadpool_destroy(pool->pool);
130         if (ret != 0) {
131                 return ret;
132         }
133         pool->pool = NULL;
134
135         for (state = pool->jobs; state != NULL; state = next) {
136                 next = state->next;
137                 DLIST_REMOVE(pool->jobs, state);
138                 state->pool = NULL;
139         }
140
141         /*
142          * Delete all the registered
143          * tevent_context/tevent_threaded_context
144          * pairs.
145          */
146         for (glue = pool->glue_list; glue != NULL; glue = pool->glue_list) {
147                 /* The glue destructor removes it from the list */
148                 TALLOC_FREE(glue);
149         }
150         pool->glue_list = NULL;
151
152         return 0;
153 }
154
155 static int pthreadpool_tevent_glue_destructor(
156         struct pthreadpool_tevent_glue *glue)
157 {
158         if (glue->pool->glue_list != NULL) {
159                 DLIST_REMOVE(glue->pool->glue_list, glue);
160         }
161
162         /* Ensure the ev_link destructor knows we're gone */
163         glue->ev_link->glue = NULL;
164
165         TALLOC_FREE(glue->ev_link);
166         TALLOC_FREE(glue->tctx);
167
168         return 0;
169 }
170
171 /*
172  * Destructor called either explicitly from
173  * pthreadpool_tevent_glue_destructor(), or indirectly
174  * when owning tevent_context is destroyed.
175  *
176  * When called from pthreadpool_tevent_glue_destructor()
177  * ev_link->glue is already NULL, so this does nothing.
178  *
179  * When called from talloc_free() of the owning
180  * tevent_context we must ensure we also remove the
181  * linked glue object from the list inside
182  * struct pthreadpool_tevent.
183  */
184 static int pthreadpool_tevent_glue_link_destructor(
185         struct pthreadpool_tevent_glue_ev_link *ev_link)
186 {
187         TALLOC_FREE(ev_link->glue);
188         return 0;
189 }
190
191 static int pthreadpool_tevent_register_ev(struct pthreadpool_tevent *pool,
192                                           struct tevent_context *ev)
193 {
194         struct pthreadpool_tevent_glue *glue = NULL;
195         struct pthreadpool_tevent_glue_ev_link *ev_link = NULL;
196
197         /*
198          * See if this tevent_context was already registered by
199          * searching the glue object list. If so we have nothing
200          * to do here - we already have a tevent_context/tevent_threaded_context
201          * pair.
202          */
203         for (glue = pool->glue_list; glue != NULL; glue = glue->next) {
204                 if (glue->ev == ev) {
205                         return 0;
206                 }
207         }
208
209         /*
210          * Event context not yet registered - create a new glue
211          * object containing a tevent_context/tevent_threaded_context
212          * pair and put it on the list to remember this registration.
213          * We also need a link object to ensure the event context
214          * can't go away without us knowing about it.
215          */
216         glue = talloc_zero(pool, struct pthreadpool_tevent_glue);
217         if (glue == NULL) {
218                 return ENOMEM;
219         }
220         *glue = (struct pthreadpool_tevent_glue) {
221                 .pool = pool,
222                 .ev = ev,
223         };
224         talloc_set_destructor(glue, pthreadpool_tevent_glue_destructor);
225
226         /*
227          * Now allocate the link object to the event context. Note this
228          * is allocated OFF THE EVENT CONTEXT ITSELF, so if the event
229          * context is freed we are able to cleanup the glue object
230          * in the link object destructor.
231          */
232
233         ev_link = talloc_zero(ev, struct pthreadpool_tevent_glue_ev_link);
234         if (ev_link == NULL) {
235                 TALLOC_FREE(glue);
236                 return ENOMEM;
237         }
238         ev_link->glue = glue;
239         talloc_set_destructor(ev_link, pthreadpool_tevent_glue_link_destructor);
240
241         glue->ev_link = ev_link;
242
243 #ifdef HAVE_PTHREAD
244         glue->tctx = tevent_threaded_context_create(pool, ev);
245         if (glue->tctx == NULL) {
246                 TALLOC_FREE(ev_link);
247                 TALLOC_FREE(glue);
248                 return ENOMEM;
249         }
250 #endif
251
252         DLIST_ADD(pool->glue_list, glue);
253         return 0;
254 }
255
256 static void pthreadpool_tevent_job_fn(void *private_data);
257 static void pthreadpool_tevent_job_done(struct tevent_context *ctx,
258                                         struct tevent_immediate *im,
259                                         void *private_data);
260
261 static int pthreadpool_tevent_job_state_destructor(
262         struct pthreadpool_tevent_job_state *state)
263 {
264         if (state->pool == NULL) {
265                 return 0;
266         }
267
268         /*
269          * We should never be called with state->req == NULL,
270          * state->pool must be cleared before the 2nd talloc_free().
271          */
272         if (state->req == NULL) {
273                 abort();
274         }
275
276         /*
277          * We need to reparent to a long term context.
278          */
279         (void)talloc_reparent(state->req, NULL, state);
280         state->req = NULL;
281         return -1;
282 }
283
284 struct tevent_req *pthreadpool_tevent_job_send(
285         TALLOC_CTX *mem_ctx, struct tevent_context *ev,
286         struct pthreadpool_tevent *pool,
287         void (*fn)(void *private_data), void *private_data)
288 {
289         struct tevent_req *req;
290         struct pthreadpool_tevent_job_state *state;
291         int ret;
292
293         req = tevent_req_create(mem_ctx, &state,
294                                 struct pthreadpool_tevent_job_state);
295         if (req == NULL) {
296                 return NULL;
297         }
298         state->pool = pool;
299         state->ev = ev;
300         state->req = req;
301         state->fn = fn;
302         state->private_data = private_data;
303
304         if (pool == NULL) {
305                 tevent_req_error(req, EINVAL);
306                 return tevent_req_post(req, ev);
307         }
308         if (pool->pool == NULL) {
309                 tevent_req_error(req, EINVAL);
310                 return tevent_req_post(req, ev);
311         }
312
313         state->im = tevent_create_immediate(state);
314         if (tevent_req_nomem(state->im, req)) {
315                 return tevent_req_post(req, ev);
316         }
317
318         ret = pthreadpool_tevent_register_ev(pool, ev);
319         if (tevent_req_error(req, ret)) {
320                 return tevent_req_post(req, ev);
321         }
322
323         ret = pthreadpool_add_job(pool->pool, 0,
324                                   pthreadpool_tevent_job_fn,
325                                   state);
326         if (tevent_req_error(req, ret)) {
327                 return tevent_req_post(req, ev);
328         }
329
330         /*
331          * Once the job is scheduled, we need to protect
332          * our memory.
333          */
334         talloc_set_destructor(state, pthreadpool_tevent_job_state_destructor);
335
336         DLIST_ADD_END(pool->jobs, state);
337
338         return req;
339 }
340
341 static void pthreadpool_tevent_job_fn(void *private_data)
342 {
343         struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
344                 private_data, struct pthreadpool_tevent_job_state);
345         state->fn(state->private_data);
346 }
347
348 static int pthreadpool_tevent_job_signal(int jobid,
349                                          void (*job_fn)(void *private_data),
350                                          void *job_private_data,
351                                          void *private_data)
352 {
353         struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
354                 job_private_data, struct pthreadpool_tevent_job_state);
355         struct tevent_threaded_context *tctx = NULL;
356         struct pthreadpool_tevent_glue *g = NULL;
357
358         if (state->pool == NULL) {
359                 /* The pthreadpool_tevent is already gone */
360                 return 0;
361         }
362
363 #ifdef HAVE_PTHREAD
364         for (g = state->pool->glue_list; g != NULL; g = g->next) {
365                 if (g->ev == state->ev) {
366                         tctx = g->tctx;
367                         break;
368                 }
369         }
370
371         if (tctx == NULL) {
372                 abort();
373         }
374 #endif
375
376         if (tctx != NULL) {
377                 /* with HAVE_PTHREAD */
378                 tevent_threaded_schedule_immediate(tctx, state->im,
379                                                    pthreadpool_tevent_job_done,
380                                                    state);
381         } else {
382                 /* without HAVE_PTHREAD */
383                 tevent_schedule_immediate(state->im, state->ev,
384                                           pthreadpool_tevent_job_done,
385                                           state);
386         }
387
388         return 0;
389 }
390
391 static void pthreadpool_tevent_job_done(struct tevent_context *ctx,
392                                         struct tevent_immediate *im,
393                                         void *private_data)
394 {
395         struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
396                 private_data, struct pthreadpool_tevent_job_state);
397
398         if (state->pool != NULL) {
399                 DLIST_REMOVE(state->pool->jobs, state);
400                 state->pool = NULL;
401         }
402
403         if (state->req == NULL) {
404                 /*
405                  * There was a talloc_free() state->req
406                  * while the job was pending,
407                  * which mean we're reparented on a longterm
408                  * talloc context.
409                  *
410                  * We just cleanup here...
411                  */
412                 talloc_free(state);
413                 return;
414         }
415
416         tevent_req_done(state->req);
417 }
418
419 int pthreadpool_tevent_job_recv(struct tevent_req *req)
420 {
421         return tevent_req_simple_recv_unix(req);
422 }