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