git.sur5r.net Git - openldap/blob - libraries/libldap_r/tpool.c

   1 /* $OpenLDAP$ */
   2 /* This work is part of OpenLDAP Software <http://www.openldap.org/>.
   3  *
   4  * Copyright 1998-2011 The OpenLDAP Foundation.
   5  * All rights reserved.
   6  *
   7  * Redistribution and use in source and binary forms, with or without
   8  * modification, are permitted only as authorized by the OpenLDAP
   9  * Public License.
  10  *
  11  * A copy of this license is available in file LICENSE in the
  12  * top-level directory of the distribution or, alternatively, at
  13  * <http://www.OpenLDAP.org/license.html>.
  14  */
  15
  16 #include "portable.h"
  17
  18 #include <stdio.h>
  19
  20 #include <ac/signal.h>
  21 #include <ac/stdarg.h>
  22 #include <ac/stdlib.h>
  23 #include <ac/string.h>
  24 #include <ac/time.h>
  25 #include <ac/errno.h>
  26
  27 #include "ldap-int.h"
  28 #include "ldap_pvt_thread.h" /* Get the thread interface */
  29 #include "ldap_queue.h"
  30 #define LDAP_THREAD_POOL_IMPLEMENTATION
  31 #include "ldap_thr_debug.h"  /* May rename symbols defined below */
  32
  33 #ifndef LDAP_THREAD_HAVE_TPOOL
  34
  35 /* Thread-specific key with data and optional free function */
  36 typedef struct ldap_int_tpool_key_s {
  37         void *ltk_key;
  38         void *ltk_data;
  39         ldap_pvt_thread_pool_keyfree_t *ltk_free;
  40 } ldap_int_tpool_key_t;
  41
  42 /* Max number of thread-specific keys we store per thread.
  43  * We don't expect to use many...
  44  */
  45 #define MAXKEYS 32
  46
  47 /* Max number of threads */
  48 #define LDAP_MAXTHR     1024    /* must be a power of 2 */
  49
  50 /* (Theoretical) max number of pending requests */
  51 #define MAX_PENDING (INT_MAX/2) /* INT_MAX - (room to avoid overflow) */
  52
  53 /* Context: thread ID and thread-specific key/data pairs */
  54 typedef struct ldap_int_thread_userctx_s {
  55         ldap_pvt_thread_t ltu_id;
  56         ldap_int_tpool_key_t ltu_key[MAXKEYS];
  57 } ldap_int_thread_userctx_t;
  58
  59
  60 /* Simple {thread ID -> context} hash table; key=ctx->ltu_id.
  61  * Protected by ldap_pvt_thread_pool_mutex except during pauses,
  62  * when it is read-only (used by pool_purgekey and pool_context).
  63  * Protected by tpool->ltp_mutex during pauses.
  64  */
  65 static struct {
  66         ldap_int_thread_userctx_t *ctx;
  67         /* ctx is valid when not NULL or DELETED_THREAD_CTX */
  68 #       define DELETED_THREAD_CTX (&ldap_int_main_thrctx + 1) /* dummy addr */
  69 } thread_keys[LDAP_MAXTHR];
  70
  71 #define TID_HASH(tid, hash) do { \
  72         unsigned const char *ptr_ = (unsigned const char *)&(tid); \
  73         unsigned i_; \
  74         for (i_ = 0, (hash) = ptr_[0]; ++i_ < sizeof(tid);) \
  75                 (hash) += ((hash) << 5) ^ ptr_[i_]; \
  76 } while(0)
  77
  78
  79 /* Task for a thread to perform */
  80 typedef struct ldap_int_thread_task_s {
  81         union {
  82                 LDAP_STAILQ_ENTRY(ldap_int_thread_task_s) q;
  83                 LDAP_SLIST_ENTRY(ldap_int_thread_task_s) l;
  84         } ltt_next;
  85         ldap_pvt_thread_start_t *ltt_start_routine;
  86         void *ltt_arg;
  87 } ldap_int_thread_task_t;
  88
  89 typedef LDAP_STAILQ_HEAD(tcq, ldap_int_thread_task_s) ldap_int_tpool_plist_t;
  90
  91 struct ldap_int_thread_pool_s {
  92         LDAP_STAILQ_ENTRY(ldap_int_thread_pool_s) ltp_next;
  93
  94         /* protect members below, and protect thread_keys[] during pauses */
  95         ldap_pvt_thread_mutex_t ltp_mutex;
  96
  97         /* not paused and something to do for pool_<wrapper/pause/destroy>() */
  98         ldap_pvt_thread_cond_t ltp_cond;
  99
 100         /* ltp_active_count <= 1 && ltp_pause */
 101         ldap_pvt_thread_cond_t ltp_pcond;
 102
 103         /* ltp_pause == 0 ? &ltp_pending_list : &empty_pending_list,
 104          * maintaned to reduce work for pool_wrapper()
 105          */
 106         ldap_int_tpool_plist_t *ltp_work_list;
 107
 108         /* pending tasks, and unused task objects */
 109         ldap_int_tpool_plist_t ltp_pending_list;
 110         LDAP_SLIST_HEAD(tcl, ldap_int_thread_task_s) ltp_free_list;
 111
 112         /* The pool is finishing, waiting for its threads to close.
 113          * They close when ltp_pending_list is done.  pool_submit()
 114          * rejects new tasks.  ltp_max_pending = -(its old value).
 115          */
 116         int ltp_finishing;
 117
 118         /* Some active task needs to be the sole active task.
 119          * Atomic variable so ldap_pvt_thread_pool_pausing() can read it.
 120          * Note: Pauses adjust ltp_<open_count/vary_open_count/work_list>,
 121          * so pool_<submit/wrapper>() mostly can avoid testing ltp_pause.
 122          */
 123         volatile sig_atomic_t ltp_pause;
 124
 125         /* Max number of threads in pool, or 0 for default (LDAP_MAXTHR) */
 126         int ltp_max_count;
 127
 128         /* Max number of pending + paused requests, negated when ltp_finishing */
 129         int ltp_max_pending;
 130
 131         int ltp_pending_count;          /* Pending or paused requests */
 132         int ltp_active_count;           /* Active, not paused requests */
 133         int ltp_open_count;                     /* Number of threads, negated when ltp_pause */
 134         int ltp_starting;                       /* Currenlty starting threads */
 135
 136         /* >0 if paused or we may open a thread, <0 if we should close a thread.
 137          * Updated when ltp_<finishing/pause/max_count/open_count> change.
 138          * Maintained to reduce the time ltp_mutex must be locked in
 139          * ldap_pvt_thread_pool_<submit/wrapper>().
 140          */
 141         int ltp_vary_open_count;
 142 #       define SET_VARY_OPEN_COUNT(pool)        \
 143                 ((pool)->ltp_vary_open_count =  \
 144                  (pool)->ltp_pause      ?  1 :  \
 145                  (pool)->ltp_finishing  ? -1 :  \
 146                  ((pool)->ltp_max_count ? (pool)->ltp_max_count : LDAP_MAXTHR) \
 147                  - (pool)->ltp_open_count)
 148 };
 149
 150 static ldap_int_tpool_plist_t empty_pending_list =
 151         LDAP_STAILQ_HEAD_INITIALIZER(empty_pending_list);
 152
 153 static int ldap_int_has_thread_pool = 0;
 154 static LDAP_STAILQ_HEAD(tpq, ldap_int_thread_pool_s)
 155         ldap_int_thread_pool_list =
 156         LDAP_STAILQ_HEAD_INITIALIZER(ldap_int_thread_pool_list);
 157
 158 static ldap_pvt_thread_mutex_t ldap_pvt_thread_pool_mutex;
 159
 160 static void *ldap_int_thread_pool_wrapper( void *pool );
 161
 162 static ldap_pvt_thread_key_t    ldap_tpool_key;
 163
 164 /* Context of the main thread */
 165 static ldap_int_thread_userctx_t ldap_int_main_thrctx;
 166
 167 int
 168 ldap_int_thread_pool_startup ( void )
 169 {
 170         ldap_int_main_thrctx.ltu_id = ldap_pvt_thread_self();
 171         ldap_pvt_thread_key_create( &ldap_tpool_key );
 172         return ldap_pvt_thread_mutex_init(&ldap_pvt_thread_pool_mutex);
 173 }
 174
 175 int
 176 ldap_int_thread_pool_shutdown ( void )
 177 {
 178         struct ldap_int_thread_pool_s *pool;
 179
 180         while ((pool = LDAP_STAILQ_FIRST(&ldap_int_thread_pool_list)) != NULL) {
 181                 (ldap_pvt_thread_pool_destroy)(&pool, 0); /* ignore thr_debug macro */
 182         }
 183         ldap_pvt_thread_mutex_destroy(&ldap_pvt_thread_pool_mutex);
 184         ldap_pvt_thread_key_destroy( ldap_tpool_key );
 185         return(0);
 186 }
 187
 188
 189 /* Create a thread pool */
 190 int
 191 ldap_pvt_thread_pool_init (
 192         ldap_pvt_thread_pool_t *tpool,
 193         int max_threads,
 194         int max_pending )
 195 {
 196         ldap_pvt_thread_pool_t pool;
 197         int rc;
 198
 199         /* multiple pools are currently not supported (ITS#4943) */
 200         assert(!ldap_int_has_thread_pool);
 201
 202         if (! (0 <= max_threads && max_threads <= LDAP_MAXTHR))
 203                 max_threads = 0;
 204         if (! (1 <= max_pending && max_pending <= MAX_PENDING))
 205                 max_pending = MAX_PENDING;
 206
 207         *tpool = NULL;
 208         pool = (ldap_pvt_thread_pool_t) LDAP_CALLOC(1,
 209                 sizeof(struct ldap_int_thread_pool_s));
 210
 211         if (pool == NULL) return(-1);
 212
 213         rc = ldap_pvt_thread_mutex_init(&pool->ltp_mutex);
 214         if (rc != 0)
 215                 return(rc);
 216         rc = ldap_pvt_thread_cond_init(&pool->ltp_cond);
 217         if (rc != 0)
 218                 return(rc);
 219         rc = ldap_pvt_thread_cond_init(&pool->ltp_pcond);
 220         if (rc != 0)
 221                 return(rc);
 222
 223         ldap_int_has_thread_pool = 1;
 224
 225         pool->ltp_max_count = max_threads;
 226         SET_VARY_OPEN_COUNT(pool);
 227         pool->ltp_max_pending = max_pending;
 228
 229         LDAP_STAILQ_INIT(&pool->ltp_pending_list);
 230         pool->ltp_work_list = &pool->ltp_pending_list;
 231         LDAP_SLIST_INIT(&pool->ltp_free_list);
 232
 233         ldap_pvt_thread_mutex_lock(&ldap_pvt_thread_pool_mutex);
 234         LDAP_STAILQ_INSERT_TAIL(&ldap_int_thread_pool_list, pool, ltp_next);
 235         ldap_pvt_thread_mutex_unlock(&ldap_pvt_thread_pool_mutex);
 236
 237         /* Start no threads just yet.  That can break if the process forks
 238          * later, as slapd does in order to daemonize.  On at least POSIX,
 239          * only the forking thread would survive in the child.  Yet fork()
 240          * can't unlock/clean up other threads' locks and data structures,
 241          * unless pthread_atfork() handlers have been set up to do so.
 242          */
 243
 244         *tpool = pool;
 245         return(0);
 246 }
 247
 248
 249 /* Submit a task to be performed by the thread pool */
 250 int
 251 ldap_pvt_thread_pool_submit (
 252         ldap_pvt_thread_pool_t *tpool,
 253         ldap_pvt_thread_start_t *start_routine, void *arg )
 254 {
 255         struct ldap_int_thread_pool_s *pool;
 256         ldap_int_thread_task_t *task;
 257         ldap_pvt_thread_t thr;
 258
 259         if (tpool == NULL)
 260                 return(-1);
 261
 262         pool = *tpool;
 263
 264         if (pool == NULL)
 265                 return(-1);
 266
 267         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 268
 269         if (pool->ltp_pending_count >= pool->ltp_max_pending)
 270                 goto failed;
 271
 272         task = LDAP_SLIST_FIRST(&pool->ltp_free_list);
 273         if (task) {
 274                 LDAP_SLIST_REMOVE_HEAD(&pool->ltp_free_list, ltt_next.l);
 275         } else {
 276                 task = (ldap_int_thread_task_t *) LDAP_MALLOC(sizeof(*task));
 277                 if (task == NULL)
 278                         goto failed;
 279         }
 280
 281         task->ltt_start_routine = start_routine;
 282         task->ltt_arg = arg;
 283
 284         pool->ltp_pending_count++;
 285         LDAP_STAILQ_INSERT_TAIL(&pool->ltp_pending_list, task, ltt_next.q);
 286
 287         /* true if ltp_pause != 0 or we should open (create) a thread */
 288         if (pool->ltp_vary_open_count > 0 &&
 289                 pool->ltp_open_count < pool->ltp_active_count+pool->ltp_pending_count)
 290         {
 291                 if (pool->ltp_pause)
 292                         goto done;
 293
 294                 pool->ltp_starting++;
 295                 pool->ltp_open_count++;
 296                 SET_VARY_OPEN_COUNT(pool);
 297
 298                 if (0 != ldap_pvt_thread_create(
 299                         &thr, 1, ldap_int_thread_pool_wrapper, pool))
 300                 {
 301                         /* couldn't create thread.  back out of
 302                          * ltp_open_count and check for even worse things.
 303                          */
 304                         pool->ltp_starting--;
 305                         pool->ltp_open_count--;
 306                         SET_VARY_OPEN_COUNT(pool);
 307
 308                         if (pool->ltp_open_count == 0) {
 309                                 /* no open threads at all?!?
 310                                  */
 311                                 ldap_int_thread_task_t *ptr;
 312
 313                                 /* let pool_destroy know there are no more threads */
 314                                 ldap_pvt_thread_cond_signal(&pool->ltp_cond);
 315
 316                                 LDAP_STAILQ_FOREACH(ptr, &pool->ltp_pending_list, ltt_next.q)
 317                                         if (ptr == task) break;
 318                                 if (ptr == task) {
 319                                         /* no open threads, task not handled, so
 320                                          * back out of ltp_pending_count, free the task,
 321                                          * report the error.
 322                                          */
 323                                         pool->ltp_pending_count--;
 324                                         LDAP_STAILQ_REMOVE(&pool->ltp_pending_list, task,
 325                                                 ldap_int_thread_task_s, ltt_next.q);
 326                                         LDAP_SLIST_INSERT_HEAD(&pool->ltp_free_list, task,
 327                                                 ltt_next.l);
 328                                         goto failed;
 329                                 }
 330                         }
 331                         /* there is another open thread, so this
 332                          * task will be handled eventually.
 333                          */
 334                 }
 335         }
 336         ldap_pvt_thread_cond_signal(&pool->ltp_cond);
 337
 338  done:
 339         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 340         return(0);
 341
 342  failed:
 343         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 344         return(-1);
 345 }
 346
 347 static void *
 348 no_task( void *ctx, void *arg )
 349 {
 350         return NULL;
 351 }
 352
 353 /* Cancel a pending task that was previously submitted.
 354  * Return 1 if the task was successfully cancelled, 0 if
 355  * not found, -1 for invalid parameters
 356  */
 357 int
 358 ldap_pvt_thread_pool_retract (
 359         ldap_pvt_thread_pool_t *tpool,
 360         ldap_pvt_thread_start_t *start_routine, void *arg )
 361 {
 362         struct ldap_int_thread_pool_s *pool;
 363         ldap_int_thread_task_t *task;
 364
 365         if (tpool == NULL)
 366                 return(-1);
 367
 368         pool = *tpool;
 369
 370         if (pool == NULL)
 371                 return(-1);
 372
 373         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 374         LDAP_STAILQ_FOREACH(task, &pool->ltp_pending_list, ltt_next.q)
 375                 if (task->ltt_start_routine == start_routine &&
 376                         task->ltt_arg == arg) {
 377                         /* Could LDAP_STAILQ_REMOVE the task, but that
 378                          * walks ltp_pending_list again to find it.
 379                          */
 380                         task->ltt_start_routine = no_task;
 381                         task->ltt_arg = NULL;
 382                         break;
 383                 }
 384         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 385         return task != NULL;
 386 }
 387
 388 /* Set max #threads.  value <= 0 means max supported #threads (LDAP_MAXTHR) */
 389 int
 390 ldap_pvt_thread_pool_maxthreads(
 391         ldap_pvt_thread_pool_t *tpool,
 392         int max_threads )
 393 {
 394         struct ldap_int_thread_pool_s *pool;
 395
 396         if (! (0 <= max_threads && max_threads <= LDAP_MAXTHR))
 397                 max_threads = 0;
 398
 399         if (tpool == NULL)
 400                 return(-1);
 401
 402         pool = *tpool;
 403
 404         if (pool == NULL)
 405                 return(-1);
 406
 407         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 408
 409         pool->ltp_max_count = max_threads;
 410         SET_VARY_OPEN_COUNT(pool);
 411
 412         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 413         return(0);
 414 }
 415
 416 /* Inspect the pool */
 417 int
 418 ldap_pvt_thread_pool_query(
 419         ldap_pvt_thread_pool_t *tpool,
 420         ldap_pvt_thread_pool_param_t param,
 421         void *value )
 422 {
 423         struct ldap_int_thread_pool_s   *pool;
 424         int                             count = -1;
 425
 426         if ( tpool == NULL || value == NULL ) {
 427                 return -1;
 428         }
 429
 430         pool = *tpool;
 431
 432         if ( pool == NULL ) {
 433                 return 0;
 434         }
 435
 436         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 437         switch ( param ) {
 438         case LDAP_PVT_THREAD_POOL_PARAM_MAX:
 439                 count = pool->ltp_max_count;
 440                 break;
 441
 442         case LDAP_PVT_THREAD_POOL_PARAM_MAX_PENDING:
 443                 count = pool->ltp_max_pending;
 444                 if (count < 0)
 445                         count = -count;
 446                 if (count == MAX_PENDING)
 447                         count = 0;
 448                 break;
 449
 450         case LDAP_PVT_THREAD_POOL_PARAM_OPEN:
 451                 count = pool->ltp_open_count;
 452                 if (count < 0)
 453                         count = -count;
 454                 break;
 455
 456         case LDAP_PVT_THREAD_POOL_PARAM_STARTING:
 457                 count = pool->ltp_starting;
 458                 break;
 459
 460         case LDAP_PVT_THREAD_POOL_PARAM_ACTIVE:
 461                 count = pool->ltp_active_count;
 462                 break;
 463
 464         case LDAP_PVT_THREAD_POOL_PARAM_PAUSING:
 465                 count = pool->ltp_pause;
 466                 break;
 467
 468         case LDAP_PVT_THREAD_POOL_PARAM_PENDING:
 469                 count = pool->ltp_pending_count;
 470                 break;
 471
 472         case LDAP_PVT_THREAD_POOL_PARAM_BACKLOAD:
 473                 count = pool->ltp_pending_count + pool->ltp_active_count;
 474                 break;
 475
 476         case LDAP_PVT_THREAD_POOL_PARAM_ACTIVE_MAX:
 477                 break;
 478
 479         case LDAP_PVT_THREAD_POOL_PARAM_PENDING_MAX:
 480                 break;
 481
 482         case LDAP_PVT_THREAD_POOL_PARAM_BACKLOAD_MAX:
 483                 break;
 484
 485         case LDAP_PVT_THREAD_POOL_PARAM_STATE:
 486                 *((char **)value) =
 487                         pool->ltp_pause ? "pausing" :
 488                         !pool->ltp_finishing ? "running" :
 489                         pool->ltp_pending_count ? "finishing" : "stopping";
 490                 break;
 491
 492         case LDAP_PVT_THREAD_POOL_PARAM_UNKNOWN:
 493                 break;
 494         }
 495         ldap_pvt_thread_mutex_unlock( &pool->ltp_mutex );
 496
 497         if ( count > -1 ) {
 498                 *((int *)value) = count;
 499         }
 500
 501         return ( count == -1 ? -1 : 0 );
 502 }
 503
 504 /*
 505  * true if pool is pausing; does not lock any mutex to check.
 506  * 0 if not pause, 1 if pause, -1 if error or no pool.
 507  */
 508 int
 509 ldap_pvt_thread_pool_pausing( ldap_pvt_thread_pool_t *tpool )
 510 {
 511         int rc = -1;
 512         struct ldap_int_thread_pool_s *pool;
 513
 514         if ( tpool != NULL && (pool = *tpool) != NULL ) {
 515                 rc = pool->ltp_pause;
 516         }
 517
 518         return rc;
 519 }
 520
 521 /*
 522  * wrapper for ldap_pvt_thread_pool_query(), left around
 523  * for backwards compatibility
 524  */
 525 int
 526 ldap_pvt_thread_pool_backload ( ldap_pvt_thread_pool_t *tpool )
 527 {
 528         int     rc, count;
 529
 530         rc = ldap_pvt_thread_pool_query( tpool,
 531                 LDAP_PVT_THREAD_POOL_PARAM_BACKLOAD, (void *)&count );
 532
 533         if ( rc == 0 ) {
 534                 return count;
 535         }
 536
 537         return rc;
 538 }
 539
 540 /* Destroy the pool after making its threads finish */
 541 int
 542 ldap_pvt_thread_pool_destroy ( ldap_pvt_thread_pool_t *tpool, int run_pending )
 543 {
 544         struct ldap_int_thread_pool_s *pool, *pptr;
 545         ldap_int_thread_task_t *task;
 546
 547         if (tpool == NULL)
 548                 return(-1);
 549
 550         pool = *tpool;
 551
 552         if (pool == NULL) return(-1);
 553
 554         ldap_pvt_thread_mutex_lock(&ldap_pvt_thread_pool_mutex);
 555         LDAP_STAILQ_FOREACH(pptr, &ldap_int_thread_pool_list, ltp_next)
 556                 if (pptr == pool) break;
 557         if (pptr == pool)
 558                 LDAP_STAILQ_REMOVE(&ldap_int_thread_pool_list, pool,
 559                         ldap_int_thread_pool_s, ltp_next);
 560         ldap_pvt_thread_mutex_unlock(&ldap_pvt_thread_pool_mutex);
 561
 562         if (pool != pptr) return(-1);
 563
 564         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 565
 566         pool->ltp_finishing = 1;
 567         SET_VARY_OPEN_COUNT(pool);
 568         if (pool->ltp_max_pending > 0)
 569                 pool->ltp_max_pending = -pool->ltp_max_pending;
 570
 571         if (!run_pending) {
 572                 while ((task = LDAP_STAILQ_FIRST(&pool->ltp_pending_list)) != NULL) {
 573                         LDAP_STAILQ_REMOVE_HEAD(&pool->ltp_pending_list, ltt_next.q);
 574                         LDAP_FREE(task);
 575                 }
 576                 pool->ltp_pending_count = 0;
 577         }
 578
 579         while (pool->ltp_open_count) {
 580                 if (!pool->ltp_pause)
 581                         ldap_pvt_thread_cond_broadcast(&pool->ltp_cond);
 582                 ldap_pvt_thread_cond_wait(&pool->ltp_cond, &pool->ltp_mutex);
 583         }
 584
 585         while ((task = LDAP_SLIST_FIRST(&pool->ltp_free_list)) != NULL)
 586         {
 587                 LDAP_SLIST_REMOVE_HEAD(&pool->ltp_free_list, ltt_next.l);
 588                 LDAP_FREE(task);
 589         }
 590
 591         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 592         ldap_pvt_thread_cond_destroy(&pool->ltp_pcond);
 593         ldap_pvt_thread_cond_destroy(&pool->ltp_cond);
 594         ldap_pvt_thread_mutex_destroy(&pool->ltp_mutex);
 595         LDAP_FREE(pool);
 596         *tpool = NULL;
 597         ldap_int_has_thread_pool = 0;
 598         return(0);
 599 }
 600
 601 /* Thread loop.  Accept and handle submitted tasks. */
 602 static void *
 603 ldap_int_thread_pool_wrapper (
 604         void *xpool )
 605 {
 606         struct ldap_int_thread_pool_s *pool = xpool;
 607         ldap_int_thread_task_t *task;
 608         ldap_int_tpool_plist_t *work_list;
 609         ldap_int_thread_userctx_t ctx, *kctx;
 610         unsigned i, keyslot, hash;
 611
 612         assert(pool != NULL);
 613
 614         for ( i=0; i<MAXKEYS; i++ ) {
 615                 ctx.ltu_key[i].ltk_key = NULL;
 616         }
 617
 618         ctx.ltu_id = ldap_pvt_thread_self();
 619         TID_HASH(ctx.ltu_id, hash);
 620
 621         ldap_pvt_thread_key_setdata( ldap_tpool_key, &ctx );
 622
 623         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 624
 625         /* thread_keys[] is read-only when paused */
 626         while (pool->ltp_pause)
 627                 ldap_pvt_thread_cond_wait(&pool->ltp_cond, &pool->ltp_mutex);
 628
 629         /* find a key slot to give this thread ID and store a
 630          * pointer to our keys there; start at the thread ID
 631          * itself (mod LDAP_MAXTHR) and look for an empty slot.
 632          */
 633         ldap_pvt_thread_mutex_lock(&ldap_pvt_thread_pool_mutex);
 634         for (keyslot = hash & (LDAP_MAXTHR-1);
 635                 (kctx = thread_keys[keyslot].ctx) && kctx != DELETED_THREAD_CTX;
 636                 keyslot = (keyslot+1) & (LDAP_MAXTHR-1));
 637         thread_keys[keyslot].ctx = &ctx;
 638         ldap_pvt_thread_mutex_unlock(&ldap_pvt_thread_pool_mutex);
 639
 640         pool->ltp_starting--;
 641         pool->ltp_active_count++;
 642
 643         for (;;) {
 644                 work_list = pool->ltp_work_list; /* help the compiler a bit */
 645                 task = LDAP_STAILQ_FIRST(work_list);
 646                 if (task == NULL) {     /* paused or no pending tasks */
 647                         if (--(pool->ltp_active_count) < 2) {
 648                                 /* Notify pool_pause it is the sole active thread. */
 649                                 ldap_pvt_thread_cond_signal(&pool->ltp_pcond);
 650                         }
 651
 652                         do {
 653                                 if (pool->ltp_vary_open_count < 0) {
 654                                         /* Not paused, and either finishing or too many
 655                                          * threads running (can happen if ltp_max_count
 656                                          * was reduced).  Let this thread die.
 657                                          */
 658                                         goto done;
 659                                 }
 660
 661                                 /* We could check an idle timer here, and let the
 662                                  * thread die if it has been inactive for a while.
 663                                  * Only die if there are other open threads (i.e.,
 664                                  * always have at least one thread open).
 665                                  * The check should be like this:
 666                                  *   if (pool->ltp_open_count>1 && pool->ltp_starting==0)
 667                                  *       check timer, wait if ltp_pause, leave thread;
 668                                  *
 669                                  * Just use pthread_cond_timedwait() if we want to
 670                                  * check idle time.
 671                                  */
 672                                 ldap_pvt_thread_cond_wait(&pool->ltp_cond, &pool->ltp_mutex);
 673
 674                                 work_list = pool->ltp_work_list;
 675                                 task = LDAP_STAILQ_FIRST(work_list);
 676                         } while (task == NULL);
 677
 678                         pool->ltp_active_count++;
 679                 }
 680
 681                 LDAP_STAILQ_REMOVE_HEAD(work_list, ltt_next.q);
 682                 pool->ltp_pending_count--;
 683                 ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 684
 685                 task->ltt_start_routine(&ctx, task->ltt_arg);
 686
 687                 ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 688                 LDAP_SLIST_INSERT_HEAD(&pool->ltp_free_list, task, ltt_next.l);
 689         }
 690  done:
 691
 692         assert(!pool->ltp_pause); /* thread_keys writable, ltp_open_count >= 0 */
 693
 694         /* The ltp_mutex lock protects ctx->ltu_key from pool_purgekey()
 695          * during this call, since it prevents new pauses. */
 696         ldap_pvt_thread_pool_context_reset(&ctx);
 697
 698         ldap_pvt_thread_mutex_lock(&ldap_pvt_thread_pool_mutex);
 699         thread_keys[keyslot].ctx = DELETED_THREAD_CTX;
 700         ldap_pvt_thread_mutex_unlock(&ldap_pvt_thread_pool_mutex);
 701
 702         pool->ltp_open_count--;
 703         SET_VARY_OPEN_COUNT(pool);
 704         /* let pool_destroy know we're all done */
 705         if (pool->ltp_open_count == 0)
 706                 ldap_pvt_thread_cond_signal(&pool->ltp_cond);
 707
 708         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 709
 710         ldap_pvt_thread_exit(NULL);
 711         return(NULL);
 712 }
 713
 714 #define DO_PAUSE        0
 715 #define CHECK_PAUSE     1
 716 #define GO_IDLE 2
 717 #define GO_UNIDLE       3
 718
 719 static int
 720 handle_pause( ldap_pvt_thread_pool_t *tpool, int pause_type )
 721 {
 722         struct ldap_int_thread_pool_s *pool;
 723         int ret = 0;
 724
 725         if (tpool == NULL)
 726                 return(-1);
 727
 728         pool = *tpool;
 729
 730         if (pool == NULL)
 731                 return(0);
 732
 733         if (pause_type == CHECK_PAUSE && !pool->ltp_pause)
 734                 return(0);
 735
 736         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 737
 738         switch( pause_type ) {
 739         case DO_PAUSE:
 740                 /* Wait for everyone else to pause or finish */
 741                 pool->ltp_pause = 1;
 742                 /* Let ldap_pvt_thread_pool_submit() through to its ltp_pause test,
 743                  * and do not finish threads in ldap_pvt_thread_pool_wrapper() */
 744                 pool->ltp_open_count = -pool->ltp_open_count;
 745                 SET_VARY_OPEN_COUNT(pool);
 746                 /* Hide pending tasks from ldap_pvt_thread_pool_wrapper() */
 747                 pool->ltp_work_list = &empty_pending_list;
 748
 749                 while (pool->ltp_active_count > 1) {
 750                         ldap_pvt_thread_cond_wait(&pool->ltp_pcond, &pool->ltp_mutex);
 751                 }
 752                 break;
 753         case GO_UNIDLE:
 754                 pool->ltp_pending_count--;
 755                 pool->ltp_active_count++;
 756                 /* FALLTHRU */
 757         case CHECK_PAUSE:
 758         case GO_IDLE:
 759         /* If someone else has already requested a pause, we have to wait */
 760                 if (pool->ltp_pause) {
 761                         pool->ltp_pending_count++;
 762                         pool->ltp_active_count--;
 763                         /* let the other pool_pause() know when it can proceed */
 764                         if (pool->ltp_active_count < 2)
 765                                 ldap_pvt_thread_cond_signal(&pool->ltp_pcond);
 766                         do {
 767                                 ldap_pvt_thread_cond_wait(&pool->ltp_cond, &pool->ltp_mutex);
 768                         } while (pool->ltp_pause);
 769                         pool->ltp_pending_count--;
 770                         pool->ltp_active_count++;
 771                         ret = 1;
 772                 }
 773                 if (pause_type != GO_IDLE)
 774                         break;
 775                 pool->ltp_pending_count++;
 776                 pool->ltp_active_count--;
 777                 break;
 778         }
 779
 780         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 781         return(ret);
 782 }
 783
 784 void
 785 ldap_pvt_thread_pool_idle( ldap_pvt_thread_pool_t *tpool )
 786 {
 787         handle_pause( tpool, GO_IDLE );
 788 }
 789
 790 void
 791 ldap_pvt_thread_pool_unidle( ldap_pvt_thread_pool_t *tpool )
 792 {
 793         handle_pause( tpool, GO_UNIDLE );
 794 }
 795
 796 /*
 797  * If a pause was requested, wait for it.  If several threads
 798  * are waiting to pause, let through one or more pauses.
 799  * Return 1 if we waited, 0 if not, -1 at parameter error.
 800  */
 801 int
 802 ldap_pvt_thread_pool_pausecheck( ldap_pvt_thread_pool_t *tpool )
 803 {
 804         return handle_pause( tpool, CHECK_PAUSE );
 805 }
 806
 807 /* Pause the pool.  Return when all other threads are paused. */
 808 int
 809 ldap_pvt_thread_pool_pause( ldap_pvt_thread_pool_t *tpool )
 810 {
 811         return handle_pause( tpool, DO_PAUSE );
 812 }
 813
 814 /* End a pause */
 815 int
 816 ldap_pvt_thread_pool_resume (
 817         ldap_pvt_thread_pool_t *tpool )
 818 {
 819         struct ldap_int_thread_pool_s *pool;
 820
 821         if (tpool == NULL)
 822                 return(-1);
 823
 824         pool = *tpool;
 825
 826         if (pool == NULL)
 827                 return(0);
 828
 829         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 830
 831         assert(pool->ltp_pause);
 832         pool->ltp_pause = 0;
 833         if (pool->ltp_open_count <= 0) /* true when paused, but be paranoid */
 834                 pool->ltp_open_count = -pool->ltp_open_count;
 835         SET_VARY_OPEN_COUNT(pool);
 836         pool->ltp_work_list = &pool->ltp_pending_list;
 837
 838         ldap_pvt_thread_cond_broadcast(&pool->ltp_cond);
 839
 840         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 841         return(0);
 842 }
 843
 844 /*
 845  * Get the key's data and optionally free function in the given context.
 846  */
 847 int ldap_pvt_thread_pool_getkey(
 848         void *xctx,
 849         void *key,
 850         void **data,
 851         ldap_pvt_thread_pool_keyfree_t **kfree )
 852 {
 853         ldap_int_thread_userctx_t *ctx = xctx;
 854         int i;
 855
 856         if ( !ctx || !key || !data ) return EINVAL;
 857
 858         for ( i=0; i<MAXKEYS && ctx->ltu_key[i].ltk_key; i++ ) {
 859                 if ( ctx->ltu_key[i].ltk_key == key ) {
 860                         *data = ctx->ltu_key[i].ltk_data;
 861                         if ( kfree ) *kfree = ctx->ltu_key[i].ltk_free;
 862                         return 0;
 863                 }
 864         }
 865         return ENOENT;
 866 }
 867
 868 static void
 869 clear_key_idx( ldap_int_thread_userctx_t *ctx, int i )
 870 {
 871         for ( ; i < MAXKEYS-1 && ctx->ltu_key[i+1].ltk_key; i++ )
 872                 ctx->ltu_key[i] = ctx->ltu_key[i+1];
 873         ctx->ltu_key[i].ltk_key = NULL;
 874 }
 875
 876 /*
 877  * Set or remove data for the key in the given context.
 878  * key can be any unique pointer.
 879  * kfree() is an optional function to free the data (but not the key):
 880  *   pool_context_reset() and pool_purgekey() call kfree(key, data),
 881  *   but pool_setkey() does not.  For pool_setkey() it is the caller's
 882  *   responsibility to free any existing data with the same key.
 883  *   kfree() must not call functions taking a tpool argument.
 884  */
 885 int ldap_pvt_thread_pool_setkey(
 886         void *xctx,
 887         void *key,
 888         void *data,
 889         ldap_pvt_thread_pool_keyfree_t *kfree,
 890         void **olddatap,
 891         ldap_pvt_thread_pool_keyfree_t **oldkfreep )
 892 {
 893         ldap_int_thread_userctx_t *ctx = xctx;
 894         int i, found;
 895
 896         if ( !ctx || !key ) return EINVAL;
 897
 898         for ( i=found=0; i<MAXKEYS; i++ ) {
 899                 if ( ctx->ltu_key[i].ltk_key == key ) {
 900                         found = 1;
 901                         break;
 902                 } else if ( !ctx->ltu_key[i].ltk_key ) {
 903                         break;
 904                 }
 905         }
 906
 907         if ( olddatap ) {
 908                 if ( found ) {
 909                         *olddatap = ctx->ltu_key[i].ltk_data;
 910                 } else {
 911                         *olddatap = NULL;
 912                 }
 913         }
 914
 915         if ( oldkfreep ) {
 916                 if ( found ) {
 917                         *oldkfreep = ctx->ltu_key[i].ltk_free;
 918                 } else {
 919                         *oldkfreep = 0;
 920                 }
 921         }
 922
 923         if ( data || kfree ) {
 924                 if ( i>=MAXKEYS )
 925                         return ENOMEM;
 926                 ctx->ltu_key[i].ltk_key = key;
 927                 ctx->ltu_key[i].ltk_data = data;
 928                 ctx->ltu_key[i].ltk_free = kfree;
 929         } else if ( found ) {
 930                 clear_key_idx( ctx, i );
 931         }
 932
 933         return 0;
 934 }
 935
 936 /* Free all elements with this key, no matter which thread they're in.
 937  * May only be called while the pool is paused.
 938  */
 939 void ldap_pvt_thread_pool_purgekey( void *key )
 940 {
 941         int i, j;
 942         ldap_int_thread_userctx_t *ctx;
 943
 944         assert ( key != NULL );
 945
 946         for ( i=0; i<LDAP_MAXTHR; i++ ) {
 947                 ctx = thread_keys[i].ctx;
 948                 if ( ctx && ctx != DELETED_THREAD_CTX ) {
 949                         for ( j=0; j<MAXKEYS && ctx->ltu_key[j].ltk_key; j++ ) {
 950                                 if ( ctx->ltu_key[j].ltk_key == key ) {
 951                                         if (ctx->ltu_key[j].ltk_free)
 952                                                 ctx->ltu_key[j].ltk_free( ctx->ltu_key[j].ltk_key,
 953                                                 ctx->ltu_key[j].ltk_data );
 954                                         clear_key_idx( ctx, j );
 955                                         break;
 956                                 }
 957                         }
 958                 }
 959         }
 960 }
 961
 962 /*
 963  * Find the context of the current thread.
 964  * This is necessary if the caller does not have access to the
 965  * thread context handle (for example, a slapd plugin calling
 966  * slapi_search_internal()). No doubt it is more efficient
 967  * for the application to keep track of the thread context
 968  * handles itself.
 969  */
 970 void *ldap_pvt_thread_pool_context( )
 971 {
 972         void *ctx = NULL;
 973
 974         ldap_pvt_thread_key_getdata( ldap_tpool_key, &ctx );
 975         return ctx ? ctx : (void *) &ldap_int_main_thrctx;
 976 }
 977
 978 /*
 979  * Free the context's keys.
 980  * Must not call functions taking a tpool argument (because this
 981  * thread already holds ltp_mutex when called from pool_wrapper()).
 982  */
 983 void ldap_pvt_thread_pool_context_reset( void *vctx )
 984 {
 985         ldap_int_thread_userctx_t *ctx = vctx;
 986         int i;
 987
 988         for ( i=MAXKEYS-1; i>=0; i--) {
 989                 if ( !ctx->ltu_key[i].ltk_key )
 990                         continue;
 991                 if ( ctx->ltu_key[i].ltk_free )
 992                         ctx->ltu_key[i].ltk_free( ctx->ltu_key[i].ltk_key,
 993                         ctx->ltu_key[i].ltk_data );
 994                 ctx->ltu_key[i].ltk_key = NULL;
 995         }
 996 }
 997
 998 ldap_pvt_thread_t ldap_pvt_thread_pool_tid( void *vctx )
 999 {
1000         ldap_int_thread_userctx_t *ctx = vctx;
1001
1002         return ctx->ltu_id;
1003 }
1004 #endif /* LDAP_THREAD_HAVE_TPOOL */