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