source3/lib/pthreadpool/pthreadpool.c

   1 /*
   2  * Unix SMB/CIFS implementation.
   3  * thread pool implementation
   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 <pthread.h>
  26 #include <signal.h>
  27 #include <assert.h>
  28 #include <fcntl.h>
  29 #include <sys/time.h>
  30
  31 #include "pthreadpool.h"
  32 #include "lib/util/dlinklist.h"
  33
  34 struct pthreadpool_job {
  35         struct pthreadpool_job *next;
  36         int id;
  37         void (*fn)(void *private_data);
  38         void *private_data;
  39 };
  40
  41 struct pthreadpool {
  42         /*
  43          * List pthreadpools for fork safety
  44          */
  45         struct pthreadpool *prev, *next;
  46
  47         /*
  48          * Control access to this struct
  49          */
  50         pthread_mutex_t mutex;
  51
  52         /*
  53          * Threads waiting for work do so here
  54          */
  55         pthread_cond_t condvar;
  56
  57         /*
  58          * List of work jobs
  59          */
  60         struct pthreadpool_job *jobs, *last_job;
  61
  62         /*
  63          * pipe for signalling
  64          */
  65         int sig_pipe[2];
  66
  67         /*
  68          * indicator to worker threads that they should shut down
  69          */
  70         int shutdown;
  71
  72         /*
  73          * maximum number of threads
  74          */
  75         int max_threads;
  76
  77         /*
  78          * Number of threads
  79          */
  80         int num_threads;
  81
  82         /*
  83          * Number of idle threads
  84          */
  85         int num_idle;
  86
  87         /*
  88          * An array of threads that require joining.
  89          */
  90         int                     num_exited;
  91         pthread_t               *exited; /* We alloc more */
  92 };
  93
  94 static pthread_mutex_t pthreadpools_mutex = PTHREAD_MUTEX_INITIALIZER;
  95 static struct pthreadpool *pthreadpools = NULL;
  96 static pthread_once_t pthreadpool_atfork_initialized = PTHREAD_ONCE_INIT;
  97
  98 static void pthreadpool_prep_atfork(void);
  99
 100 /*
 101  * Initialize a thread pool
 102  */
 103
 104 int pthreadpool_init(unsigned max_threads, struct pthreadpool **presult)
 105 {
 106         struct pthreadpool *pool;
 107         int ret;
 108
 109         pool = (struct pthreadpool *)malloc(sizeof(struct pthreadpool));
 110         if (pool == NULL) {
 111                 return ENOMEM;
 112         }
 113
 114         ret = pipe(pool->sig_pipe);
 115         if (ret == -1) {
 116                 int err = errno;
 117                 free(pool);
 118                 return err;
 119         }
 120
 121         ret = pthread_mutex_init(&pool->mutex, NULL);
 122         if (ret != 0) {
 123                 close(pool->sig_pipe[0]);
 124                 close(pool->sig_pipe[1]);
 125                 free(pool);
 126                 return ret;
 127         }
 128
 129         ret = pthread_cond_init(&pool->condvar, NULL);
 130         if (ret != 0) {
 131                 pthread_mutex_destroy(&pool->mutex);
 132                 close(pool->sig_pipe[0]);
 133                 close(pool->sig_pipe[1]);
 134                 free(pool);
 135                 return ret;
 136         }
 137
 138         pool->shutdown = 0;
 139         pool->jobs = pool->last_job = NULL;
 140         pool->num_threads = 0;
 141         pool->num_exited = 0;
 142         pool->exited = NULL;
 143         pool->max_threads = max_threads;
 144         pool->num_idle = 0;
 145
 146         ret = pthread_mutex_lock(&pthreadpools_mutex);
 147         if (ret != 0) {
 148                 pthread_cond_destroy(&pool->condvar);
 149                 pthread_mutex_destroy(&pool->mutex);
 150                 close(pool->sig_pipe[0]);
 151                 close(pool->sig_pipe[1]);
 152                 free(pool);
 153                 return ret;
 154         }
 155         DLIST_ADD(pthreadpools, pool);
 156
 157         ret = pthread_mutex_unlock(&pthreadpools_mutex);
 158         assert(ret == 0);
 159
 160         pthread_once(&pthreadpool_atfork_initialized, pthreadpool_prep_atfork);
 161
 162         *presult = pool;
 163
 164         return 0;
 165 }
 166
 167 static void pthreadpool_prepare(void)
 168 {
 169         int ret;
 170         struct pthreadpool *pool;
 171
 172         ret = pthread_mutex_lock(&pthreadpools_mutex);
 173         assert(ret == 0);
 174
 175         pool = pthreadpools;
 176
 177         while (pool != NULL) {
 178                 ret = pthread_mutex_lock(&pool->mutex);
 179                 assert(ret == 0);
 180                 pool = pool->next;
 181         }
 182 }
 183
 184 static void pthreadpool_parent(void)
 185 {
 186         int ret;
 187         struct pthreadpool *pool;
 188
 189         pool = DLIST_TAIL(pthreadpools);
 190
 191         while (1) {
 192                 ret = pthread_mutex_unlock(&pool->mutex);
 193                 assert(ret == 0);
 194
 195                 if (pool == pthreadpools) {
 196                         break;
 197                 }
 198                 pool = pool->prev;
 199         }
 200
 201         ret = pthread_mutex_unlock(&pthreadpools_mutex);
 202         assert(ret == 0);
 203 }
 204
 205 static void pthreadpool_child(void)
 206 {
 207         int ret;
 208         struct pthreadpool *pool;
 209
 210         pool = DLIST_TAIL(pthreadpools);
 211
 212         while (1) {
 213                 close(pool->sig_pipe[0]);
 214                 close(pool->sig_pipe[1]);
 215
 216                 ret = pipe(pool->sig_pipe);
 217                 assert(ret == 0);
 218
 219                 pool->num_threads = 0;
 220
 221                 pool->num_exited = 0;
 222                 free(pool->exited);
 223                 pool->exited = NULL;
 224
 225                 pool->num_idle = 0;
 226
 227                 while (pool->jobs != NULL) {
 228                         struct pthreadpool_job *job;
 229                         job = pool->jobs;
 230                         pool->jobs = job->next;
 231                         free(job);
 232                 }
 233                 pool->last_job = NULL;
 234
 235                 ret = pthread_mutex_unlock(&pool->mutex);
 236                 assert(ret == 0);
 237
 238                 if (pool == pthreadpools) {
 239                         break;
 240                 }
 241                 pool = pool->prev;
 242         }
 243
 244         ret = pthread_mutex_unlock(&pthreadpools_mutex);
 245         assert(ret == 0);
 246 }
 247
 248 static void pthreadpool_prep_atfork(void)
 249 {
 250         pthread_atfork(pthreadpool_prepare, pthreadpool_parent,
 251                        pthreadpool_child);
 252 }
 253
 254 /*
 255  * Return the file descriptor which becomes readable when a job has
 256  * finished
 257  */
 258
 259 int pthreadpool_signal_fd(struct pthreadpool *pool)
 260 {
 261         return pool->sig_pipe[0];
 262 }
 263
 264 /*
 265  * Do a pthread_join() on all children that have exited, pool->mutex must be
 266  * locked
 267  */
 268 static void pthreadpool_join_children(struct pthreadpool *pool)
 269 {
 270         int i;
 271
 272         for (i=0; i<pool->num_exited; i++) {
 273                 pthread_join(pool->exited[i], NULL);
 274         }
 275         pool->num_exited = 0;
 276
 277         /*
 278          * Deliberately not free and NULL pool->exited. That will be
 279          * re-used by realloc later.
 280          */
 281 }
 282
 283 /*
 284  * Fetch a finished job number from the signal pipe
 285  */
 286
 287 int pthreadpool_finished_job(struct pthreadpool *pool, int *jobid)
 288 {
 289         int ret_jobid;
 290         ssize_t nread;
 291
 292         nread = -1;
 293         errno = EINTR;
 294
 295         while ((nread == -1) && (errno == EINTR)) {
 296                 nread = read(pool->sig_pipe[0], &ret_jobid, sizeof(int));
 297         }
 298         if (nread == -1) {
 299                 return errno;
 300         }
 301         if (nread != sizeof(int)) {
 302                 return EINVAL;
 303         }
 304         *jobid = ret_jobid;
 305         return 0;
 306 }
 307
 308 /*
 309  * Destroy a thread pool, finishing all threads working for it
 310  */
 311
 312 int pthreadpool_destroy(struct pthreadpool *pool)
 313 {
 314         int ret, ret1;
 315
 316         ret = pthread_mutex_lock(&pool->mutex);
 317         if (ret != 0) {
 318                 return ret;
 319         }
 320
 321         if ((pool->jobs != NULL) || pool->shutdown) {
 322                 ret = pthread_mutex_unlock(&pool->mutex);
 323                 assert(ret == 0);
 324                 return EBUSY;
 325         }
 326
 327         if (pool->num_threads > 0) {
 328                 /*
 329                  * We have active threads, tell them to finish, wait for that.
 330                  */
 331
 332                 pool->shutdown = 1;
 333
 334                 if (pool->num_idle > 0) {
 335                         /*
 336                          * Wake the idle threads. They will find pool->quit to
 337                          * be set and exit themselves
 338                          */
 339                         ret = pthread_cond_broadcast(&pool->condvar);
 340                         if (ret != 0) {
 341                                 pthread_mutex_unlock(&pool->mutex);
 342                                 return ret;
 343                         }
 344                 }
 345
 346                 while ((pool->num_threads > 0) || (pool->num_exited > 0)) {
 347
 348                         if (pool->num_exited > 0) {
 349                                 pthreadpool_join_children(pool);
 350                                 continue;
 351                         }
 352                         /*
 353                          * A thread that shuts down will also signal
 354                          * pool->condvar
 355                          */
 356                         ret = pthread_cond_wait(&pool->condvar, &pool->mutex);
 357                         if (ret != 0) {
 358                                 pthread_mutex_unlock(&pool->mutex);
 359                                 return ret;
 360                         }
 361                 }
 362         }
 363
 364         ret = pthread_mutex_unlock(&pool->mutex);
 365         if (ret != 0) {
 366                 return ret;
 367         }
 368         ret = pthread_mutex_destroy(&pool->mutex);
 369         ret1 = pthread_cond_destroy(&pool->condvar);
 370
 371         if (ret != 0) {
 372                 return ret;
 373         }
 374         if (ret1 != 0) {
 375                 return ret1;
 376         }
 377
 378         ret = pthread_mutex_lock(&pthreadpools_mutex);
 379         if (ret != 0) {
 380                 return ret;
 381         }
 382         DLIST_REMOVE(pthreadpools, pool);
 383         ret = pthread_mutex_unlock(&pthreadpools_mutex);
 384         assert(ret == 0);
 385
 386         close(pool->sig_pipe[0]);
 387         pool->sig_pipe[0] = -1;
 388
 389         close(pool->sig_pipe[1]);
 390         pool->sig_pipe[1] = -1;
 391
 392         free(pool->exited);
 393         free(pool);
 394
 395         return 0;
 396 }
 397
 398 /*
 399  * Prepare for pthread_exit(), pool->mutex must be locked
 400  */
 401 static void pthreadpool_server_exit(struct pthreadpool *pool)
 402 {
 403         pthread_t *exited;
 404
 405         pool->num_threads -= 1;
 406
 407         exited = (pthread_t *)realloc(
 408                 pool->exited, sizeof(pthread_t *) * (pool->num_exited + 1));
 409
 410         if (exited == NULL) {
 411                 /* lost a thread status */
 412                 return;
 413         }
 414         pool->exited = exited;
 415
 416         pool->exited[pool->num_exited] = pthread_self();
 417         pool->num_exited += 1;
 418 }
 419
 420 static void *pthreadpool_server(void *arg)
 421 {
 422         struct pthreadpool *pool = (struct pthreadpool *)arg;
 423         int res;
 424
 425         res = pthread_mutex_lock(&pool->mutex);
 426         if (res != 0) {
 427                 return NULL;
 428         }
 429
 430         while (1) {
 431                 struct timeval tv;
 432                 struct timespec ts;
 433                 struct pthreadpool_job *job;
 434
 435                 /*
 436                  * idle-wait at most 1 second. If nothing happens in that
 437                  * time, exit this thread.
 438                  */
 439
 440                 gettimeofday(&tv, NULL);
 441                 ts.tv_sec = tv.tv_sec + 1;
 442                 ts.tv_nsec = tv.tv_usec*1000;
 443
 444                 while ((pool->jobs == NULL) && (pool->shutdown == 0)) {
 445
 446                         pool->num_idle += 1;
 447                         res = pthread_cond_timedwait(
 448                                 &pool->condvar, &pool->mutex, &ts);
 449                         pool->num_idle -= 1;
 450
 451                         if (res == ETIMEDOUT) {
 452
 453                                 if (pool->jobs == NULL) {
 454                                         /*
 455                                          * we timed out and still no work for
 456                                          * us. Exit.
 457                                          */
 458                                         pthreadpool_server_exit(pool);
 459                                         pthread_mutex_unlock(&pool->mutex);
 460                                         return NULL;
 461                                 }
 462
 463                                 break;
 464                         }
 465                         assert(res == 0);
 466                 }
 467
 468                 job = pool->jobs;
 469
 470                 if (job != NULL) {
 471                         ssize_t written;
 472
 473                         /*
 474                          * Ok, there's work for us to do, remove the job from
 475                          * the pthreadpool list
 476                          */
 477                         pool->jobs = job->next;
 478                         if (pool->last_job == job) {
 479                                 pool->last_job = NULL;
 480                         }
 481
 482                         /*
 483                          * Do the work with the mutex unlocked
 484                          */
 485
 486                         res = pthread_mutex_unlock(&pool->mutex);
 487                         assert(res == 0);
 488
 489                         job->fn(job->private_data);
 490
 491                         written = write(pool->sig_pipe[1], &job->id,
 492                                         sizeof(int));
 493
 494                         free(job);
 495
 496                         res = pthread_mutex_lock(&pool->mutex);
 497                         assert(res == 0);
 498
 499                         if (written != sizeof(int)) {
 500                                 pthreadpool_server_exit(pool);
 501                                 pthread_mutex_unlock(&pool->mutex);
 502                                 return NULL;
 503                         }
 504                 }
 505
 506                 if ((pool->jobs == NULL) && (pool->shutdown != 0)) {
 507                         /*
 508                          * No more work to do and we're asked to shut down, so
 509                          * exit
 510                          */
 511                         pthreadpool_server_exit(pool);
 512
 513                         if (pool->num_threads == 0) {
 514                                 /*
 515                                  * Ping the main thread waiting for all of us
 516                                  * workers to have quit.
 517                                  */
 518                                 pthread_cond_broadcast(&pool->condvar);
 519                         }
 520
 521                         pthread_mutex_unlock(&pool->mutex);
 522                         return NULL;
 523                 }
 524         }
 525 }
 526
 527 int pthreadpool_add_job(struct pthreadpool *pool, int job_id,
 528                         void (*fn)(void *private_data), void *private_data)
 529 {
 530         struct pthreadpool_job *job;
 531         pthread_t thread_id;
 532         int res;
 533         sigset_t mask, omask;
 534
 535         job = (struct pthreadpool_job *)malloc(sizeof(struct pthreadpool_job));
 536         if (job == NULL) {
 537                 return ENOMEM;
 538         }
 539
 540         job->fn = fn;
 541         job->private_data = private_data;
 542         job->id = job_id;
 543         job->next = NULL;
 544
 545         res = pthread_mutex_lock(&pool->mutex);
 546         if (res != 0) {
 547                 free(job);
 548                 return res;
 549         }
 550
 551         if (pool->shutdown) {
 552                 /*
 553                  * Protect against the pool being shut down while
 554                  * trying to add a job
 555                  */
 556                 res = pthread_mutex_unlock(&pool->mutex);
 557                 assert(res == 0);
 558                 free(job);
 559                 return EINVAL;
 560         }
 561
 562         /*
 563          * Just some cleanup under the mutex
 564          */
 565         pthreadpool_join_children(pool);
 566
 567         /*
 568          * Add job to the end of the queue
 569          */
 570         if (pool->jobs == NULL) {
 571                 pool->jobs = job;
 572         }
 573         else {
 574                 pool->last_job->next = job;
 575         }
 576         pool->last_job = job;
 577
 578         if (pool->num_idle > 0) {
 579                 /*
 580                  * We have idle threads, wake one.
 581                  */
 582                 res = pthread_cond_signal(&pool->condvar);
 583                 pthread_mutex_unlock(&pool->mutex);
 584                 return res;
 585         }
 586
 587         if ((pool->max_threads != 0) &&
 588             (pool->num_threads >= pool->max_threads)) {
 589                 /*
 590                  * No more new threads, we just queue the request
 591                  */
 592                 pthread_mutex_unlock(&pool->mutex);
 593                 return 0;
 594         }
 595
 596         /*
 597          * Create a new worker thread. It should not receive any signals.
 598          */
 599
 600         sigfillset(&mask);
 601
 602         res = pthread_sigmask(SIG_BLOCK, &mask, &omask);
 603         if (res != 0) {
 604                 pthread_mutex_unlock(&pool->mutex);
 605                 return res;
 606         }
 607
 608         res = pthread_create(&thread_id, NULL, pthreadpool_server,
 609                                 (void *)pool);
 610         if (res == 0) {
 611                 pool->num_threads += 1;
 612         }
 613
 614         assert(pthread_sigmask(SIG_SETMASK, &omask, NULL) == 0);
 615
 616         pthread_mutex_unlock(&pool->mutex);
 617         return res;
 618 }