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_pipe.h"
  23 #include "lib/util/time.h"
  24 #include "smbprofile.h"
  25
  26 struct asys_pwrite_args {
  27         int fildes;
  28         const void *buf;
  29         size_t nbyte;
  30         off_t offset;
  31 };
  32
  33 struct asys_pread_args {
  34         int fildes;
  35         void *buf;
  36         size_t nbyte;
  37         off_t offset;
  38 };
  39
  40 struct asys_fsync_args {
  41         int fildes;
  42 };
  43
  44 union asys_job_args {
  45         struct asys_pwrite_args pwrite_args;
  46         struct asys_pread_args pread_args;
  47         struct asys_fsync_args fsync_args;
  48 };
  49
  50 struct asys_job {
  51         void *private_data;
  52         union asys_job_args args;
  53         ssize_t ret;
  54         int err;
  55         char busy;
  56         char canceled;
  57         struct timespec start_time;
  58         struct timespec end_time;
  59 };
  60
  61 struct asys_context {
  62         struct pthreadpool_pipe *pool;
  63         int pthreadpool_pipe_fd;
  64
  65         unsigned num_jobs;
  66         struct asys_job **jobs;
  67 };
  68
  69 struct asys_creds_context {
  70         int dummy;
  71 };
  72
  73 int asys_context_init(struct asys_context **pctx, unsigned max_parallel)
  74 {
  75         struct asys_context *ctx;
  76         int ret;
  77
  78         ctx = calloc(1, sizeof(struct asys_context));
  79         if (ctx == NULL) {
  80                 return ENOMEM;
  81         }
  82         ret = pthreadpool_pipe_init(max_parallel, &ctx->pool);
  83         if (ret != 0) {
  84                 free(ctx);
  85                 return ret;
  86         }
  87         ctx->pthreadpool_pipe_fd = pthreadpool_pipe_signal_fd(ctx->pool);
  88
  89         *pctx = ctx;
  90         return 0;
  91 }
  92
  93 int asys_signalfd(struct asys_context *ctx)
  94 {
  95         return ctx->pthreadpool_pipe_fd;
  96 }
  97
  98 int asys_context_destroy(struct asys_context *ctx)
  99 {
 100         int ret;
 101         unsigned i;
 102
 103         for (i=0; i<ctx->num_jobs; i++) {
 104                 if (ctx->jobs[i]->busy) {
 105                         return EBUSY;
 106                 }
 107         }
 108
 109         ret = pthreadpool_pipe_destroy(ctx->pool);
 110         if (ret != 0) {
 111                 return ret;
 112         }
 113         for (i=0; i<ctx->num_jobs; i++) {
 114                 free(ctx->jobs[i]);
 115         }
 116         free(ctx->jobs);
 117         free(ctx);
 118         return 0;
 119 }
 120
 121 static int asys_new_job(struct asys_context *ctx, int *jobid,
 122                         struct asys_job **pjob)
 123 {
 124         struct asys_job **tmp;
 125         struct asys_job *job;
 126         unsigned i;
 127
 128         for (i=0; i<ctx->num_jobs; i++) {
 129                 job = ctx->jobs[i];
 130                 if (!job->busy) {
 131                         job->err = 0;
 132                         *pjob = job;
 133                         *jobid = i;
 134                         return 0;
 135                 }
 136         }
 137
 138         if (ctx->num_jobs+1 == 0) {
 139                 return EBUSY;   /* overflow */
 140         }
 141
 142         tmp = realloc(ctx->jobs, sizeof(struct asys_job *)*(ctx->num_jobs+1));
 143         if (tmp == NULL) {
 144                 return ENOMEM;
 145         }
 146         ctx->jobs = tmp;
 147
 148         job = calloc(1, sizeof(struct asys_job));
 149         if (job == NULL) {
 150                 return ENOMEM;
 151         }
 152         ctx->jobs[ctx->num_jobs] = job;
 153
 154         *jobid = ctx->num_jobs;
 155         *pjob = job;
 156         ctx->num_jobs += 1;
 157         return 0;
 158 }
 159
 160 static void asys_pwrite_do(void *private_data);
 161
 162 int asys_pwrite(struct asys_context *ctx, int fildes, const void *buf,
 163                 size_t nbyte, off_t offset, void *private_data)
 164 {
 165         struct asys_job *job;
 166         struct asys_pwrite_args *args;
 167         int jobid;
 168         int ret;
 169
 170         ret = asys_new_job(ctx, &jobid, &job);
 171         if (ret != 0) {
 172                 return ret;
 173         }
 174         job->private_data = private_data;
 175
 176         args = &job->args.pwrite_args;
 177         args->fildes = fildes;
 178         args->buf = buf;
 179         args->nbyte = nbyte;
 180         args->offset = offset;
 181
 182         ret = pthreadpool_pipe_add_job(ctx->pool, jobid, asys_pwrite_do, job);
 183         if (ret != 0) {
 184                 return ret;
 185         }
 186         job->busy = 1;
 187
 188         return 0;
 189 }
 190
 191 static void asys_pwrite_do(void *private_data)
 192 {
 193         struct asys_job *job = (struct asys_job *)private_data;
 194         struct asys_pwrite_args *args = &job->args.pwrite_args;
 195
 196         PROFILE_TIMESTAMP(&job->start_time);
 197         job->ret = pwrite(args->fildes, args->buf, args->nbyte, args->offset);
 198         PROFILE_TIMESTAMP(&job->end_time);
 199
 200         if (job->ret == -1) {
 201                 job->err = errno;
 202         }
 203 }
 204
 205 static void asys_pread_do(void *private_data);
 206
 207 int asys_pread(struct asys_context *ctx, int fildes, void *buf,
 208                size_t nbyte, off_t offset, void *private_data)
 209 {
 210         struct asys_job *job;
 211         struct asys_pread_args *args;
 212         int jobid;
 213         int ret;
 214
 215         ret = asys_new_job(ctx, &jobid, &job);
 216         if (ret != 0) {
 217                 return ret;
 218         }
 219         job->private_data = private_data;
 220
 221         args = &job->args.pread_args;
 222         args->fildes = fildes;
 223         args->buf = buf;
 224         args->nbyte = nbyte;
 225         args->offset = offset;
 226
 227         ret = pthreadpool_pipe_add_job(ctx->pool, jobid, asys_pread_do, job);
 228         if (ret != 0) {
 229                 return ret;
 230         }
 231         job->busy = 1;
 232
 233         return 0;
 234 }
 235
 236 static void asys_pread_do(void *private_data)
 237 {
 238         struct asys_job *job = (struct asys_job *)private_data;
 239         struct asys_pread_args *args = &job->args.pread_args;
 240
 241         PROFILE_TIMESTAMP(&job->start_time);
 242         job->ret = pread(args->fildes, args->buf, args->nbyte, args->offset);
 243         PROFILE_TIMESTAMP(&job->end_time);
 244
 245         if (job->ret == -1) {
 246                 job->err = errno;
 247         }
 248 }
 249
 250 static void asys_fsync_do(void *private_data);
 251
 252 int asys_fsync(struct asys_context *ctx, int fildes, void *private_data)
 253 {
 254         struct asys_job *job;
 255         struct asys_fsync_args *args;
 256         int jobid;
 257         int ret;
 258
 259         ret = asys_new_job(ctx, &jobid, &job);
 260         if (ret != 0) {
 261                 return ret;
 262         }
 263         job->private_data = private_data;
 264
 265         args = &job->args.fsync_args;
 266         args->fildes = fildes;
 267
 268         ret = pthreadpool_pipe_add_job(ctx->pool, jobid, asys_fsync_do, job);
 269         if (ret != 0) {
 270                 return ret;
 271         }
 272         job->busy = 1;
 273
 274         return 0;
 275 }
 276
 277 static void asys_fsync_do(void *private_data)
 278 {
 279         struct asys_job *job = (struct asys_job *)private_data;
 280         struct asys_fsync_args *args = &job->args.fsync_args;
 281
 282         PROFILE_TIMESTAMP(&job->start_time);
 283         job->ret = fsync(args->fildes);
 284         PROFILE_TIMESTAMP(&job->end_time);
 285
 286         if (job->ret == -1) {
 287                 job->err = errno;
 288         }
 289 }
 290
 291 void asys_cancel(struct asys_context *ctx, void *private_data)
 292 {
 293         unsigned i;
 294
 295         for (i=0; i<ctx->num_jobs; i++) {
 296                 struct asys_job *job = ctx->jobs[i];
 297
 298                 if (job->private_data == private_data) {
 299                         job->canceled = 1;
 300                 }
 301         }
 302 }
 303
 304 int asys_results(struct asys_context *ctx, struct asys_result *results,
 305                  unsigned num_results)
 306 {
 307         int jobids[num_results];
 308         int i, ret;
 309
 310         ret = pthreadpool_pipe_finished_jobs(ctx->pool, jobids, num_results);
 311         if (ret <= 0) {
 312                 return ret;
 313         }
 314
 315         for (i=0; i<ret; i++) {
 316                 struct asys_result *result = &results[i];
 317                 struct asys_job *job;
 318                 int jobid;
 319
 320                 jobid = jobids[i];
 321
 322                 if ((jobid < 0) || (jobid >= ctx->num_jobs)) {
 323                         return -EIO;
 324                 }
 325
 326                 job = ctx->jobs[jobid];
 327
 328                 if (job->canceled) {
 329                         result->ret = -1;
 330                         result->err = ECANCELED;
 331                 } else {
 332                         result->ret = job->ret;
 333                         result->err = job->err;
 334                 }
 335                 result->private_data = job->private_data;
 336                 result->duration = nsec_time_diff(&job->end_time, &job->start_time);
 337
 338                 job->busy = 0;
 339         }
 340
 341         return ret;
 342 }