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