source3/lib/pthreadpool/pthreadpool_sync.c

   1 /*
   2  * Unix SMB/CIFS implementation.
   3  * sync dummy implementation of the pthreadpool API
   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 <errno.h>
  21 #include <stdio.h>
  22 #include <unistd.h>
  23 #include <stdlib.h>
  24 #include <string.h>
  25 #include <signal.h>
  26 #include <assert.h>
  27 #include <fcntl.h>
  28 #include <sys/time.h>
  29
  30 #include "pthreadpool.h"
  31
  32 struct pthreadpool {
  33         /*
  34          * pipe for signalling
  35          */
  36         int sig_pipe[2];
  37
  38         /*
  39          * Have we sent something into the pipe that has not been
  40          * retrieved yet?
  41          */
  42         int pipe_busy;
  43
  44         /*
  45          * Jobids that we have not sent into the pipe yet
  46          */
  47         size_t num_ids;
  48         int *ids;
  49 };
  50
  51 int pthreadpool_init(unsigned max_threads, struct pthreadpool **presult)
  52 {
  53         struct pthreadpool *pool;
  54         int ret;
  55
  56         pool = (struct pthreadpool *)calloc(1, sizeof(struct pthreadpool));
  57         if (pool == NULL) {
  58                 return ENOMEM;
  59         }
  60         ret = pipe(pool->sig_pipe);
  61         if (ret == -1) {
  62                 int err = errno;
  63                 free(pool);
  64                 return err;
  65         }
  66         *presult = pool;
  67         return 0;
  68 }
  69
  70 int pthreadpool_signal_fd(struct pthreadpool *pool)
  71 {
  72         return pool->sig_pipe[0];
  73 }
  74
  75 static int pthreadpool_write_to_pipe(struct pthreadpool *pool)
  76 {
  77         ssize_t written;
  78
  79         if (pool->pipe_busy) {
  80                 return 0;
  81         }
  82         if (pool->num_ids == 0) {
  83                 return 0;
  84         }
  85
  86         written = -1;
  87         errno = EINTR;
  88
  89         while ((written == -1) && (errno == EINTR)) {
  90                 written = write(pool->sig_pipe[1], &pool->ids[0], sizeof(int));
  91         }
  92         if (written == -1) {
  93                 return errno;
  94         }
  95         if (written != sizeof(int)) {
  96                 /*
  97                  * If a single int only partially fits into the pipe,
  98                  * we can assume ourselves pretty broken
  99                  */
 100                 close(pool->sig_pipe[1]);
 101                 pool->sig_pipe[1] = -1;
 102                 return EIO;
 103         }
 104
 105         if (pool->num_ids > 1) {
 106                 memmove(pool->ids, pool->ids+1, sizeof(int) * (pool->num_ids-1));
 107         }
 108         pool->num_ids -= 1;
 109         pool->pipe_busy = 1;
 110         return 0;
 111 }
 112
 113 int pthreadpool_add_job(struct pthreadpool *pool, int job_id,
 114                         void (*fn)(void *private_data), void *private_data)
 115 {
 116         int *tmp;
 117
 118         if (pool->sig_pipe[1] == -1) {
 119                 return EIO;
 120         }
 121
 122         fn(private_data);
 123
 124         tmp = realloc(pool->ids, sizeof(int) * (pool->num_ids+1));
 125         if (tmp == NULL) {
 126                 return ENOMEM;
 127         }
 128         pool->ids = tmp;
 129         pool->ids[pool->num_ids] = job_id;
 130         pool->num_ids += 1;
 131
 132         return pthreadpool_write_to_pipe(pool);
 133
 134 }
 135
 136 int pthreadpool_finished_job(struct pthreadpool *pool, int *jobid)
 137 {
 138         int ret_jobid;
 139         ssize_t nread;
 140
 141         nread = -1;
 142         errno = EINTR;
 143
 144         while ((nread == -1) && (errno == EINTR)) {
 145                 nread = read(pool->sig_pipe[0], &ret_jobid, sizeof(int));
 146         }
 147         if (nread == -1) {
 148                 return errno;
 149         }
 150         if (nread != sizeof(int)) {
 151                 return EINVAL;
 152         }
 153         *jobid = ret_jobid;
 154
 155         pool->pipe_busy = 0;
 156         return pthreadpool_write_to_pipe(pool);
 157 }
 158
 159 int pthreadpool_destroy(struct pthreadpool *pool)
 160 {
 161         if (pool->sig_pipe[0] != -1) {
 162                 close(pool->sig_pipe[0]);
 163                 pool->sig_pipe[0] = -1;
 164         }
 165
 166         if (pool->sig_pipe[1] != -1) {
 167                 close(pool->sig_pipe[1]);
 168                 pool->sig_pipe[1] = -1;
 169         }
 170         free(pool->ids);
 171         free(pool);
 172         return 0;
 173 }