source3/lib/fncall.c

   1 /*
   2  * Unix SMB/CIFS implementation.
   3  * Async fn calls
   4  * Copyright (C) Volker Lendecke 2009
   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 "includes.h"
  21 #include "../lib/util/tevent_unix.h"
  22
  23 #if WITH_PTHREADPOOL
  24
  25 #include "lib/pthreadpool/pthreadpool.h"
  26
  27 struct fncall_state {
  28         struct fncall_context *ctx;
  29         int job_id;
  30         bool done;
  31
  32         void *private_parent;
  33         void *job_private;
  34 };
  35
  36 struct fncall_context {
  37         struct pthreadpool *pool;
  38         int next_job_id;
  39         int sig_fd;
  40         struct tevent_req **pending;
  41
  42         struct fncall_state **orphaned;
  43         int num_orphaned;
  44
  45         struct fd_event *fde;
  46 };
  47
  48 static void fncall_handler(struct tevent_context *ev, struct tevent_fd *fde,
  49                            uint16_t flags, void *private_data);
  50
  51 static int fncall_context_destructor(struct fncall_context *ctx)
  52 {
  53         while (talloc_array_length(ctx->pending) != 0) {
  54                 /* No TALLOC_FREE here */
  55                 talloc_free(ctx->pending[0]);
  56         }
  57
  58         while (ctx->num_orphaned != 0) {
  59                 /*
  60                  * We've got jobs in the queue for which the tevent_req has
  61                  * been finished already. Wait for all of them to finish.
  62                  */
  63                 fncall_handler(NULL, NULL, TEVENT_FD_READ, ctx);
  64         }
  65
  66         pthreadpool_destroy(ctx->pool);
  67         ctx->pool = NULL;
  68
  69         return 0;
  70 }
  71
  72 struct fncall_context *fncall_context_init(TALLOC_CTX *mem_ctx,
  73                                            int max_threads)
  74 {
  75         struct fncall_context *ctx;
  76         int ret;
  77
  78         ctx = talloc_zero(mem_ctx, struct fncall_context);
  79         if (ctx == NULL) {
  80                 return NULL;
  81         }
  82
  83         ret = pthreadpool_init(max_threads, &ctx->pool);
  84         if (ret != 0) {
  85                 TALLOC_FREE(ctx);
  86                 return NULL;
  87         }
  88         talloc_set_destructor(ctx, fncall_context_destructor);
  89
  90         ctx->sig_fd = pthreadpool_signal_fd(ctx->pool);
  91         if (ctx->sig_fd == -1) {
  92                 TALLOC_FREE(ctx);
  93                 return NULL;
  94         }
  95
  96         return ctx;
  97 }
  98
  99 static int fncall_next_job_id(struct fncall_context *ctx)
 100 {
 101         int num_pending = talloc_array_length(ctx->pending);
 102         int result;
 103
 104         while (true) {
 105                 int i;
 106
 107                 result = ctx->next_job_id++;
 108                 if (result == 0) {
 109                         continue;
 110                 }
 111
 112                 for (i=0; i<num_pending; i++) {
 113                         struct fncall_state *state = tevent_req_data(
 114                                 ctx->pending[i], struct fncall_state);
 115
 116                         if (result == state->job_id) {
 117                                 break;
 118                         }
 119                 }
 120                 if (i == num_pending) {
 121                         return result;
 122                 }
 123         }
 124 }
 125
 126 static void fncall_unset_pending(struct tevent_req *req);
 127 static int fncall_destructor(struct tevent_req *req);
 128
 129 static bool fncall_set_pending(struct tevent_req *req,
 130                                struct fncall_context *ctx,
 131                                struct tevent_context *ev)
 132 {
 133         struct tevent_req **pending;
 134         int num_pending, orphaned_array_length;
 135
 136         num_pending = talloc_array_length(ctx->pending);
 137
 138         pending = talloc_realloc(ctx, ctx->pending, struct tevent_req *,
 139                                  num_pending+1);
 140         if (pending == NULL) {
 141                 return false;
 142         }
 143         pending[num_pending] = req;
 144         num_pending += 1;
 145         ctx->pending = pending;
 146         talloc_set_destructor(req, fncall_destructor);
 147
 148         /*
 149          * Make sure that the orphaned array of fncall_state structs has
 150          * enough space. A job can change from pending to orphaned in
 151          * fncall_destructor, and to fail in a talloc destructor should be
 152          * avoided if possible.
 153          */
 154
 155         orphaned_array_length = talloc_array_length(ctx->orphaned);
 156         if (num_pending > orphaned_array_length) {
 157                 struct fncall_state **orphaned;
 158
 159                 orphaned = talloc_realloc(ctx, ctx->orphaned,
 160                                           struct fncall_state *,
 161                                           orphaned_array_length + 1);
 162                 if (orphaned == NULL) {
 163                         fncall_unset_pending(req);
 164                         return false;
 165                 }
 166                 ctx->orphaned = orphaned;
 167         }
 168
 169         if (ctx->fde != NULL) {
 170                 return true;
 171         }
 172
 173         ctx->fde = tevent_add_fd(ev, ctx->pending, ctx->sig_fd, TEVENT_FD_READ,
 174                                  fncall_handler, ctx);
 175         if (ctx->fde == NULL) {
 176                 fncall_unset_pending(req);
 177                 return false;
 178         }
 179         return true;
 180 }
 181
 182 static void fncall_unset_pending(struct tevent_req *req)
 183 {
 184         struct fncall_state *state = tevent_req_data(req, struct fncall_state);
 185         struct fncall_context *ctx = state->ctx;
 186         int num_pending = talloc_array_length(ctx->pending);
 187         int i;
 188
 189         if (num_pending == 1) {
 190                 TALLOC_FREE(ctx->fde);
 191                 TALLOC_FREE(ctx->pending);
 192                 return;
 193         }
 194
 195         for (i=0; i<num_pending; i++) {
 196                 if (req == ctx->pending[i]) {
 197                         break;
 198                 }
 199         }
 200         if (i == num_pending) {
 201                 return;
 202         }
 203         if (num_pending > 1) {
 204                 ctx->pending[i] = ctx->pending[num_pending-1];
 205         }
 206         ctx->pending = talloc_realloc(NULL, ctx->pending, struct tevent_req *,
 207                                       num_pending - 1);
 208 }
 209
 210 static int fncall_destructor(struct tevent_req *req)
 211 {
 212         struct fncall_state *state = tevent_req_data(
 213                 req, struct fncall_state);
 214         struct fncall_context *ctx = state->ctx;
 215
 216         fncall_unset_pending(req);
 217
 218         if (state->done) {
 219                 return 0;
 220         }
 221
 222         /*
 223          * Keep around the state of the deleted request until the request has
 224          * finished in the helper thread. fncall_handler will destroy it.
 225          */
 226         ctx->orphaned[ctx->num_orphaned] = talloc_move(ctx->orphaned, &state);
 227         ctx->num_orphaned += 1;
 228
 229         return 0;
 230 }
 231
 232 struct tevent_req *fncall_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 233                                struct fncall_context *ctx,
 234                                void (*fn)(void *private_data),
 235                                void *private_data)
 236 {
 237         struct tevent_req *req;
 238         struct fncall_state *state;
 239         int ret;
 240
 241         req = tevent_req_create(mem_ctx, &state, struct fncall_state);
 242         if (req == NULL) {
 243                 return NULL;
 244         }
 245         state->ctx = ctx;
 246         state->job_id = fncall_next_job_id(state->ctx);
 247         state->done = false;
 248
 249         /*
 250          * We need to keep the private data we handed out to the thread around
 251          * as long as the job is not finished. This is a bit of an abstraction
 252          * violation, because the "req->state1->subreq->state2" (we're
 253          * "subreq", "req" is the request our caller creates) is broken to
 254          * "ctx->state2->state1", but we are right now in the destructor for
 255          * "subreq2", so what can we do. We need to keep state1 around,
 256          * otherwise the helper thread will have no place to put its results.
 257          */
 258
 259         state->private_parent = talloc_parent(private_data);
 260         state->job_private = talloc_move(state, &private_data);
 261
 262         ret = pthreadpool_add_job(state->ctx->pool, state->job_id, fn,
 263                                   state->job_private);
 264         if (ret == -1) {
 265                 tevent_req_error(req, errno);
 266                 return tevent_req_post(req, ev);
 267         }
 268         if (!fncall_set_pending(req, state->ctx, ev)) {
 269                 tevent_req_oom(req);
 270                 return tevent_req_post(req, ev);
 271         }
 272         return req;
 273 }
 274
 275 static void fncall_handler(struct tevent_context *ev, struct tevent_fd *fde,
 276                            uint16_t flags, void *private_data)
 277 {
 278         struct fncall_context *ctx = talloc_get_type_abort(
 279                 private_data, struct fncall_context);
 280         int i, num_pending;
 281         int job_id;
 282
 283         if (pthreadpool_finished_job(ctx->pool, &job_id) != 0) {
 284                 return;
 285         }
 286
 287         num_pending = talloc_array_length(ctx->pending);
 288
 289         for (i=0; i<num_pending; i++) {
 290                 struct fncall_state *state = tevent_req_data(
 291                         ctx->pending[i], struct fncall_state);
 292
 293                 if (job_id == state->job_id) {
 294                         state->done = true;
 295                         talloc_move(state->private_parent,
 296                                     &state->job_private);
 297                         tevent_req_done(ctx->pending[i]);
 298                         return;
 299                 }
 300         }
 301
 302         for (i=0; i<ctx->num_orphaned; i++) {
 303                 if (job_id == ctx->orphaned[i]->job_id) {
 304                         break;
 305                 }
 306         }
 307         if (i == ctx->num_orphaned) {
 308                 return;
 309         }
 310
 311         TALLOC_FREE(ctx->orphaned[i]);
 312
 313         if (i < ctx->num_orphaned-1) {
 314                 ctx->orphaned[i] = ctx->orphaned[ctx->num_orphaned-1];
 315         }
 316         ctx->num_orphaned -= 1;
 317 }
 318
 319 int fncall_recv(struct tevent_req *req, int *perr)
 320 {
 321         if (tevent_req_is_unix_error(req, perr)) {
 322                 return -1;
 323         }
 324         return 0;
 325 }
 326
 327 #else  /* WITH_PTHREADPOOL */
 328
 329 struct fncall_context {
 330         uint8_t dummy;
 331 };
 332
 333 struct fncall_context *fncall_context_init(TALLOC_CTX *mem_ctx,
 334                                            int max_threads)
 335 {
 336         return talloc(mem_ctx, struct fncall_context);
 337 }
 338
 339 struct fncall_state {
 340         uint8_t dummy;
 341 };
 342
 343 struct tevent_req *fncall_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 344                                struct fncall_context *ctx,
 345                                void (*fn)(void *private_data),
 346                                void *private_data)
 347 {
 348         struct tevent_req *req;
 349         struct fncall_state *state;
 350
 351         req = tevent_req_create(mem_ctx, &state, struct fncall_state);
 352         if (req == NULL) {
 353                 return NULL;
 354         }
 355         fn(private_data);
 356         tevent_req_post(req, ev);
 357         return req;
 358 }
 359
 360 int fncall_recv(struct tevent_req *req, int *perr)
 361 {
 362         return 0;
 363 }
 364
 365 #endif