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         state->im = tevent_create_immediate(state);
 305         if (tevent_req_nomem(state->im, req)) {
 306                 return tevent_req_post(req, ev);
 307         }
 308
 309         ret = pthreadpool_tevent_register_ev(pool, ev);
 310         if (tevent_req_error(req, ret)) {
 311                 return tevent_req_post(req, ev);
 312         }
 313
 314         ret = pthreadpool_add_job(pool->pool, 0,
 315                                   pthreadpool_tevent_job_fn,
 316                                   state);
 317         if (tevent_req_error(req, ret)) {
 318                 return tevent_req_post(req, ev);
 319         }
 320
 321         /*
 322          * Once the job is scheduled, we need to protect
 323          * our memory.
 324          */
 325         talloc_set_destructor(state, pthreadpool_tevent_job_state_destructor);
 326
 327         DLIST_ADD_END(pool->jobs, state);
 328
 329         return req;
 330 }
 331
 332 static void pthreadpool_tevent_job_fn(void *private_data)
 333 {
 334         struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
 335                 private_data, struct pthreadpool_tevent_job_state);
 336         state->fn(state->private_data);
 337 }
 338
 339 static int pthreadpool_tevent_job_signal(int jobid,
 340                                          void (*job_fn)(void *private_data),
 341                                          void *job_private_data,
 342                                          void *private_data)
 343 {
 344         struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
 345                 job_private_data, struct pthreadpool_tevent_job_state);
 346         struct tevent_threaded_context *tctx = NULL;
 347         struct pthreadpool_tevent_glue *g = NULL;
 348
 349         if (state->pool == NULL) {
 350                 /* The pthreadpool_tevent is already gone */
 351                 return 0;
 352         }
 353
 354 #ifdef HAVE_PTHREAD
 355         for (g = state->pool->glue_list; g != NULL; g = g->next) {
 356                 if (g->ev == state->ev) {
 357                         tctx = g->tctx;
 358                         break;
 359                 }
 360         }
 361
 362         if (tctx == NULL) {
 363                 abort();
 364         }
 365 #endif
 366
 367         if (tctx != NULL) {
 368                 /* with HAVE_PTHREAD */
 369                 tevent_threaded_schedule_immediate(tctx, state->im,
 370                                                    pthreadpool_tevent_job_done,
 371                                                    state);
 372         } else {
 373                 /* without HAVE_PTHREAD */
 374                 tevent_schedule_immediate(state->im, state->ev,
 375                                           pthreadpool_tevent_job_done,
 376                                           state);
 377         }
 378
 379         return 0;
 380 }
 381
 382 static void pthreadpool_tevent_job_done(struct tevent_context *ctx,
 383                                         struct tevent_immediate *im,
 384                                         void *private_data)
 385 {
 386         struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
 387                 private_data, struct pthreadpool_tevent_job_state);
 388
 389         if (state->pool != NULL) {
 390                 DLIST_REMOVE(state->pool->jobs, state);
 391                 state->pool = NULL;
 392         }
 393
 394         if (state->req == NULL) {
 395                 /*
 396                  * There was a talloc_free() state->req
 397                  * while the job was pending,
 398                  * which mean we're reparented on a longterm
 399                  * talloc context.
 400                  *
 401                  * We just cleanup here...
 402                  */
 403                 talloc_free(state);
 404                 return;
 405         }
 406
 407         tevent_req_done(state->req);
 408 }
 409
 410 int pthreadpool_tevent_job_recv(struct tevent_req *req)
 411 {
 412         return tevent_req_simple_recv_unix(req);
 413 }