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