source: vendor/current/source3/lib/asys/asys.c

/*
 * Async syscalls
 * Copyright (C) Volker Lendecke 2012
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#include "asys.h"
#include <stdlib.h>
#include <errno.h>
#include "../pthreadpool/pthreadpool.h"

struct asys_pwrite_args {
        int fildes;
        const void *buf;
        size_t nbyte;
        off_t offset;
};

struct asys_pread_args {
        int fildes;
        void *buf;
        size_t nbyte;
        off_t offset;
};

struct asys_fsync_args {
        int fildes;
};

union asys_job_args {
        struct asys_pwrite_args pwrite_args;
        struct asys_pread_args pread_args;
        struct asys_fsync_args fsync_args;
};

struct asys_job {
        void *private_data;
        union asys_job_args args;
        ssize_t ret;
        int err;
        char busy;
        char canceled;
};

struct asys_context {
        struct pthreadpool *pool;
        int pthreadpool_fd;

        unsigned num_jobs;
        struct asys_job **jobs;
};

struct asys_creds_context {
        int dummy;
};

int asys_context_init(struct asys_context **pctx, unsigned max_parallel)
{
        struct asys_context *ctx;
        int ret;

        ctx = calloc(1, sizeof(struct asys_context));
        if (ctx == NULL) {
                return ENOMEM;
        }
        ret = pthreadpool_init(max_parallel, &ctx->pool);
        if (ret != 0) {
                free(ctx);
                return ret;
        }
        ctx->pthreadpool_fd = pthreadpool_signal_fd(ctx->pool);

        *pctx = ctx;
        return 0;
}

int asys_signalfd(struct asys_context *ctx)
{
        return ctx->pthreadpool_fd;
}

int asys_context_destroy(struct asys_context *ctx)
{
        int ret;
        unsigned i;

        for (i=0; i<ctx->num_jobs; i++) {
                if (ctx->jobs[i]->busy) {
                        return EBUSY;
                }
        }

        ret = pthreadpool_destroy(ctx->pool);
        if (ret != 0) {
                return ret;
        }
        for (i=0; i<ctx->num_jobs; i++) {
                free(ctx->jobs[i]);
        }
        free(ctx->jobs);
        free(ctx);
        return 0;
}

static int asys_new_job(struct asys_context *ctx, int *jobid,
                        struct asys_job **pjob)
{
        struct asys_job **tmp;
        struct asys_job *job;
        unsigned i;

        for (i=0; i<ctx->num_jobs; i++) {
                job = ctx->jobs[i];
                if (!job->busy) {
                        job->err = 0;
                        *pjob = job;
                        *jobid = i;
                        return 0;
                }
        }

        if (ctx->num_jobs+1 == 0) {
                return EBUSY;   /* overflow */
        }

        tmp = realloc(ctx->jobs, sizeof(struct asys_job *)*(ctx->num_jobs+1));
        if (tmp == NULL) {
                return ENOMEM;
        }
        ctx->jobs = tmp;

        job = calloc(1, sizeof(struct asys_job));
        if (job == NULL) {
                return ENOMEM;
        }
        ctx->jobs[ctx->num_jobs] = job;

        *jobid = ctx->num_jobs;
        *pjob = job;
        ctx->num_jobs += 1;
        return 0;
}

static void asys_pwrite_do(void *private_data);

int asys_pwrite(struct asys_context *ctx, int fildes, const void *buf,
                size_t nbyte, off_t offset, void *private_data)
{
        struct asys_job *job;
        struct asys_pwrite_args *args;
        int jobid;
        int ret;

        ret = asys_new_job(ctx, &jobid, &job);
        if (ret != 0) {
                return ret;
        }
        job->private_data = private_data;

        args = &job->args.pwrite_args;
        args->fildes = fildes;
        args->buf = buf;
        args->nbyte = nbyte;
        args->offset = offset;

        ret = pthreadpool_add_job(ctx->pool, jobid, asys_pwrite_do, job);
        if (ret != 0) {
                return ret;
        }
        job->busy = 1;

        return 0;
}

static void asys_pwrite_do(void *private_data)
{
        struct asys_job *job = (struct asys_job *)private_data;
        struct asys_pwrite_args *args = &job->args.pwrite_args;

        job->ret = pwrite(args->fildes, args->buf, args->nbyte, args->offset);
        if (job->ret == -1) {
                job->err = errno;
        }
}

static void asys_pread_do(void *private_data);

int asys_pread(struct asys_context *ctx, int fildes, void *buf,
               size_t nbyte, off_t offset, void *private_data)
{
        struct asys_job *job;
        struct asys_pread_args *args;
        int jobid;
        int ret;

        ret = asys_new_job(ctx, &jobid, &job);
        if (ret != 0) {
                return ret;
        }
        job->private_data = private_data;

        args = &job->args.pread_args;
        args->fildes = fildes;
        args->buf = buf;
        args->nbyte = nbyte;
        args->offset = offset;

        ret = pthreadpool_add_job(ctx->pool, jobid, asys_pread_do, job);
        if (ret != 0) {
                return ret;
        }
        job->busy = 1;

        return 0;
}

static void asys_pread_do(void *private_data)
{
        struct asys_job *job = (struct asys_job *)private_data;
        struct asys_pread_args *args = &job->args.pread_args;

        job->ret = pread(args->fildes, args->buf, args->nbyte, args->offset);
        if (job->ret == -1) {
                job->err = errno;
        }
}

static void asys_fsync_do(void *private_data);

int asys_fsync(struct asys_context *ctx, int fildes, void *private_data)
{
        struct asys_job *job;
        struct asys_fsync_args *args;
        int jobid;
        int ret;

        ret = asys_new_job(ctx, &jobid, &job);
        if (ret != 0) {
                return ret;
        }
        job->private_data = private_data;

        args = &job->args.fsync_args;
        args->fildes = fildes;

        ret = pthreadpool_add_job(ctx->pool, jobid, asys_fsync_do, job);
        if (ret != 0) {
                return ret;
        }
        job->busy = 1;

        return 0;
}

static void asys_fsync_do(void *private_data)
{
        struct asys_job *job = (struct asys_job *)private_data;
        struct asys_fsync_args *args = &job->args.fsync_args;

        job->ret = fsync(args->fildes);
        if (job->ret == -1) {
                job->err = errno;
        }
}

void asys_cancel(struct asys_context *ctx, void *private_data)
{
        unsigned i;

        for (i=0; i<ctx->num_jobs; i++) {
                struct asys_job *job = ctx->jobs[i];

                if (job->private_data == private_data) {
                        job->canceled = 1;
                }
        }
}

int asys_results(struct asys_context *ctx, struct asys_result *results,
                 unsigned num_results)
{
        int jobids[num_results];
        int i, ret;

        ret = pthreadpool_finished_jobs(ctx->pool, jobids, num_results);
        if (ret <= 0) {
                return ret;
        }

        for (i=0; i<ret; i++) {
                struct asys_result *result = &results[i];
                struct asys_job *job;
                int jobid;

                jobid = jobids[i];

                if ((jobid < 0) || (jobid >= ctx->num_jobs)) {
                        return -EIO;
                }

                job = ctx->jobs[jobid];

                if (job->canceled) {
                        result->ret = -1;
                        result->err = ECANCELED;
                } else {
                        result->ret = job->ret;
                        result->err = job->err;
                }
                result->private_data = job->private_data;

                job->busy = 0;
        }

        return ret;
}