source3/lib/asys/asys.c

   1 /*
   2  * Async syscalls
   3  * Copyright (C) Volker Lendecke 2012
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation; either version 3 of the License, or
   8  * (at your option) any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License
  16  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17  */
  18
  19 #include "asys.h"
  20 #include <stdlib.h>
  21 #include <errno.h>
  22 #include "../pthreadpool/pthreadpool.h"
  23
  24 struct asys_pwrite_args {
  25         int fildes;
  26         const void *buf;
  27         size_t nbyte;
  28         off_t offset;
  29 };
  30
  31 struct asys_pread_args {
  32         int fildes;
  33         void *buf;
  34         size_t nbyte;
  35         off_t offset;
  36 };
  37
  38 struct asys_fsync_args {
  39         int fildes;
  40 };
  41
  42 union asys_job_args {
  43         struct asys_pwrite_args pwrite_args;
  44         struct asys_pread_args pread_args;
  45         struct asys_fsync_args fsync_args;
  46 };
  47
  48 struct asys_job {
  49         void *private_data;
  50         union asys_job_args args;
  51         ssize_t ret;
  52         int err;
  53         char busy;
  54         char canceled;
  55 };
  56
  57 struct asys_context {
  58         struct pthreadpool *pool;
  59         int pthreadpool_fd;
  60
  61         unsigned num_jobs;
  62         struct asys_job **jobs;
  63 };
  64
  65 struct asys_creds_context {
  66         int dummy;
  67 };
  68
  69 int asys_context_init(struct asys_context **pctx, unsigned max_parallel)
  70 {
  71         struct asys_context *ctx;
  72         int ret;
  73
  74         ctx = calloc(1, sizeof(struct asys_context));
  75         if (ctx == NULL) {
  76                 return ENOMEM;
  77         }
  78         ret = pthreadpool_init(max_parallel, &ctx->pool);
  79         if (ret != 0) {
  80                 free(ctx);
  81                 return ret;
  82         }
  83         ctx->pthreadpool_fd = pthreadpool_signal_fd(ctx->pool);
  84
  85         *pctx = ctx;
  86         return 0;
  87 }
  88
  89 int asys_signalfd(struct asys_context *ctx)
  90 {
  91         return ctx->pthreadpool_fd;
  92 }
  93
  94 int asys_context_destroy(struct asys_context *ctx)
  95 {
  96         int ret;
  97         unsigned i;
  98
  99         for (i=0; i<ctx->num_jobs; i++) {
 100                 if (ctx->jobs[i]->busy) {
 101                         return EBUSY;
 102                 }
 103         }
 104
 105         ret = pthreadpool_destroy(ctx->pool);
 106         if (ret != 0) {
 107                 return ret;
 108         }
 109         for (i=0; i<ctx->num_jobs; i++) {
 110                 free(ctx->jobs[i]);
 111         }
 112         free(ctx->jobs);
 113         free(ctx);
 114         return 0;
 115 }
 116
 117 static int asys_new_job(struct asys_context *ctx, int *jobid,
 118                         struct asys_job **pjob)
 119 {
 120         struct asys_job **tmp;
 121         struct asys_job *job;
 122         unsigned i;
 123
 124         for (i=0; i<ctx->num_jobs; i++) {
 125                 job = ctx->jobs[i];
 126                 if (!job->busy) {
 127                         job->err = 0;
 128                         *pjob = job;
 129                         *jobid = i;
 130                         return 0;
 131                 }
 132         }
 133
 134         if (ctx->num_jobs+1 == 0) {
 135                 return EBUSY;   /* overflow */
 136         }
 137
 138         tmp = realloc(ctx->jobs, sizeof(struct asys_job *)*(ctx->num_jobs+1));
 139         if (tmp == NULL) {
 140                 return ENOMEM;
 141         }
 142         ctx->jobs = tmp;
 143
 144         job = calloc(1, sizeof(struct asys_job));
 145         if (job == NULL) {
 146                 return ENOMEM;
 147         }
 148         ctx->jobs[ctx->num_jobs] = job;
 149
 150         *jobid = ctx->num_jobs;
 151         *pjob = job;
 152         ctx->num_jobs += 1;
 153         return 0;
 154 }
 155
 156 static void asys_pwrite_do(void *private_data);
 157
 158 int asys_pwrite(struct asys_context *ctx, int fildes, const void *buf,
 159                 size_t nbyte, off_t offset, void *private_data)
 160 {
 161         struct asys_job *job;
 162         struct asys_pwrite_args *args;
 163         int jobid;
 164         int ret;
 165
 166         ret = asys_new_job(ctx, &jobid, &job);
 167         if (ret != 0) {
 168                 return ret;
 169         }
 170         job->private_data = private_data;
 171
 172         args = &job->args.pwrite_args;
 173         args->fildes = fildes;
 174         args->buf = buf;
 175         args->nbyte = nbyte;
 176         args->offset = offset;
 177
 178         ret = pthreadpool_add_job(ctx->pool, jobid, asys_pwrite_do, job);
 179         if (ret != 0) {
 180                 return ret;
 181         }
 182         job->busy = 1;
 183
 184         return 0;
 185 }
 186
 187 static void asys_pwrite_do(void *private_data)
 188 {
 189         struct asys_job *job = (struct asys_job *)private_data;
 190         struct asys_pwrite_args *args = &job->args.pwrite_args;
 191
 192         job->ret = pwrite(args->fildes, args->buf, args->nbyte, args->offset);
 193         if (job->ret == -1) {
 194                 job->err = errno;
 195         }
 196 }
 197
 198 static void asys_pread_do(void *private_data);
 199
 200 int asys_pread(struct asys_context *ctx, int fildes, void *buf,
 201                size_t nbyte, off_t offset, void *private_data)
 202 {
 203         struct asys_job *job;
 204         struct asys_pread_args *args;
 205         int jobid;
 206         int ret;
 207
 208         ret = asys_new_job(ctx, &jobid, &job);
 209         if (ret != 0) {
 210                 return ret;
 211         }
 212         job->private_data = private_data;
 213
 214         args = &job->args.pread_args;
 215         args->fildes = fildes;
 216         args->buf = buf;
 217         args->nbyte = nbyte;
 218         args->offset = offset;
 219
 220         ret = pthreadpool_add_job(ctx->pool, jobid, asys_pread_do, job);
 221         if (ret != 0) {
 222                 return ret;
 223         }
 224         job->busy = 1;
 225
 226         return 0;
 227 }
 228
 229 static void asys_pread_do(void *private_data)
 230 {
 231         struct asys_job *job = (struct asys_job *)private_data;
 232         struct asys_pread_args *args = &job->args.pread_args;
 233
 234         job->ret = pread(args->fildes, args->buf, args->nbyte, args->offset);
 235         if (job->ret == -1) {
 236                 job->err = errno;
 237         }
 238 }
 239
 240 static void asys_fsync_do(void *private_data);
 241
 242 int asys_fsync(struct asys_context *ctx, int fildes, void *private_data)
 243 {
 244         struct asys_job *job;
 245         struct asys_fsync_args *args;
 246         int jobid;
 247         int ret;
 248
 249         ret = asys_new_job(ctx, &jobid, &job);
 250         if (ret != 0) {
 251                 return ret;
 252         }
 253         job->private_data = private_data;
 254
 255         args = &job->args.fsync_args;
 256         args->fildes = fildes;
 257
 258         ret = pthreadpool_add_job(ctx->pool, jobid, asys_fsync_do, job);
 259         if (ret != 0) {
 260                 return ret;
 261         }
 262         job->busy = 1;
 263
 264         return 0;
 265 }
 266
 267 static void asys_fsync_do(void *private_data)
 268 {
 269         struct asys_job *job = (struct asys_job *)private_data;
 270         struct asys_fsync_args *args = &job->args.fsync_args;
 271
 272         job->ret = fsync(args->fildes);
 273         if (job->ret == -1) {
 274                 job->err = errno;
 275         }
 276 }
 277
 278 void asys_cancel(struct asys_context *ctx, void *private_data)
 279 {
 280         unsigned i;
 281
 282         for (i=0; i<ctx->num_jobs; i++) {
 283                 struct asys_job *job = ctx->jobs[i];
 284
 285                 if (job->private_data == private_data) {
 286                         job->canceled = 1;
 287                 }
 288         }
 289 }
 290
 291 int asys_results(struct asys_context *ctx, struct asys_result *results,
 292                  unsigned num_results)
 293 {
 294         int jobids[num_results];
 295         int i, ret;
 296
 297         ret = pthreadpool_finished_jobs(ctx->pool, jobids, num_results);
 298         if (ret <= 0) {
 299                 return ret;
 300         }
 301
 302         for (i=0; i<ret; i++) {
 303                 struct asys_result *result = &results[i];
 304                 struct asys_job *job;
 305                 int jobid;
 306
 307                 jobid = jobids[i];
 308
 309                 if ((jobid < 0) || (jobid >= ctx->num_jobs)) {
 310                         return -EIO;
 311                 }
 312
 313                 job = ctx->jobs[jobid];
 314
 315                 if (job->canceled) {
 316                         result->ret = -1;
 317                         result->err = ECANCELED;
 318                 } else {
 319                         result->ret = job->ret;
 320                         result->err = job->err;
 321                 }
 322                 result->private_data = job->private_data;
 323
 324                 job->busy = 0;
 325         }
 326
 327         return ret;
 328 }