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 static int pthreadpool_tevent_destructor(struct pthreadpool_tevent *pool)
 106 {
 107         struct pthreadpool_tevent_job_state *state, *next;
 108         struct pthreadpool_tevent_glue *glue = NULL;
 109         int ret;
 110
 111         ret = pthreadpool_destroy(pool->pool);
 112         if (ret != 0) {
 113                 return ret;
 114         }
 115         pool->pool = NULL;
 116
 117         for (state = pool->jobs; state != NULL; state = next) {
 118                 next = state->next;
 119                 DLIST_REMOVE(pool->jobs, state);
 120                 state->pool = NULL;
 121         }
 122
 123         /*
 124          * Delete all the registered
 125          * tevent_context/tevent_threaded_context
 126          * pairs.
 127          */
 128         for (glue = pool->glue_list; glue != NULL; glue = pool->glue_list) {
 129                 /* The glue destructor removes it from the list */
 130                 TALLOC_FREE(glue);
 131         }
 132         pool->glue_list = NULL;
 133
 134         return 0;
 135 }
 136
 137 static int pthreadpool_tevent_glue_destructor(
 138         struct pthreadpool_tevent_glue *glue)
 139 {
 140         if (glue->pool->glue_list != NULL) {
 141                 DLIST_REMOVE(glue->pool->glue_list, glue);
 142         }
 143
 144         /* Ensure the ev_link destructor knows we're gone */
 145         glue->ev_link->glue = NULL;
 146
 147         TALLOC_FREE(glue->ev_link);
 148         TALLOC_FREE(glue->tctx);
 149
 150         return 0;
 151 }
 152
 153 /*
 154  * Destructor called either explicitly from
 155  * pthreadpool_tevent_glue_destructor(), or indirectly
 156  * when owning tevent_context is destroyed.
 157  *
 158  * When called from pthreadpool_tevent_glue_destructor()
 159  * ev_link->glue is already NULL, so this does nothing.
 160  *
 161  * When called from talloc_free() of the owning
 162  * tevent_context we must ensure we also remove the
 163  * linked glue object from the list inside
 164  * struct pthreadpool_tevent.
 165  */
 166 static int pthreadpool_tevent_glue_link_destructor(
 167         struct pthreadpool_tevent_glue_ev_link *ev_link)
 168 {
 169         TALLOC_FREE(ev_link->glue);
 170         return 0;
 171 }
 172
 173 static int pthreadpool_tevent_register_ev(struct pthreadpool_tevent *pool,
 174                                           struct tevent_context *ev)
 175 {
 176         struct pthreadpool_tevent_glue *glue = NULL;
 177         struct pthreadpool_tevent_glue_ev_link *ev_link = NULL;
 178
 179         /*
 180          * See if this tevent_context was already registered by
 181          * searching the glue object list. If so we have nothing
 182          * to do here - we already have a tevent_context/tevent_threaded_context
 183          * pair.
 184          */
 185         for (glue = pool->glue_list; glue != NULL; glue = glue->next) {
 186                 if (glue->ev == ev) {
 187                         return 0;
 188                 }
 189         }
 190
 191         /*
 192          * Event context not yet registered - create a new glue
 193          * object containing a tevent_context/tevent_threaded_context
 194          * pair and put it on the list to remember this registration.
 195          * We also need a link object to ensure the event context
 196          * can't go away without us knowing about it.
 197          */
 198         glue = talloc_zero(pool, struct pthreadpool_tevent_glue);
 199         if (glue == NULL) {
 200                 return ENOMEM;
 201         }
 202         *glue = (struct pthreadpool_tevent_glue) {
 203                 .pool = pool,
 204                 .ev = ev,
 205         };
 206         talloc_set_destructor(glue, pthreadpool_tevent_glue_destructor);
 207
 208         /*
 209          * Now allocate the link object to the event context. Note this
 210          * is allocated OFF THE EVENT CONTEXT ITSELF, so if the event
 211          * context is freed we are able to cleanup the glue object
 212          * in the link object destructor.
 213          */
 214
 215         ev_link = talloc_zero(ev, struct pthreadpool_tevent_glue_ev_link);
 216         if (ev_link == NULL) {
 217                 TALLOC_FREE(glue);
 218                 return ENOMEM;
 219         }
 220         ev_link->glue = glue;
 221         talloc_set_destructor(ev_link, pthreadpool_tevent_glue_link_destructor);
 222
 223         glue->ev_link = ev_link;
 224
 225 #ifdef HAVE_PTHREAD
 226         glue->tctx = tevent_threaded_context_create(pool, ev);
 227         if (glue->tctx == NULL) {
 228                 TALLOC_FREE(ev_link);
 229                 TALLOC_FREE(glue);
 230                 return ENOMEM;
 231         }
 232 #endif
 233
 234         DLIST_ADD(pool->glue_list, glue);
 235         return 0;
 236 }
 237
 238 static void pthreadpool_tevent_job_fn(void *private_data);
 239 static void pthreadpool_tevent_job_done(struct tevent_context *ctx,
 240                                         struct tevent_immediate *im,
 241                                         void *private_data);
 242
 243 static int pthreadpool_tevent_job_state_destructor(
 244         struct pthreadpool_tevent_job_state *state)
 245 {
 246         if (state->pool == NULL) {
 247                 return 0;
 248         }
 249
 250         /*
 251          * We should never be called with state->req == NULL,
 252          * state->pool must be cleared before the 2nd talloc_free().
 253          */
 254         if (state->req == NULL) {
 255                 abort();
 256         }
 257
 258         /*
 259          * We need to reparent to a long term context.
 260          */
 261         (void)talloc_reparent(state->req, NULL, state);
 262         state->req = NULL;
 263         return -1;
 264 }
 265
 266 struct tevent_req *pthreadpool_tevent_job_send(
 267         TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 268         struct pthreadpool_tevent *pool,
 269         void (*fn)(void *private_data), void *private_data)
 270 {
 271         struct tevent_req *req;
 272         struct pthreadpool_tevent_job_state *state;
 273         int ret;
 274
 275         req = tevent_req_create(mem_ctx, &state,
 276                                 struct pthreadpool_tevent_job_state);
 277         if (req == NULL) {
 278                 return NULL;
 279         }
 280         state->pool = pool;
 281         state->ev = ev;
 282         state->req = req;
 283         state->fn = fn;
 284         state->private_data = private_data;
 285
 286         state->im = tevent_create_immediate(state);
 287         if (tevent_req_nomem(state->im, req)) {
 288                 return tevent_req_post(req, ev);
 289         }
 290
 291         ret = pthreadpool_tevent_register_ev(pool, ev);
 292         if (ret != 0) {
 293                 tevent_req_error(req, errno);
 294                 return tevent_req_post(req, ev);
 295         }
 296
 297         ret = pthreadpool_add_job(pool->pool, 0,
 298                                   pthreadpool_tevent_job_fn,
 299                                   state);
 300         if (tevent_req_error(req, ret)) {
 301                 return tevent_req_post(req, ev);
 302         }
 303
 304         /*
 305          * Once the job is scheduled, we need to protect
 306          * our memory.
 307          */
 308         talloc_set_destructor(state, pthreadpool_tevent_job_state_destructor);
 309
 310         DLIST_ADD_END(pool->jobs, state);
 311
 312         return req;
 313 }
 314
 315 static void pthreadpool_tevent_job_fn(void *private_data)
 316 {
 317         struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
 318                 private_data, struct pthreadpool_tevent_job_state);
 319         state->fn(state->private_data);
 320 }
 321
 322 static int pthreadpool_tevent_job_signal(int jobid,
 323                                          void (*job_fn)(void *private_data),
 324                                          void *job_private_data,
 325                                          void *private_data)
 326 {
 327         struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
 328                 job_private_data, struct pthreadpool_tevent_job_state);
 329         struct tevent_threaded_context *tctx = NULL;
 330         struct pthreadpool_tevent_glue *g = NULL;
 331
 332         if (state->pool == NULL) {
 333                 /* The pthreadpool_tevent is already gone */
 334                 return 0;
 335         }
 336
 337 #ifdef HAVE_PTHREAD
 338         for (g = state->pool->glue_list; g != NULL; g = g->next) {
 339                 if (g->ev == state->ev) {
 340                         tctx = g->tctx;
 341                         break;
 342                 }
 343         }
 344
 345         if (tctx == NULL) {
 346                 abort();
 347         }
 348 #endif
 349
 350         if (tctx != NULL) {
 351                 /* with HAVE_PTHREAD */
 352                 tevent_threaded_schedule_immediate(tctx, state->im,
 353                                                    pthreadpool_tevent_job_done,
 354                                                    state);
 355         } else {
 356                 /* without HAVE_PTHREAD */
 357                 tevent_schedule_immediate(state->im, state->ev,
 358                                           pthreadpool_tevent_job_done,
 359                                           state);
 360         }
 361
 362         return 0;
 363 }
 364
 365 static void pthreadpool_tevent_job_done(struct tevent_context *ctx,
 366                                         struct tevent_immediate *im,
 367                                         void *private_data)
 368 {
 369         struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
 370                 private_data, struct pthreadpool_tevent_job_state);
 371
 372         if (state->pool != NULL) {
 373                 DLIST_REMOVE(state->pool->jobs, state);
 374                 state->pool = NULL;
 375         }
 376
 377         if (state->req == NULL) {
 378                 /*
 379                  * There was a talloc_free() state->req
 380                  * while the job was pending,
 381                  * which mean we're reparented on a longterm
 382                  * talloc context.
 383                  *
 384                  * We just cleanup here...
 385                  */
 386                 talloc_free(state);
 387                 return;
 388         }
 389
 390         tevent_req_done(state->req);
 391 }
 392
 393 int pthreadpool_tevent_job_recv(struct tevent_req *req)
 394 {
 395         return tevent_req_simple_recv_unix(req);
 396 }