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, the array has
  89          * "max_threads" elements. It contains "num_exited" ids.
  90          */
  91         int                     num_exited;
  92         pthread_t               exited[1]; /* We alloc more */
  93 };
  94
  95 static pthread_mutex_t pthreadpools_mutex = PTHREAD_MUTEX_INITIALIZER;
  96 static struct pthreadpool *pthreadpools = NULL;
  97 static pthread_once_t pthreadpool_atfork_initialized = PTHREAD_ONCE_INIT;
  98
  99 static void pthreadpool_prep_atfork(void);
 100
 101 /*
 102  * Initialize a thread pool
 103  */
 104
 105 int pthreadpool_init(unsigned max_threads, struct pthreadpool **presult)
 106 {
 107         struct pthreadpool *pool;
 108         size_t size;
 109         int ret;
 110
 111         size = sizeof(struct pthreadpool)
 112                 + (max_threads-1) * sizeof(pthread_t);
 113
 114         pool = (struct pthreadpool *)malloc(size);
 115         if (pool == NULL) {
 116                 return ENOMEM;
 117         }
 118
 119         ret = pipe(pool->sig_pipe);
 120         if (ret == -1) {
 121                 int err = errno;
 122                 free(pool);
 123                 return err;
 124         }
 125
 126         ret = pthread_mutex_init(&pool->mutex, NULL);
 127         if (ret != 0) {
 128                 free(pool);
 129                 return ret;
 130         }
 131
 132         ret = pthread_cond_init(&pool->condvar, NULL);
 133         if (ret != 0) {
 134                 pthread_mutex_destroy(&pool->mutex);
 135                 free(pool);
 136                 return ret;
 137         }
 138
 139         pool->shutdown = 0;
 140         pool->jobs = pool->last_job = NULL;
 141         pool->num_threads = 0;
 142         pool->num_exited = 0;
 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                 free(pool);
 151                 return ret;
 152         }
 153         DLIST_ADD(pthreadpools, pool);
 154
 155         ret = pthread_mutex_unlock(&pthreadpools_mutex);
 156         assert(ret == 0);
 157
 158         pthread_once(&pthreadpool_atfork_initialized, pthreadpool_prep_atfork);
 159
 160         *presult = pool;
 161
 162         return 0;
 163 }
 164
 165 static void pthreadpool_prepare(void)
 166 {
 167         int ret;
 168         struct pthreadpool *pool;
 169
 170         ret = pthread_mutex_lock(&pthreadpools_mutex);
 171         assert(ret == 0);
 172
 173         pool = pthreadpools;
 174
 175         while (pool != NULL) {
 176                 ret = pthread_mutex_lock(&pool->mutex);
 177                 assert(ret == 0);
 178                 pool = pool->next;
 179         }
 180 }
 181
 182 static void pthreadpool_parent(void)
 183 {
 184         int ret;
 185         struct pthreadpool *pool;
 186
 187         pool = DLIST_TAIL(pthreadpools);
 188
 189         while (1) {
 190                 ret = pthread_mutex_unlock(&pool->mutex);
 191                 assert(ret == 0);
 192
 193                 if (pool == pthreadpools) {
 194                         break;
 195                 }
 196                 pool = pool->prev;
 197         }
 198
 199         ret = pthread_mutex_unlock(&pthreadpools_mutex);
 200         assert(ret == 0);
 201 }
 202
 203 static void pthreadpool_child(void)
 204 {
 205         int ret;
 206         struct pthreadpool *pool;
 207
 208         pool = DLIST_TAIL(pthreadpools);
 209
 210         while (1) {
 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                 pool->num_exited = 0;
 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_sig_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 /*
 273  * Fetch a finished job number from the signal pipe
 274  */
 275
 276 int pthreadpool_finished_job(struct pthreadpool *pool)
 277 {
 278         int result;
 279         ssize_t nread;
 280
 281         nread = -1;
 282         errno = EINTR;
 283
 284         while ((nread == -1) && (errno == EINTR)) {
 285                 nread = read(pool->sig_pipe[0], &result, sizeof(int));
 286         }
 287         if (nread == -1) {
 288                 return errno;
 289         }
 290         if (nread != sizeof(int)) {
 291                 return EINVAL;
 292         }
 293         return result;
 294 }
 295
 296 /*
 297  * Destroy a thread pool, finishing all threads working for it
 298  */
 299
 300 int pthreadpool_destroy(struct pthreadpool *pool)
 301 {
 302         int ret, ret1;
 303
 304         ret = pthread_mutex_lock(&pool->mutex);
 305         if (ret != 0) {
 306                 return ret;
 307         }
 308
 309         if ((pool->jobs != NULL) || pool->shutdown) {
 310                 ret = pthread_mutex_unlock(&pool->mutex);
 311                 assert(ret == 0);
 312                 return EBUSY;
 313         }
 314
 315         if (pool->num_threads > 0) {
 316                 /*
 317                  * We have active threads, tell them to finish, wait for that.
 318                  */
 319
 320                 pool->shutdown = 1;
 321
 322                 if (pool->num_idle > 0) {
 323                         /*
 324                          * Wake the idle threads. They will find pool->quit to
 325                          * be set and exit themselves
 326                          */
 327                         ret = pthread_cond_broadcast(&pool->condvar);
 328                         if (ret != 0) {
 329                                 pthread_mutex_unlock(&pool->mutex);
 330                                 return ret;
 331                         }
 332                 }
 333
 334                 while ((pool->num_threads > 0) || (pool->num_exited > 0)) {
 335
 336                         if (pool->num_exited > 0) {
 337                                 pthreadpool_join_children(pool);
 338                                 continue;
 339                         }
 340                         /*
 341                          * A thread that shuts down will also signal
 342                          * pool->condvar
 343                          */
 344                         ret = pthread_cond_wait(&pool->condvar, &pool->mutex);
 345                         if (ret != 0) {
 346                                 pthread_mutex_unlock(&pool->mutex);
 347                                 return ret;
 348                         }
 349                 }
 350         }
 351
 352         ret = pthread_mutex_unlock(&pool->mutex);
 353         if (ret != 0) {
 354                 return ret;
 355         }
 356         ret = pthread_mutex_destroy(&pool->mutex);
 357         ret1 = pthread_cond_destroy(&pool->condvar);
 358
 359         if (ret != 0) {
 360                 return ret;
 361         }
 362         if (ret1 != 0) {
 363                 return ret1;
 364         }
 365
 366         ret = pthread_mutex_lock(&pthreadpools_mutex);
 367         if (ret != 0) {
 368                 return ret;
 369         }
 370         DLIST_REMOVE(pthreadpools, pool);
 371         ret = pthread_mutex_unlock(&pthreadpools_mutex);
 372         assert(ret == 0);
 373
 374         close(pool->sig_pipe[0]);
 375         pool->sig_pipe[0] = -1;
 376
 377         close(pool->sig_pipe[1]);
 378         pool->sig_pipe[1] = -1;
 379
 380         free(pool);
 381
 382         return 0;
 383 }
 384
 385 /*
 386  * Prepare for pthread_exit(), pool->mutex must be locked
 387  */
 388 static void pthreadpool_server_exit(struct pthreadpool *pool)
 389 {
 390         pool->num_threads -= 1;
 391         pool->exited[pool->num_exited] = pthread_self();
 392         pool->num_exited += 1;
 393 }
 394
 395 static void *pthreadpool_server(void *arg)
 396 {
 397         struct pthreadpool *pool = (struct pthreadpool *)arg;
 398         int res;
 399
 400         res = pthread_mutex_lock(&pool->mutex);
 401         if (res != 0) {
 402                 return NULL;
 403         }
 404
 405         while (1) {
 406                 struct timeval tv;
 407                 struct timespec ts;
 408                 struct pthreadpool_job *job;
 409
 410                 /*
 411                  * idle-wait at most 1 second. If nothing happens in that
 412                  * time, exit this thread.
 413                  */
 414
 415                 gettimeofday(&tv, NULL);
 416                 ts.tv_sec = tv.tv_sec + 1;
 417                 ts.tv_nsec = tv.tv_usec*1000;
 418
 419                 while ((pool->jobs == NULL) && (pool->shutdown == 0)) {
 420
 421                         pool->num_idle += 1;
 422                         res = pthread_cond_timedwait(
 423                                 &pool->condvar, &pool->mutex, &ts);
 424                         pool->num_idle -= 1;
 425
 426                         if (res == ETIMEDOUT) {
 427
 428                                 if (pool->jobs == NULL) {
 429                                         /*
 430                                          * we timed out and still no work for
 431                                          * us. Exit.
 432                                          */
 433                                         pthreadpool_server_exit(pool);
 434                                         pthread_mutex_unlock(&pool->mutex);
 435                                         return NULL;
 436                                 }
 437
 438                                 break;
 439                         }
 440                         assert(res == 0);
 441                 }
 442
 443                 job = pool->jobs;
 444
 445                 if (job != NULL) {
 446                         ssize_t written;
 447
 448                         /*
 449                          * Ok, there's work for us to do, remove the job from
 450                          * the pthreadpool list
 451                          */
 452                         pool->jobs = job->next;
 453                         if (pool->last_job == job) {
 454                                 pool->last_job = NULL;
 455                         }
 456
 457                         /*
 458                          * Do the work with the mutex unlocked
 459                          */
 460
 461                         res = pthread_mutex_unlock(&pool->mutex);
 462                         assert(res == 0);
 463
 464                         job->fn(job->private_data);
 465
 466                         written = write(pool->sig_pipe[1], &job->id,
 467                                         sizeof(int));
 468
 469                         free(job);
 470
 471                         res = pthread_mutex_lock(&pool->mutex);
 472                         assert(res == 0);
 473
 474                         if (written != sizeof(int)) {
 475                                 pthreadpool_server_exit(pool);
 476                                 pthread_mutex_unlock(&pool->mutex);
 477                                 return NULL;
 478                         }
 479                 }
 480
 481                 if ((pool->jobs == NULL) && (pool->shutdown != 0)) {
 482                         /*
 483                          * No more work to do and we're asked to shut down, so
 484                          * exit
 485                          */
 486                         pthreadpool_server_exit(pool);
 487
 488                         if (pool->num_threads == 0) {
 489                                 /*
 490                                  * Ping the main thread waiting for all of us
 491                                  * workers to have quit.
 492                                  */
 493                                 pthread_cond_broadcast(&pool->condvar);
 494                         }
 495
 496                         pthread_mutex_unlock(&pool->mutex);
 497                         return NULL;
 498                 }
 499         }
 500 }
 501
 502 int pthreadpool_add_job(struct pthreadpool *pool, int job_id,
 503                         void (*fn)(void *private_data), void *private_data)
 504 {
 505         struct pthreadpool_job *job;
 506         pthread_t thread_id;
 507         int res;
 508         sigset_t mask, omask;
 509
 510         job = (struct pthreadpool_job *)malloc(sizeof(struct pthreadpool_job));
 511         if (job == NULL) {
 512                 return ENOMEM;
 513         }
 514
 515         job->fn = fn;
 516         job->private_data = private_data;
 517         job->id = job_id;
 518         job->next = NULL;
 519
 520         res = pthread_mutex_lock(&pool->mutex);
 521         if (res != 0) {
 522                 free(job);
 523                 return res;
 524         }
 525
 526         if (pool->shutdown) {
 527                 /*
 528                  * Protect against the pool being shut down while
 529                  * trying to add a job
 530                  */
 531                 res = pthread_mutex_unlock(&pool->mutex);
 532                 assert(res == 0);
 533                 free(job);
 534                 return EINVAL;
 535         }
 536
 537         /*
 538          * Just some cleanup under the mutex
 539          */
 540         pthreadpool_join_children(pool);
 541
 542         /*
 543          * Add job to the end of the queue
 544          */
 545         if (pool->jobs == NULL) {
 546                 pool->jobs = job;
 547         }
 548         else {
 549                 pool->last_job->next = job;
 550         }
 551         pool->last_job = job;
 552
 553         if (pool->num_idle > 0) {
 554                 /*
 555                  * We have idle threads, wake one.
 556                  */
 557                 res = pthread_cond_signal(&pool->condvar);
 558                 pthread_mutex_unlock(&pool->mutex);
 559                 return res;
 560         }
 561
 562         if (pool->num_threads >= pool->max_threads) {
 563                 /*
 564                  * No more new threads, we just queue the request
 565                  */
 566                 pthread_mutex_unlock(&pool->mutex);
 567                 return 0;
 568         }
 569
 570         /*
 571          * Create a new worker thread. It should not receive any signals.
 572          */
 573
 574         sigfillset(&mask);
 575
 576         res = pthread_sigmask(SIG_BLOCK, &mask, &omask);
 577         if (res != 0) {
 578                 pthread_mutex_unlock(&pool->mutex);
 579                 return res;
 580         }
 581
 582         res = pthread_create(&thread_id, NULL, pthreadpool_server,
 583                                 (void *)pool);
 584         if (res == 0) {
 585                 pool->num_threads += 1;
 586         }
 587
 588         assert(pthread_sigmask(SIG_SETMASK, &omask, NULL) == 0);
 589
 590         pthread_mutex_unlock(&pool->mutex);
 591         return res;
 592 }