pthreadpool: allocate glue->tctx on glue as memory context.
[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 *jobs;
61 };
62
63 struct pthreadpool_tevent_job_state {
64         struct tevent_context *ev;
65         struct tevent_req *req;
66         struct pthreadpool_tevent_job *job;
67 };
68
69 struct pthreadpool_tevent_job {
70         struct pthreadpool_tevent_job *prev, *next;
71
72         struct pthreadpool_tevent *pool;
73         struct pthreadpool_tevent_job_state *state;
74         struct tevent_immediate *im;
75
76         void (*fn)(void *private_data);
77         void *private_data;
78 };
79
80 static int pthreadpool_tevent_destructor(struct pthreadpool_tevent *pool);
81
82 static void pthreadpool_tevent_job_orphan(struct pthreadpool_tevent_job *job);
83
84 static struct pthreadpool_tevent_job *orphaned_jobs;
85
86 void pthreadpool_tevent_cleanup_orphaned_jobs(void)
87 {
88         struct pthreadpool_tevent_job *job = NULL;
89         struct pthreadpool_tevent_job *njob = NULL;
90
91         for (job = orphaned_jobs; job != NULL; job = njob) {
92                 njob = job->next;
93
94                 /*
95                  * The job destructor keeps the job alive
96                  * (and in the list) or removes it from the list.
97                  */
98                 TALLOC_FREE(job);
99         }
100 }
101
102 static int pthreadpool_tevent_job_signal(int jobid,
103                                          void (*job_fn)(void *private_data),
104                                          void *job_private_data,
105                                          void *private_data);
106
107 int pthreadpool_tevent_init(TALLOC_CTX *mem_ctx, unsigned max_threads,
108                             struct pthreadpool_tevent **presult)
109 {
110         struct pthreadpool_tevent *pool;
111         int ret;
112
113         pthreadpool_tevent_cleanup_orphaned_jobs();
114
115         pool = talloc_zero(mem_ctx, struct pthreadpool_tevent);
116         if (pool == NULL) {
117                 return ENOMEM;
118         }
119
120         ret = pthreadpool_init(max_threads, &pool->pool,
121                                pthreadpool_tevent_job_signal, pool);
122         if (ret != 0) {
123                 TALLOC_FREE(pool);
124                 return ret;
125         }
126
127         talloc_set_destructor(pool, pthreadpool_tevent_destructor);
128
129         *presult = pool;
130         return 0;
131 }
132
133 size_t pthreadpool_tevent_max_threads(struct pthreadpool_tevent *pool)
134 {
135         if (pool->pool == NULL) {
136                 return 0;
137         }
138
139         return pthreadpool_max_threads(pool->pool);
140 }
141
142 size_t pthreadpool_tevent_queued_jobs(struct pthreadpool_tevent *pool)
143 {
144         if (pool->pool == NULL) {
145                 return 0;
146         }
147
148         return pthreadpool_queued_jobs(pool->pool);
149 }
150
151 static int pthreadpool_tevent_destructor(struct pthreadpool_tevent *pool)
152 {
153         struct pthreadpool_tevent_job *job = NULL;
154         struct pthreadpool_tevent_job *njob = NULL;
155         struct pthreadpool_tevent_glue *glue = NULL;
156         int ret;
157
158         ret = pthreadpool_stop(pool->pool);
159         if (ret != 0) {
160                 return ret;
161         }
162
163         for (job = pool->jobs; job != NULL; job = njob) {
164                 njob = job->next;
165
166                 /* The job this removes it from the list */
167                 pthreadpool_tevent_job_orphan(job);
168         }
169
170         /*
171          * Delete all the registered
172          * tevent_context/tevent_threaded_context
173          * pairs.
174          */
175         for (glue = pool->glue_list; glue != NULL; glue = pool->glue_list) {
176                 /* The glue destructor removes it from the list */
177                 TALLOC_FREE(glue);
178         }
179         pool->glue_list = NULL;
180
181         ret = pthreadpool_destroy(pool->pool);
182         if (ret != 0) {
183                 return ret;
184         }
185         pool->pool = NULL;
186
187         pthreadpool_tevent_cleanup_orphaned_jobs();
188
189         return 0;
190 }
191
192 static int pthreadpool_tevent_glue_destructor(
193         struct pthreadpool_tevent_glue *glue)
194 {
195         if (glue->pool->glue_list != NULL) {
196                 DLIST_REMOVE(glue->pool->glue_list, glue);
197         }
198
199         /* Ensure the ev_link destructor knows we're gone */
200         glue->ev_link->glue = NULL;
201
202         TALLOC_FREE(glue->ev_link);
203         TALLOC_FREE(glue->tctx);
204
205         return 0;
206 }
207
208 /*
209  * Destructor called either explicitly from
210  * pthreadpool_tevent_glue_destructor(), or indirectly
211  * when owning tevent_context is destroyed.
212  *
213  * When called from pthreadpool_tevent_glue_destructor()
214  * ev_link->glue is already NULL, so this does nothing.
215  *
216  * When called from talloc_free() of the owning
217  * tevent_context we must ensure we also remove the
218  * linked glue object from the list inside
219  * struct pthreadpool_tevent.
220  */
221 static int pthreadpool_tevent_glue_link_destructor(
222         struct pthreadpool_tevent_glue_ev_link *ev_link)
223 {
224         TALLOC_FREE(ev_link->glue);
225         return 0;
226 }
227
228 static int pthreadpool_tevent_register_ev(struct pthreadpool_tevent *pool,
229                                           struct tevent_context *ev)
230 {
231         struct pthreadpool_tevent_glue *glue = NULL;
232         struct pthreadpool_tevent_glue_ev_link *ev_link = NULL;
233
234         /*
235          * See if this tevent_context was already registered by
236          * searching the glue object list. If so we have nothing
237          * to do here - we already have a tevent_context/tevent_threaded_context
238          * pair.
239          */
240         for (glue = pool->glue_list; glue != NULL; glue = glue->next) {
241                 if (glue->ev == ev) {
242                         return 0;
243                 }
244         }
245
246         /*
247          * Event context not yet registered - create a new glue
248          * object containing a tevent_context/tevent_threaded_context
249          * pair and put it on the list to remember this registration.
250          * We also need a link object to ensure the event context
251          * can't go away without us knowing about it.
252          */
253         glue = talloc_zero(pool, struct pthreadpool_tevent_glue);
254         if (glue == NULL) {
255                 return ENOMEM;
256         }
257         *glue = (struct pthreadpool_tevent_glue) {
258                 .pool = pool,
259                 .ev = ev,
260         };
261         talloc_set_destructor(glue, pthreadpool_tevent_glue_destructor);
262
263         /*
264          * Now allocate the link object to the event context. Note this
265          * is allocated OFF THE EVENT CONTEXT ITSELF, so if the event
266          * context is freed we are able to cleanup the glue object
267          * in the link object destructor.
268          */
269
270         ev_link = talloc_zero(ev, struct pthreadpool_tevent_glue_ev_link);
271         if (ev_link == NULL) {
272                 TALLOC_FREE(glue);
273                 return ENOMEM;
274         }
275         ev_link->glue = glue;
276         talloc_set_destructor(ev_link, pthreadpool_tevent_glue_link_destructor);
277
278         glue->ev_link = ev_link;
279
280 #ifdef HAVE_PTHREAD
281         glue->tctx = tevent_threaded_context_create(glue, ev);
282         if (glue->tctx == NULL) {
283                 TALLOC_FREE(ev_link);
284                 TALLOC_FREE(glue);
285                 return ENOMEM;
286         }
287 #endif
288
289         DLIST_ADD(pool->glue_list, glue);
290         return 0;
291 }
292
293 static void pthreadpool_tevent_job_fn(void *private_data);
294 static void pthreadpool_tevent_job_done(struct tevent_context *ctx,
295                                         struct tevent_immediate *im,
296                                         void *private_data);
297 static bool pthreadpool_tevent_job_cancel(struct tevent_req *req);
298
299 static int pthreadpool_tevent_job_destructor(struct pthreadpool_tevent_job *job)
300 {
301         /*
302          * We should never be called with state->state != NULL.
303          * Only pthreadpool_tevent_job_orphan() will call TALLOC_FREE(job)
304          * after detaching from the request state and pool list.
305          */
306         if (job->state != NULL) {
307                 abort();
308         }
309
310         /*
311          * If the job is not finished (job->im still there)
312          * and it's still attached to the pool,
313          * we try to cancel it (before it was starts)
314          */
315         if (job->im != NULL && job->pool != NULL) {
316                 size_t num;
317
318                 num = pthreadpool_cancel_job(job->pool->pool, 0,
319                                              pthreadpool_tevent_job_fn,
320                                              job);
321                 if (num != 0) {
322                         /*
323                          * It was not too late to cancel the request.
324                          *
325                          * We can remove job->im, as it will never be used.
326                          */
327                         TALLOC_FREE(job->im);
328                 }
329         }
330
331         /*
332          * pthreadpool_tevent_job_orphan() already removed
333          * it from pool->jobs. And we don't need try
334          * pthreadpool_cancel_job() again.
335          */
336         job->pool = NULL;
337
338         if (job->im != NULL) {
339                 /*
340                  * state->im still there means, we need to wait for the
341                  * immediate event to be triggered or just leak the memory.
342                  *
343                  * Move it to the orphaned list, if it's not already there.
344                  */
345                 return -1;
346         }
347
348         /*
349          * Finally remove from the orphaned_jobs list
350          * and let talloc destroy us.
351          */
352         DLIST_REMOVE(orphaned_jobs, job);
353
354         return 0;
355 }
356
357 static void pthreadpool_tevent_job_orphan(struct pthreadpool_tevent_job *job)
358 {
359         /*
360          * We're the only function that sets
361          * job->state = NULL;
362          */
363         if (job->state == NULL) {
364                 abort();
365         }
366
367         /*
368          * We need to reparent to a long term context.
369          * And detach from the request state.
370          * Maybe the destructor will keep the memory
371          * and leak it for now.
372          */
373         (void)talloc_reparent(job->state, NULL, job);
374         job->state->job = NULL;
375         job->state = NULL;
376
377         /*
378          * job->pool will only be set to NULL
379          * in the first destructur run.
380          */
381         if (job->pool == NULL) {
382                 abort();
383         }
384
385         /*
386          * Dettach it from the pool.
387          *
388          * The job might still be running,
389          * so we keep job->pool.
390          * The destructor will set it to NULL
391          * after trying pthreadpool_cancel_job()
392          */
393         DLIST_REMOVE(job->pool->jobs, job);
394
395         /*
396          * Add it to the list of orphaned jobs,
397          * which may be cleaned up later.
398          *
399          * The destructor removes it from the list
400          * when possible or it denies the free
401          * and keep it in the list.
402          */
403         DLIST_ADD_END(orphaned_jobs, job);
404         TALLOC_FREE(job);
405 }
406
407 static void pthreadpool_tevent_job_cleanup(struct tevent_req *req,
408                                            enum tevent_req_state req_state)
409 {
410         struct pthreadpool_tevent_job_state *state =
411                 tevent_req_data(req,
412                 struct pthreadpool_tevent_job_state);
413
414         if (state->job == NULL) {
415                 /*
416                  * The job request is not scheduled in the pool
417                  * yet or anymore.
418                  */
419                 return;
420         }
421
422         /*
423          * We need to reparent to a long term context.
424          * Maybe the destructor will keep the memory
425          * and leak it for now.
426          */
427         pthreadpool_tevent_job_orphan(state->job);
428         state->job = NULL; /* not needed but looks better */
429         return;
430 }
431
432 struct tevent_req *pthreadpool_tevent_job_send(
433         TALLOC_CTX *mem_ctx, struct tevent_context *ev,
434         struct pthreadpool_tevent *pool,
435         void (*fn)(void *private_data), void *private_data)
436 {
437         struct tevent_req *req = NULL;
438         struct pthreadpool_tevent_job_state *state = NULL;
439         struct pthreadpool_tevent_job *job = NULL;
440         int ret;
441
442         pthreadpool_tevent_cleanup_orphaned_jobs();
443
444         req = tevent_req_create(mem_ctx, &state,
445                                 struct pthreadpool_tevent_job_state);
446         if (req == NULL) {
447                 return NULL;
448         }
449         state->ev = ev;
450         state->req = req;
451
452         tevent_req_set_cleanup_fn(req, pthreadpool_tevent_job_cleanup);
453
454         if (pool == NULL) {
455                 tevent_req_error(req, EINVAL);
456                 return tevent_req_post(req, ev);
457         }
458         if (pool->pool == NULL) {
459                 tevent_req_error(req, EINVAL);
460                 return tevent_req_post(req, ev);
461         }
462
463         ret = pthreadpool_tevent_register_ev(pool, ev);
464         if (tevent_req_error(req, ret)) {
465                 return tevent_req_post(req, ev);
466         }
467
468         job = talloc_zero(state, struct pthreadpool_tevent_job);
469         if (tevent_req_nomem(job, req)) {
470                 return tevent_req_post(req, ev);
471         }
472         job->pool = pool;
473         job->fn = fn;
474         job->private_data = private_data;
475         job->im = tevent_create_immediate(state->job);
476         if (tevent_req_nomem(job->im, req)) {
477                 return tevent_req_post(req, ev);
478         }
479         talloc_set_destructor(job, pthreadpool_tevent_job_destructor);
480         DLIST_ADD_END(job->pool->jobs, job);
481         job->state = state;
482         state->job = job;
483
484         ret = pthreadpool_add_job(job->pool->pool, 0,
485                                   pthreadpool_tevent_job_fn,
486                                   job);
487         if (tevent_req_error(req, ret)) {
488                 return tevent_req_post(req, ev);
489         }
490
491         tevent_req_set_cancel_fn(req, pthreadpool_tevent_job_cancel);
492         return req;
493 }
494
495 static void pthreadpool_tevent_job_fn(void *private_data)
496 {
497         struct pthreadpool_tevent_job *job =
498                 talloc_get_type_abort(private_data,
499                 struct pthreadpool_tevent_job);
500
501         job->fn(job->private_data);
502 }
503
504 static int pthreadpool_tevent_job_signal(int jobid,
505                                          void (*job_fn)(void *private_data),
506                                          void *job_private_data,
507                                          void *private_data)
508 {
509         struct pthreadpool_tevent_job *job =
510                 talloc_get_type_abort(job_private_data,
511                 struct pthreadpool_tevent_job);
512         struct pthreadpool_tevent_job_state *state = job->state;
513         struct tevent_threaded_context *tctx = NULL;
514         struct pthreadpool_tevent_glue *g = NULL;
515
516         if (state == NULL) {
517                 /* Request already gone */
518                 return 0;
519         }
520
521 #ifdef HAVE_PTHREAD
522         for (g = job->pool->glue_list; g != NULL; g = g->next) {
523                 if (g->ev == state->ev) {
524                         tctx = g->tctx;
525                         break;
526                 }
527         }
528
529         if (tctx == NULL) {
530                 abort();
531         }
532 #endif
533
534         if (tctx != NULL) {
535                 /* with HAVE_PTHREAD */
536                 tevent_threaded_schedule_immediate(tctx, job->im,
537                                                    pthreadpool_tevent_job_done,
538                                                    job);
539         } else {
540                 /* without HAVE_PTHREAD */
541                 tevent_schedule_immediate(job->im, state->ev,
542                                           pthreadpool_tevent_job_done,
543                                           job);
544         }
545
546         return 0;
547 }
548
549 static void pthreadpool_tevent_job_done(struct tevent_context *ctx,
550                                         struct tevent_immediate *im,
551                                         void *private_data)
552 {
553         struct pthreadpool_tevent_job *job =
554                 talloc_get_type_abort(private_data,
555                 struct pthreadpool_tevent_job);
556         struct pthreadpool_tevent_job_state *state = job->state;
557
558         TALLOC_FREE(job->im);
559
560         if (state == NULL) {
561                 /* Request already gone */
562                 TALLOC_FREE(job);
563                 return;
564         }
565
566         /*
567          * pthreadpool_tevent_job_cleanup()
568          * will destroy the job.
569          */
570         tevent_req_done(state->req);
571 }
572
573 static bool pthreadpool_tevent_job_cancel(struct tevent_req *req)
574 {
575         struct pthreadpool_tevent_job_state *state =
576                 tevent_req_data(req,
577                 struct pthreadpool_tevent_job_state);
578         struct pthreadpool_tevent_job *job = state->job;
579         size_t num;
580
581         if (job == NULL) {
582                 return false;
583         }
584
585         num = pthreadpool_cancel_job(job->pool->pool, 0,
586                                      pthreadpool_tevent_job_fn,
587                                      job);
588         if (num == 0) {
589                 /*
590                  * It was too late to cancel the request.
591                  */
592                 return false;
593         }
594
595         /*
596          * It was not too late to cancel the request.
597          *
598          * We can remove job->im, as it will never be used.
599          */
600         TALLOC_FREE(job->im);
601
602         /*
603          * pthreadpool_tevent_job_cleanup()
604          * will destroy the job.
605          */
606         tevent_req_defer_callback(req, state->ev);
607         tevent_req_error(req, ECANCELED);
608         return true;
609 }
610
611 int pthreadpool_tevent_job_recv(struct tevent_req *req)
612 {
613         return tevent_req_simple_recv_unix(req);
614 }