source3/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 struct pthreadpool_tevent {
  29         struct pthreadpool *pool;
  30
  31         struct pthreadpool_tevent_job_state *jobs;
  32 };
  33
  34 static int pthreadpool_tevent_destructor(struct pthreadpool_tevent *pool);
  35
  36 static int pthreadpool_tevent_job_signal(int jobid,
  37                                          void (*job_fn)(void *private_data),
  38                                          void *job_private_data,
  39                                          void *private_data);
  40
  41 int pthreadpool_tevent_init(TALLOC_CTX *mem_ctx, unsigned max_threads,
  42                             struct pthreadpool_tevent **presult)
  43 {
  44         struct pthreadpool_tevent *pool;
  45         int ret;
  46
  47         pool = talloc_zero(mem_ctx, struct pthreadpool_tevent);
  48         if (pool == NULL) {
  49                 return ENOMEM;
  50         }
  51
  52         ret = pthreadpool_init(max_threads, &pool->pool,
  53                                pthreadpool_tevent_job_signal, pool);
  54         if (ret != 0) {
  55                 TALLOC_FREE(pool);
  56                 return ret;
  57         }
  58
  59         talloc_set_destructor(pool, pthreadpool_tevent_destructor);
  60
  61         *presult = pool;
  62         return 0;
  63 }
  64
  65 static int pthreadpool_tevent_destructor(struct pthreadpool_tevent *pool)
  66 {
  67         int ret;
  68
  69         ret = pthreadpool_destroy(pool->pool);
  70         if (ret != 0) {
  71                 return ret;
  72         }
  73         pool->pool = NULL;
  74
  75         if (pool->jobs != NULL) {
  76                 abort();
  77         }
  78
  79         return 0;
  80 }
  81
  82 struct pthreadpool_tevent_job_state {
  83         struct pthreadpool_tevent_job_state *prev, *next;
  84         struct pthreadpool_tevent *pool;
  85         struct tevent_context *ev;
  86         struct tevent_threaded_context *tctx;
  87         struct tevent_immediate *im;
  88         struct tevent_req *req;
  89
  90         void (*fn)(void *private_data);
  91         void *private_data;
  92 };
  93
  94 static void pthreadpool_tevent_job_fn(void *private_data);
  95 static void pthreadpool_tevent_job_done(struct tevent_context *ctx,
  96                                         struct tevent_immediate *im,
  97                                         void *private_data);
  98
  99 static int pthreadpool_tevent_job_state_destructor(
 100         struct pthreadpool_tevent_job_state *state)
 101 {
 102         if (state->pool == NULL) {
 103                 return 0;
 104         }
 105
 106         /*
 107          * We should never be called with state->req == NULL,
 108          * state->pool must be cleared before the 2nd talloc_free().
 109          */
 110         if (state->req == NULL) {
 111                 abort();
 112         }
 113
 114         /*
 115          * We need to reparent to a long term context.
 116          */
 117         (void)talloc_reparent(state->req, state->pool, state);
 118         state->req = NULL;
 119         return -1;
 120 }
 121
 122 struct tevent_req *pthreadpool_tevent_job_send(
 123         TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 124         struct pthreadpool_tevent *pool,
 125         void (*fn)(void *private_data), void *private_data)
 126 {
 127         struct tevent_req *req;
 128         struct pthreadpool_tevent_job_state *state;
 129         int ret;
 130
 131         req = tevent_req_create(mem_ctx, &state,
 132                                 struct pthreadpool_tevent_job_state);
 133         if (req == NULL) {
 134                 return NULL;
 135         }
 136         state->pool = pool;
 137         state->ev = ev;
 138         state->req = req;
 139         state->fn = fn;
 140         state->private_data = private_data;
 141
 142         state->im = tevent_create_immediate(state);
 143         if (tevent_req_nomem(state->im, req)) {
 144                 return tevent_req_post(req, ev);
 145         }
 146
 147 #ifdef HAVE_PTHREAD
 148         state->tctx = tevent_threaded_context_create(state, ev);
 149         if (state->tctx == NULL && errno == ENOSYS) {
 150                 /*
 151                  * Samba build with pthread support but
 152                  * tevent without???
 153                  */
 154                 tevent_req_error(req, ENOSYS);
 155                 return tevent_req_post(req, ev);
 156         }
 157         if (tevent_req_nomem(state->tctx, req)) {
 158                 return tevent_req_post(req, ev);
 159         }
 160 #endif
 161
 162         ret = pthreadpool_add_job(pool->pool, 0,
 163                                   pthreadpool_tevent_job_fn,
 164                                   state);
 165         if (tevent_req_error(req, ret)) {
 166                 return tevent_req_post(req, ev);
 167         }
 168
 169         /*
 170          * Once the job is scheduled, we need to protect
 171          * our memory.
 172          */
 173         talloc_set_destructor(state, pthreadpool_tevent_job_state_destructor);
 174
 175         DLIST_ADD_END(pool->jobs, state);
 176
 177         return req;
 178 }
 179
 180 static void pthreadpool_tevent_job_fn(void *private_data)
 181 {
 182         struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
 183                 private_data, struct pthreadpool_tevent_job_state);
 184         state->fn(state->private_data);
 185 }
 186
 187 static int pthreadpool_tevent_job_signal(int jobid,
 188                                          void (*job_fn)(void *private_data),
 189                                          void *job_private_data,
 190                                          void *private_data)
 191 {
 192         struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
 193                 job_private_data, struct pthreadpool_tevent_job_state);
 194
 195         if (state->tctx != NULL) {
 196                 /* with HAVE_PTHREAD */
 197                 tevent_threaded_schedule_immediate(state->tctx, state->im,
 198                                                    pthreadpool_tevent_job_done,
 199                                                    state);
 200         } else {
 201                 /* without HAVE_PTHREAD */
 202                 tevent_schedule_immediate(state->im, state->ev,
 203                                           pthreadpool_tevent_job_done,
 204                                           state);
 205         }
 206
 207         return 0;
 208 }
 209
 210 static void pthreadpool_tevent_job_done(struct tevent_context *ctx,
 211                                         struct tevent_immediate *im,
 212                                         void *private_data)
 213 {
 214         struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
 215                 private_data, struct pthreadpool_tevent_job_state);
 216
 217         DLIST_REMOVE(state->pool->jobs, state);
 218         state->pool = NULL;
 219
 220         TALLOC_FREE(state->tctx);
 221
 222         if (state->req == NULL) {
 223                 /*
 224                  * There was a talloc_free() state->req
 225                  * while the job was pending,
 226                  * which mean we're reparented on a longterm
 227                  * talloc context.
 228                  *
 229                  * We just cleanup here...
 230                  */
 231                 talloc_free(state);
 232                 return;
 233         }
 234
 235         tevent_req_done(state->req);
 236 }
 237
 238 int pthreadpool_tevent_job_recv(struct tevent_req *req)
 239 {
 240         return tevent_req_simple_recv_unix(req);
 241 }