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                 free(pool);
 124                 return ret;
 125         }
 126
 127         ret = pthread_cond_init(&pool->condvar, NULL);
 128         if (ret != 0) {
 129                 pthread_mutex_destroy(&pool->mutex);
 130                 free(pool);
 131                 return ret;
 132         }
 133
 134         pool->shutdown = 0;
 135         pool->jobs = pool->last_job = NULL;
 136         pool->num_threads = 0;
 137         pool->num_exited = 0;
 138         pool->exited = NULL;
 139         pool->max_threads = max_threads;
 140         pool->num_idle = 0;
 141
 142         ret = pthread_mutex_lock(&pthreadpools_mutex);
 143         if (ret != 0) {
 144                 pthread_cond_destroy(&pool->condvar);
 145                 pthread_mutex_destroy(&pool->mutex);
 146                 free(pool);
 147                 return ret;
 148         }
 149         DLIST_ADD(pthreadpools, pool);
 150
 151         ret = pthread_mutex_unlock(&pthreadpools_mutex);
 152         assert(ret == 0);
 153
 154         pthread_once(&pthreadpool_atfork_initialized, pthreadpool_prep_atfork);
 155
 156         *presult = pool;
 157
 158         return 0;
 159 }
 160
 161 static void pthreadpool_prepare(void)
 162 {
 163         int ret;
 164         struct pthreadpool *pool;
 165
 166         ret = pthread_mutex_lock(&pthreadpools_mutex);
 167         assert(ret == 0);
 168
 169         pool = pthreadpools;
 170
 171         while (pool != NULL) {
 172                 ret = pthread_mutex_lock(&pool->mutex);
 173                 assert(ret == 0);
 174                 pool = pool->next;
 175         }
 176 }
 177
 178 static void pthreadpool_parent(void)
 179 {
 180         int ret;
 181         struct pthreadpool *pool;
 182
 183         pool = DLIST_TAIL(pthreadpools);
 184
 185         while (1) {
 186                 ret = pthread_mutex_unlock(&pool->mutex);
 187                 assert(ret == 0);
 188
 189                 if (pool == pthreadpools) {
 190                         break;
 191                 }
 192                 pool = pool->prev;
 193         }
 194
 195         ret = pthread_mutex_unlock(&pthreadpools_mutex);
 196         assert(ret == 0);
 197 }
 198
 199 static void pthreadpool_child(void)
 200 {
 201         int ret;
 202         struct pthreadpool *pool;
 203
 204         pool = DLIST_TAIL(pthreadpools);
 205
 206         while (1) {
 207                 close(pool->sig_pipe[0]);
 208                 close(pool->sig_pipe[1]);
 209
 210                 ret = pipe(pool->sig_pipe);
 211                 assert(ret == 0);
 212
 213                 pool->num_threads = 0;
 214
 215                 pool->num_exited = 0;
 216                 free(pool->exited);
 217                 pool->exited = NULL;
 218
 219                 pool->num_idle = 0;
 220
 221                 while (pool->jobs != NULL) {
 222                         struct pthreadpool_job *job;
 223                         job = pool->jobs;
 224                         pool->jobs = job->next;
 225                         free(job);
 226                 }
 227                 pool->last_job = NULL;
 228
 229                 ret = pthread_mutex_unlock(&pool->mutex);
 230                 assert(ret == 0);
 231
 232                 if (pool == pthreadpools) {
 233                         break;
 234                 }
 235                 pool = pool->prev;
 236         }
 237
 238         ret = pthread_mutex_unlock(&pthreadpools_mutex);
 239         assert(ret == 0);
 240 }
 241
 242 static void pthreadpool_prep_atfork(void)
 243 {
 244         pthread_atfork(pthreadpool_prepare, pthreadpool_parent,
 245                        pthreadpool_child);
 246 }
 247
 248 /*
 249  * Return the file descriptor which becomes readable when a job has
 250  * finished
 251  */
 252
 253 int pthreadpool_signal_fd(struct pthreadpool *pool)
 254 {
 255         return pool->sig_pipe[0];
 256 }
 257
 258 /*
 259  * Do a pthread_join() on all children that have exited, pool->mutex must be
 260  * locked
 261  */
 262 static void pthreadpool_join_children(struct pthreadpool *pool)
 263 {
 264         int i;
 265
 266         for (i=0; i<pool->num_exited; i++) {
 267                 pthread_join(pool->exited[i], NULL);
 268         }
 269         pool->num_exited = 0;
 270
 271         /*
 272          * Deliberately not free and NULL pool->exited. That will be
 273          * re-used by realloc later.
 274          */
 275 }
 276
 277 /*
 278  * Fetch a finished job number from the signal pipe
 279  */
 280
 281 int pthreadpool_finished_job(struct pthreadpool *pool)
 282 {
 283         int result;
 284         ssize_t nread;
 285
 286         nread = -1;
 287         errno = EINTR;
 288
 289         while ((nread == -1) && (errno == EINTR)) {
 290                 nread = read(pool->sig_pipe[0], &result, sizeof(int));
 291         }
 292         if (nread == -1) {
 293                 return errno;
 294         }
 295         if (nread != sizeof(int)) {
 296                 return EINVAL;
 297         }
 298         return result;
 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 pool->quit to
 330                          * 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 timeval tv;
 425                 struct timespec ts;
 426                 struct pthreadpool_job *job;
 427
 428                 /*
 429                  * idle-wait at most 1 second. If nothing happens in that
 430                  * time, exit this thread.
 431                  */
 432
 433                 gettimeofday(&tv, NULL);
 434                 ts.tv_sec = tv.tv_sec + 1;
 435                 ts.tv_nsec = tv.tv_usec*1000;
 436
 437                 while ((pool->jobs == NULL) && (pool->shutdown == 0)) {
 438
 439                         pool->num_idle += 1;
 440                         res = pthread_cond_timedwait(
 441                                 &pool->condvar, &pool->mutex, &ts);
 442                         pool->num_idle -= 1;
 443
 444                         if (res == ETIMEDOUT) {
 445
 446                                 if (pool->jobs == NULL) {
 447                                         /*
 448                                          * we timed out and still no work for
 449                                          * us. Exit.
 450                                          */
 451                                         pthreadpool_server_exit(pool);
 452                                         pthread_mutex_unlock(&pool->mutex);
 453                                         return NULL;
 454                                 }
 455
 456                                 break;
 457                         }
 458                         assert(res == 0);
 459                 }
 460
 461                 job = pool->jobs;
 462
 463                 if (job != NULL) {
 464                         ssize_t written;
 465
 466                         /*
 467                          * Ok, there's work for us to do, remove the job from
 468                          * the pthreadpool list
 469                          */
 470                         pool->jobs = job->next;
 471                         if (pool->last_job == job) {
 472                                 pool->last_job = NULL;
 473                         }
 474
 475                         /*
 476                          * Do the work with the mutex unlocked
 477                          */
 478
 479                         res = pthread_mutex_unlock(&pool->mutex);
 480                         assert(res == 0);
 481
 482                         job->fn(job->private_data);
 483
 484                         written = write(pool->sig_pipe[1], &job->id,
 485                                         sizeof(int));
 486
 487                         free(job);
 488
 489                         res = pthread_mutex_lock(&pool->mutex);
 490                         assert(res == 0);
 491
 492                         if (written != sizeof(int)) {
 493                                 pthreadpool_server_exit(pool);
 494                                 pthread_mutex_unlock(&pool->mutex);
 495                                 return NULL;
 496                         }
 497                 }
 498
 499                 if ((pool->jobs == NULL) && (pool->shutdown != 0)) {
 500                         /*
 501                          * No more work to do and we're asked to shut down, so
 502                          * exit
 503                          */
 504                         pthreadpool_server_exit(pool);
 505
 506                         if (pool->num_threads == 0) {
 507                                 /*
 508                                  * Ping the main thread waiting for all of us
 509                                  * workers to have quit.
 510                                  */
 511                                 pthread_cond_broadcast(&pool->condvar);
 512                         }
 513
 514                         pthread_mutex_unlock(&pool->mutex);
 515                         return NULL;
 516                 }
 517         }
 518 }
 519
 520 int pthreadpool_add_job(struct pthreadpool *pool, int job_id,
 521                         void (*fn)(void *private_data), void *private_data)
 522 {
 523         struct pthreadpool_job *job;
 524         pthread_t thread_id;
 525         int res;
 526         sigset_t mask, omask;
 527
 528         job = (struct pthreadpool_job *)malloc(sizeof(struct pthreadpool_job));
 529         if (job == NULL) {
 530                 return ENOMEM;
 531         }
 532
 533         job->fn = fn;
 534         job->private_data = private_data;
 535         job->id = job_id;
 536         job->next = NULL;
 537
 538         res = pthread_mutex_lock(&pool->mutex);
 539         if (res != 0) {
 540                 free(job);
 541                 return res;
 542         }
 543
 544         if (pool->shutdown) {
 545                 /*
 546                  * Protect against the pool being shut down while
 547                  * trying to add a job
 548                  */
 549                 res = pthread_mutex_unlock(&pool->mutex);
 550                 assert(res == 0);
 551                 free(job);
 552                 return EINVAL;
 553         }
 554
 555         /*
 556          * Just some cleanup under the mutex
 557          */
 558         pthreadpool_join_children(pool);
 559
 560         /*
 561          * Add job to the end of the queue
 562          */
 563         if (pool->jobs == NULL) {
 564                 pool->jobs = job;
 565         }
 566         else {
 567                 pool->last_job->next = job;
 568         }
 569         pool->last_job = job;
 570
 571         if (pool->num_idle > 0) {
 572                 /*
 573                  * We have idle threads, wake one.
 574                  */
 575                 res = pthread_cond_signal(&pool->condvar);
 576                 pthread_mutex_unlock(&pool->mutex);
 577                 return res;
 578         }
 579
 580         if ((pool->max_threads != 0) &&
 581             (pool->num_threads >= pool->max_threads)) {
 582                 /*
 583                  * No more new threads, we just queue the request
 584                  */
 585                 pthread_mutex_unlock(&pool->mutex);
 586                 return 0;
 587         }
 588
 589         /*
 590          * Create a new worker thread. It should not receive any signals.
 591          */
 592
 593         sigfillset(&mask);
 594
 595         res = pthread_sigmask(SIG_BLOCK, &mask, &omask);
 596         if (res != 0) {
 597                 pthread_mutex_unlock(&pool->mutex);
 598                 return res;
 599         }
 600
 601         res = pthread_create(&thread_id, NULL, pthreadpool_server,
 602                                 (void *)pool);
 603         if (res == 0) {
 604                 pool->num_threads += 1;
 605         }
 606
 607         assert(pthread_sigmask(SIG_SETMASK, &omask, NULL) == 0);
 608
 609         pthread_mutex_unlock(&pool->mutex);
 610         return res;
 611 }