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/wait.h"
  23 #include "system/threads.h"
  24 #include "pthreadpool.h"
  25 #include "lib/util/dlinklist.h"
  26
  27 #ifdef NDEBUG
  28 #undef NDEBUG
  29 #endif
  30
  31 #include <assert.h>
  32
  33 struct pthreadpool_job {
  34         int id;
  35         void (*fn)(void *private_data);
  36         void *private_data;
  37 };
  38
  39 struct pthreadpool {
  40         /*
  41          * List pthreadpools for fork safety
  42          */
  43         struct pthreadpool *prev, *next;
  44
  45         /*
  46          * Control access to this struct
  47          */
  48         pthread_mutex_t mutex;
  49
  50         /*
  51          * Threads waiting for work do so here
  52          */
  53         pthread_cond_t condvar;
  54
  55         /*
  56          * Array of jobs
  57          */
  58         size_t jobs_array_len;
  59         struct pthreadpool_job *jobs;
  60
  61         size_t head;
  62         size_t num_jobs;
  63
  64         /*
  65          * Indicate job completion
  66          */
  67         int (*signal_fn)(int jobid,
  68                          void (*job_fn)(void *private_data),
  69                          void *job_fn_private_data,
  70                          void *private_data);
  71         void *signal_fn_private_data;
  72
  73         /*
  74          * indicator to worker threads that they should shut down
  75          */
  76         bool shutdown;
  77
  78         /*
  79          * maximum number of threads
  80          */
  81         int max_threads;
  82
  83         /*
  84          * Number of threads
  85          */
  86         int num_threads;
  87
  88         /*
  89          * Number of idle threads
  90          */
  91         int num_idle;
  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                      int (*signal_fn)(int jobid,
 106                                       void (*job_fn)(void *private_data),
 107                                       void *job_fn_private_data,
 108                                       void *private_data),
 109                      void *signal_fn_private_data)
 110 {
 111         struct pthreadpool *pool;
 112         int ret;
 113
 114         pool = (struct pthreadpool *)malloc(sizeof(struct pthreadpool));
 115         if (pool == NULL) {
 116                 return ENOMEM;
 117         }
 118         pool->signal_fn = signal_fn;
 119         pool->signal_fn_private_data = signal_fn_private_data;
 120
 121         pool->jobs_array_len = 4;
 122         pool->jobs = calloc(
 123                 pool->jobs_array_len, sizeof(struct pthreadpool_job));
 124
 125         if (pool->jobs == NULL) {
 126                 free(pool);
 127                 return ENOMEM;
 128         }
 129
 130         pool->head = pool->num_jobs = 0;
 131
 132         ret = pthread_mutex_init(&pool->mutex, NULL);
 133         if (ret != 0) {
 134                 free(pool->jobs);
 135                 free(pool);
 136                 return ret;
 137         }
 138
 139         ret = pthread_cond_init(&pool->condvar, NULL);
 140         if (ret != 0) {
 141                 pthread_mutex_destroy(&pool->mutex);
 142                 free(pool->jobs);
 143                 free(pool);
 144                 return ret;
 145         }
 146
 147         pool->shutdown = false;
 148         pool->num_threads = 0;
 149         pool->max_threads = max_threads;
 150         pool->num_idle = 0;
 151
 152         ret = pthread_mutex_lock(&pthreadpools_mutex);
 153         if (ret != 0) {
 154                 pthread_cond_destroy(&pool->condvar);
 155                 pthread_mutex_destroy(&pool->mutex);
 156                 free(pool->jobs);
 157                 free(pool);
 158                 return ret;
 159         }
 160         DLIST_ADD(pthreadpools, pool);
 161
 162         ret = pthread_mutex_unlock(&pthreadpools_mutex);
 163         assert(ret == 0);
 164
 165         pthread_once(&pthreadpool_atfork_initialized, pthreadpool_prep_atfork);
 166
 167         *presult = pool;
 168
 169         return 0;
 170 }
 171
 172 static void pthreadpool_prepare(void)
 173 {
 174         int ret;
 175         struct pthreadpool *pool;
 176
 177         ret = pthread_mutex_lock(&pthreadpools_mutex);
 178         assert(ret == 0);
 179
 180         pool = pthreadpools;
 181
 182         while (pool != NULL) {
 183                 ret = pthread_mutex_lock(&pool->mutex);
 184                 assert(ret == 0);
 185                 pool = pool->next;
 186         }
 187 }
 188
 189 static void pthreadpool_parent(void)
 190 {
 191         int ret;
 192         struct pthreadpool *pool;
 193
 194         for (pool = DLIST_TAIL(pthreadpools);
 195              pool != NULL;
 196              pool = DLIST_PREV(pool)) {
 197                 ret = pthread_mutex_unlock(&pool->mutex);
 198                 assert(ret == 0);
 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         for (pool = DLIST_TAIL(pthreadpools);
 211              pool != NULL;
 212              pool = DLIST_PREV(pool)) {
 213
 214                 pool->num_threads = 0;
 215                 pool->num_idle = 0;
 216                 pool->head = 0;
 217                 pool->num_jobs = 0;
 218
 219                 ret = pthread_mutex_unlock(&pool->mutex);
 220                 assert(ret == 0);
 221         }
 222
 223         ret = pthread_mutex_unlock(&pthreadpools_mutex);
 224         assert(ret == 0);
 225 }
 226
 227 static void pthreadpool_prep_atfork(void)
 228 {
 229         pthread_atfork(pthreadpool_prepare, pthreadpool_parent,
 230                        pthreadpool_child);
 231 }
 232
 233 static int pthreadpool_free(struct pthreadpool *pool)
 234 {
 235         int ret, ret1;
 236
 237         ret = pthread_mutex_unlock(&pool->mutex);
 238         assert(ret == 0);
 239
 240         ret = pthread_mutex_destroy(&pool->mutex);
 241         ret1 = pthread_cond_destroy(&pool->condvar);
 242
 243         if (ret != 0) {
 244                 return ret;
 245         }
 246         if (ret1 != 0) {
 247                 return ret1;
 248         }
 249
 250         ret = pthread_mutex_lock(&pthreadpools_mutex);
 251         if (ret != 0) {
 252                 return ret;
 253         }
 254         DLIST_REMOVE(pthreadpools, pool);
 255         ret = pthread_mutex_unlock(&pthreadpools_mutex);
 256         assert(ret == 0);
 257
 258         free(pool->jobs);
 259         free(pool);
 260
 261         return 0;
 262 }
 263
 264 /*
 265  * Destroy a thread pool. Wake up all idle threads for exit. The last
 266  * one will free the pool.
 267  */
 268
 269 int pthreadpool_destroy(struct pthreadpool *pool)
 270 {
 271         int ret, ret1;
 272
 273         ret = pthread_mutex_lock(&pool->mutex);
 274         if (ret != 0) {
 275                 return ret;
 276         }
 277
 278         if (pool->num_threads == 0) {
 279                 ret = pthreadpool_free(pool);
 280                 return ret;
 281         }
 282
 283         if (pool->shutdown) {
 284                 ret = pthread_mutex_unlock(&pool->mutex);
 285                 assert(ret == 0);
 286                 return EBUSY;
 287         }
 288
 289         /*
 290          * We have active threads, tell them to finish.
 291          */
 292
 293         pool->shutdown = true;
 294
 295         ret = pthread_cond_broadcast(&pool->condvar);
 296
 297         ret1 = pthread_mutex_unlock(&pool->mutex);
 298         assert(ret1 == 0);
 299
 300         return ret;
 301 }
 302
 303 /*
 304  * Prepare for pthread_exit(), pool->mutex must be locked and will be
 305  * unlocked here. This is a bit of a layering violation, but here we
 306  * also take care of removing the pool if we're the last thread.
 307  */
 308 static void pthreadpool_server_exit(struct pthreadpool *pool)
 309 {
 310         int ret;
 311
 312         pool->num_threads -= 1;
 313
 314         if (pool->shutdown && (pool->num_threads == 0)) {
 315                 pthreadpool_free(pool);
 316                 return;
 317         }
 318
 319         ret = pthread_mutex_unlock(&pool->mutex);
 320         assert(ret == 0);
 321 }
 322
 323 static bool pthreadpool_get_job(struct pthreadpool *p,
 324                                 struct pthreadpool_job *job)
 325 {
 326         if (p->num_jobs == 0) {
 327                 return false;
 328         }
 329         *job = p->jobs[p->head];
 330         p->head = (p->head+1) % p->jobs_array_len;
 331         p->num_jobs -= 1;
 332         return true;
 333 }
 334
 335 static bool pthreadpool_put_job(struct pthreadpool *p,
 336                                 int id,
 337                                 void (*fn)(void *private_data),
 338                                 void *private_data)
 339 {
 340         struct pthreadpool_job *job;
 341
 342         if (p->num_jobs == p->jobs_array_len) {
 343                 struct pthreadpool_job *tmp;
 344                 size_t new_len = p->jobs_array_len * 2;
 345
 346                 tmp = realloc(
 347                         p->jobs, sizeof(struct pthreadpool_job) * new_len);
 348                 if (tmp == NULL) {
 349                         return false;
 350                 }
 351                 p->jobs = tmp;
 352
 353                 /*
 354                  * We just doubled the jobs array. The array implements a FIFO
 355                  * queue with a modulo-based wraparound, so we have to memcpy
 356                  * the jobs that are logically at the queue end but physically
 357                  * before the queue head into the reallocated area. The new
 358                  * space starts at the current jobs_array_len, and we have to
 359                  * copy everything before the current head job into the new
 360                  * area.
 361                  */
 362                 memcpy(&p->jobs[p->jobs_array_len], p->jobs,
 363                        sizeof(struct pthreadpool_job) * p->head);
 364
 365                 p->jobs_array_len = new_len;
 366         }
 367
 368         job = &p->jobs[(p->head + p->num_jobs) % p->jobs_array_len];
 369         job->id = id;
 370         job->fn = fn;
 371         job->private_data = private_data;
 372
 373         p->num_jobs += 1;
 374
 375         return true;
 376 }
 377
 378 static void *pthreadpool_server(void *arg)
 379 {
 380         struct pthreadpool *pool = (struct pthreadpool *)arg;
 381         int res;
 382
 383         res = pthread_mutex_lock(&pool->mutex);
 384         if (res != 0) {
 385                 return NULL;
 386         }
 387
 388         while (1) {
 389                 struct timespec ts;
 390                 struct pthreadpool_job job;
 391
 392                 /*
 393                  * idle-wait at most 1 second. If nothing happens in that
 394                  * time, exit this thread.
 395                  */
 396
 397                 clock_gettime(CLOCK_REALTIME, &ts);
 398                 ts.tv_sec += 1;
 399
 400                 while ((pool->num_jobs == 0) && !pool->shutdown) {
 401
 402                         pool->num_idle += 1;
 403                         res = pthread_cond_timedwait(
 404                                 &pool->condvar, &pool->mutex, &ts);
 405                         pool->num_idle -= 1;
 406
 407                         if (res == ETIMEDOUT) {
 408
 409                                 if (pool->num_jobs == 0) {
 410                                         /*
 411                                          * we timed out and still no work for
 412                                          * us. Exit.
 413                                          */
 414                                         pthreadpool_server_exit(pool);
 415                                         return NULL;
 416                                 }
 417
 418                                 break;
 419                         }
 420                         assert(res == 0);
 421                 }
 422
 423                 if (pthreadpool_get_job(pool, &job)) {
 424                         int ret;
 425
 426                         /*
 427                          * Do the work with the mutex unlocked
 428                          */
 429
 430                         res = pthread_mutex_unlock(&pool->mutex);
 431                         assert(res == 0);
 432
 433                         job.fn(job.private_data);
 434
 435                         ret = pool->signal_fn(job.id,
 436                                               job.fn, job.private_data,
 437                                               pool->signal_fn_private_data);
 438
 439                         res = pthread_mutex_lock(&pool->mutex);
 440                         assert(res == 0);
 441
 442                         if (ret != 0) {
 443                                 pthreadpool_server_exit(pool);
 444                                 return NULL;
 445                         }
 446                 }
 447
 448                 if ((pool->num_jobs == 0) && pool->shutdown) {
 449                         /*
 450                          * No more work to do and we're asked to shut down, so
 451                          * exit
 452                          */
 453                         pthreadpool_server_exit(pool);
 454                         return NULL;
 455                 }
 456         }
 457 }
 458
 459 int pthreadpool_add_job(struct pthreadpool *pool, int job_id,
 460                         void (*fn)(void *private_data), void *private_data)
 461 {
 462         pthread_attr_t thread_attr;
 463         pthread_t thread_id;
 464         int res;
 465         sigset_t mask, omask;
 466
 467         res = pthread_mutex_lock(&pool->mutex);
 468         if (res != 0) {
 469                 return res;
 470         }
 471
 472         if (pool->shutdown) {
 473                 /*
 474                  * Protect against the pool being shut down while
 475                  * trying to add a job
 476                  */
 477                 res = pthread_mutex_unlock(&pool->mutex);
 478                 assert(res == 0);
 479                 return EINVAL;
 480         }
 481
 482         /*
 483          * Add job to the end of the queue
 484          */
 485         if (!pthreadpool_put_job(pool, job_id, fn, private_data)) {
 486                 pthread_mutex_unlock(&pool->mutex);
 487                 return ENOMEM;
 488         }
 489
 490         if (pool->num_idle > 0) {
 491                 /*
 492                  * We have idle threads, wake one.
 493                  */
 494                 res = pthread_cond_signal(&pool->condvar);
 495                 pthread_mutex_unlock(&pool->mutex);
 496                 return res;
 497         }
 498
 499         if ((pool->max_threads != 0) &&
 500             (pool->num_threads >= pool->max_threads)) {
 501                 /*
 502                  * No more new threads, we just queue the request
 503                  */
 504                 pthread_mutex_unlock(&pool->mutex);
 505                 return 0;
 506         }
 507
 508         /*
 509          * Create a new worker thread. It should not receive any signals.
 510          */
 511
 512         sigfillset(&mask);
 513
 514         res = pthread_attr_init(&thread_attr);
 515         if (res != 0) {
 516                 pthread_mutex_unlock(&pool->mutex);
 517                 return res;
 518         }
 519
 520         res = pthread_attr_setdetachstate(
 521                 &thread_attr, PTHREAD_CREATE_DETACHED);
 522         if (res != 0) {
 523                 pthread_attr_destroy(&thread_attr);
 524                 pthread_mutex_unlock(&pool->mutex);
 525                 return res;
 526         }
 527
 528         res = pthread_sigmask(SIG_BLOCK, &mask, &omask);
 529         if (res != 0) {
 530                 pthread_attr_destroy(&thread_attr);
 531                 pthread_mutex_unlock(&pool->mutex);
 532                 return res;
 533         }
 534
 535         res = pthread_create(&thread_id, NULL, pthreadpool_server,
 536                              (void *)pool);
 537         if (res == 0) {
 538                 pool->num_threads += 1;
 539         }
 540
 541         assert(pthread_sigmask(SIG_SETMASK, &omask, NULL) == 0);
 542
 543         pthread_attr_destroy(&thread_attr);
 544
 545         pthread_mutex_unlock(&pool->mutex);
 546         return res;
 547 }