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-2014 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 #ifndef CACHELINE
  36 #define CACHELINE       64
  37 #endif
  38
  39 /* Thread-specific key with data and optional free function */
  40 typedef struct ldap_int_tpool_key_s {
  41         void *ltk_key;
  42         void *ltk_data;
  43         ldap_pvt_thread_pool_keyfree_t *ltk_free;
  44 } ldap_int_tpool_key_t;
  45
  46 /* Max number of thread-specific keys we store per thread.
  47  * We don't expect to use many...
  48  */
  49 #define MAXKEYS 32
  50
  51 /* Max number of threads */
  52 #define LDAP_MAXTHR     1024    /* must be a power of 2 */
  53
  54 /* (Theoretical) max number of pending requests */
  55 #define MAX_PENDING (INT_MAX/2) /* INT_MAX - (room to avoid overflow) */
  56
  57 /* pool->ltp_pause values */
  58 enum { NOT_PAUSED = 0, WANT_PAUSE = 1, PAUSED = 2 };
  59
  60 /* Context: thread ID and thread-specific key/data pairs */
  61 typedef struct ldap_int_thread_userctx_s {
  62         struct ldap_int_thread_poolq_s *ltu_pq;
  63         ldap_pvt_thread_t ltu_id;
  64         ldap_int_tpool_key_t ltu_key[MAXKEYS];
  65 } ldap_int_thread_userctx_t;
  66
  67
  68 /* Simple {thread ID -> context} hash table; key=ctx->ltu_id.
  69  * Protected by ldap_pvt_thread_pool_mutex except during pauses,
  70  * when it is read-only (used by pool_purgekey and pool_context).
  71  * Protected by ldap_pvt_thread_pool_mutex.
  72  */
  73 static struct {
  74         ldap_int_thread_userctx_t *ctx;
  75         /* ctx is valid when not NULL or DELETED_THREAD_CTX */
  76 #       define DELETED_THREAD_CTX (&ldap_int_main_thrctx + 1) /* dummy addr */
  77 } thread_keys[LDAP_MAXTHR];
  78
  79 #define TID_HASH(tid, hash) do { \
  80         unsigned const char *ptr_ = (unsigned const char *)&(tid); \
  81         unsigned i_; \
  82         for (i_ = 0, (hash) = ptr_[0]; ++i_ < sizeof(tid);) \
  83                 (hash) += ((hash) << 5) ^ ptr_[i_]; \
  84 } while(0)
  85
  86
  87 /* Task for a thread to perform */
  88 typedef struct ldap_int_thread_task_s {
  89         union {
  90                 LDAP_STAILQ_ENTRY(ldap_int_thread_task_s) q;
  91                 LDAP_SLIST_ENTRY(ldap_int_thread_task_s) l;
  92         } ltt_next;
  93         ldap_pvt_thread_start_t *ltt_start_routine;
  94         void *ltt_arg;
  95 } ldap_int_thread_task_t;
  96
  97 typedef LDAP_STAILQ_HEAD(tcq, ldap_int_thread_task_s) ldap_int_tpool_plist_t;
  98
  99 struct ldap_int_thread_poolq_s {
 100         void *ltp_free;
 101
 102         struct ldap_int_thread_pool_s *ltp_pool;
 103
 104         /* protect members below */
 105         ldap_pvt_thread_mutex_t ltp_mutex;
 106
 107         /* not paused and something to do for pool_<wrapper/pause/destroy>()
 108          * Used for normal pool operation, to synch between submitter and
 109          * worker threads. Not used for pauses. In normal operation multiple
 110          * queues can rendezvous without acquiring the main pool lock.
 111          */
 112         ldap_pvt_thread_cond_t ltp_cond;
 113
 114         /* ltp_pause == 0 ? &ltp_pending_list : &empty_pending_list,
 115          * maintaned to reduce work for pool_wrapper()
 116          */
 117         ldap_int_tpool_plist_t *ltp_work_list;
 118
 119         /* pending tasks, and unused task objects */
 120         ldap_int_tpool_plist_t ltp_pending_list;
 121         LDAP_SLIST_HEAD(tcl, ldap_int_thread_task_s) ltp_free_list;
 122
 123         /* Max number of threads in this queue */
 124         int ltp_max_count;
 125
 126         /* Max pending + paused + idle tasks, negated when ltp_finishing */
 127         int ltp_max_pending;
 128
 129         int ltp_pending_count;          /* Pending + paused + idle tasks */
 130         int ltp_active_count;           /* Active, not paused/idle tasks */
 131         int ltp_open_count;                     /* Number of threads, negated when ltp_pause */
 132         int ltp_starting;                       /* Currently starting threads */
 133 };
 134
 135 struct ldap_int_thread_pool_s {
 136         LDAP_STAILQ_ENTRY(ldap_int_thread_pool_s) ltp_next;
 137
 138         struct ldap_int_thread_poolq_s **ltp_wqs;
 139
 140         /* number of poolqs */
 141         int ltp_numqs;
 142
 143         /* protect members below */
 144         ldap_pvt_thread_mutex_t ltp_mutex;
 145
 146         /* paused and waiting for resume
 147          * When a pause is in effect all workers switch to waiting on
 148          * this cond instead of their per-queue cond.
 149          */
 150         ldap_pvt_thread_cond_t ltp_cond;
 151
 152         /* ltp_active_queues < 1 && ltp_pause */
 153         ldap_pvt_thread_cond_t ltp_pcond;
 154
 155         /* number of active queues */
 156         int ltp_active_queues;
 157
 158         /* The pool is finishing, waiting for its threads to close.
 159          * They close when ltp_pending_list is done.  pool_submit()
 160          * rejects new tasks.  ltp_max_pending = -(its old value).
 161          */
 162         int ltp_finishing;
 163
 164         /* Some active task needs to be the sole active task.
 165          * Atomic variable so ldap_pvt_thread_pool_pausing() can read it.
 166          */
 167         volatile sig_atomic_t ltp_pause;
 168
 169         /* Max number of threads in pool */
 170         int ltp_max_count;
 171
 172         /* Configured max number of threads in pool, 0 for default (LDAP_MAXTHR) */
 173         int ltp_conf_max_count;
 174
 175         /* Max pending + paused + idle tasks, negated when ltp_finishing */
 176         int ltp_max_pending;
 177 };
 178
 179 static ldap_int_tpool_plist_t empty_pending_list =
 180         LDAP_STAILQ_HEAD_INITIALIZER(empty_pending_list);
 181
 182 static int ldap_int_has_thread_pool = 0;
 183 static LDAP_STAILQ_HEAD(tpq, ldap_int_thread_pool_s)
 184         ldap_int_thread_pool_list =
 185         LDAP_STAILQ_HEAD_INITIALIZER(ldap_int_thread_pool_list);
 186
 187 static ldap_pvt_thread_mutex_t ldap_pvt_thread_pool_mutex;
 188
 189 static void *ldap_int_thread_pool_wrapper( void *pool );
 190
 191 static ldap_pvt_thread_key_t    ldap_tpool_key;
 192
 193 /* Context of the main thread */
 194 static ldap_int_thread_userctx_t ldap_int_main_thrctx;
 195
 196 int
 197 ldap_int_thread_pool_startup ( void )
 198 {
 199         ldap_int_main_thrctx.ltu_id = ldap_pvt_thread_self();
 200         ldap_pvt_thread_key_create( &ldap_tpool_key );
 201         return ldap_pvt_thread_mutex_init(&ldap_pvt_thread_pool_mutex);
 202 }
 203
 204 int
 205 ldap_int_thread_pool_shutdown ( void )
 206 {
 207         struct ldap_int_thread_pool_s *pool;
 208
 209         while ((pool = LDAP_STAILQ_FIRST(&ldap_int_thread_pool_list)) != NULL) {
 210                 (ldap_pvt_thread_pool_destroy)(&pool, 0); /* ignore thr_debug macro */
 211         }
 212         ldap_pvt_thread_mutex_destroy(&ldap_pvt_thread_pool_mutex);
 213         ldap_pvt_thread_key_destroy( ldap_tpool_key );
 214         return(0);
 215 }
 216
 217
 218 /* Create a thread pool */
 219 int
 220 ldap_pvt_thread_pool_init_q (
 221         ldap_pvt_thread_pool_t *tpool,
 222         int max_threads,
 223         int max_pending,
 224         int numqs )
 225 {
 226         ldap_pvt_thread_pool_t pool;
 227         struct ldap_int_thread_poolq_s *pq;
 228         int i, rc, rem_thr, rem_pend;
 229
 230         /* multiple pools are currently not supported (ITS#4943) */
 231         assert(!ldap_int_has_thread_pool);
 232
 233         if (! (0 <= max_threads && max_threads <= LDAP_MAXTHR))
 234                 max_threads = 0;
 235         if (! (1 <= max_pending && max_pending <= MAX_PENDING))
 236                 max_pending = MAX_PENDING;
 237
 238         *tpool = NULL;
 239         pool = (ldap_pvt_thread_pool_t) LDAP_CALLOC(1,
 240                 sizeof(struct ldap_int_thread_pool_s));
 241
 242         if (pool == NULL) return(-1);
 243
 244         pool->ltp_wqs = LDAP_MALLOC(numqs * sizeof(struct ldap_int_thread_poolq_s *));
 245         if (pool->ltp_wqs == NULL) {
 246                 LDAP_FREE(pool);
 247                 return(-1);
 248         }
 249
 250         for (i=0; i<numqs; i++) {
 251                 char *ptr = LDAP_CALLOC(1, sizeof(struct ldap_int_thread_poolq_s) + CACHELINE-1);
 252                 if (ptr == NULL) {
 253                         for (--i; i>=0; i--)
 254                                 LDAP_FREE(pool->ltp_wqs[i]->ltp_free);
 255                         LDAP_FREE(pool->ltp_wqs);
 256                         LDAP_FREE(pool);
 257                         return(-1);
 258                 }
 259                 pool->ltp_wqs[i] = (struct ldap_int_thread_poolq_s *)(((size_t)ptr + CACHELINE-1) & ~(CACHELINE-1));
 260                 pool->ltp_wqs[i]->ltp_free = ptr;
 261         }
 262
 263         pool->ltp_numqs = numqs;
 264         pool->ltp_conf_max_count = max_threads;
 265         if ( !max_threads )
 266                 max_threads = LDAP_MAXTHR;
 267
 268         rc = ldap_pvt_thread_mutex_init(&pool->ltp_mutex);
 269         if (rc != 0) {
 270 fail:
 271                 for (i=0; i<numqs; i++)
 272                         LDAP_FREE(pool->ltp_wqs[i]->ltp_free);
 273                 LDAP_FREE(pool->ltp_wqs);
 274                 LDAP_FREE(pool);
 275                 return(rc);
 276         }
 277
 278         rc = ldap_pvt_thread_cond_init(&pool->ltp_cond);
 279         if (rc != 0)
 280                 goto fail;
 281
 282         rc = ldap_pvt_thread_cond_init(&pool->ltp_pcond);
 283         if (rc != 0)
 284                 goto fail;
 285
 286         rem_thr = max_threads % numqs;
 287         rem_pend = max_pending % numqs;
 288         for ( i=0; i<numqs; i++ ) {
 289                 pq = pool->ltp_wqs[i];
 290                 pq->ltp_pool = pool;
 291                 rc = ldap_pvt_thread_mutex_init(&pq->ltp_mutex);
 292                 if (rc != 0)
 293                         return(rc);
 294                 rc = ldap_pvt_thread_cond_init(&pq->ltp_cond);
 295                 if (rc != 0)
 296                         return(rc);
 297                 LDAP_STAILQ_INIT(&pq->ltp_pending_list);
 298                 pq->ltp_work_list = &pq->ltp_pending_list;
 299                 LDAP_SLIST_INIT(&pq->ltp_free_list);
 300
 301                 pq->ltp_max_count = max_threads / numqs;
 302                 if ( rem_thr ) {
 303                         pq->ltp_max_count++;
 304                         rem_thr--;
 305                 }
 306                 pq->ltp_max_pending = max_pending / numqs;
 307                 if ( rem_pend ) {
 308                         pq->ltp_max_pending++;
 309                         rem_pend--;
 310                 }
 311         }
 312
 313         ldap_int_has_thread_pool = 1;
 314
 315         pool->ltp_max_count = max_threads;
 316         pool->ltp_max_pending = max_pending;
 317
 318         ldap_pvt_thread_mutex_lock(&ldap_pvt_thread_pool_mutex);
 319         LDAP_STAILQ_INSERT_TAIL(&ldap_int_thread_pool_list, pool, ltp_next);
 320         ldap_pvt_thread_mutex_unlock(&ldap_pvt_thread_pool_mutex);
 321
 322         /* Start no threads just yet.  That can break if the process forks
 323          * later, as slapd does in order to daemonize.  On at least POSIX,
 324          * only the forking thread would survive in the child.  Yet fork()
 325          * can't unlock/clean up other threads' locks and data structures,
 326          * unless pthread_atfork() handlers have been set up to do so.
 327          */
 328
 329         *tpool = pool;
 330         return(0);
 331 }
 332
 333 int
 334 ldap_pvt_thread_pool_init (
 335         ldap_pvt_thread_pool_t *tpool,
 336         int max_threads,
 337         int max_pending )
 338 {
 339         return ldap_pvt_thread_pool_init_q( tpool, max_threads, max_pending, 1 );
 340 }
 341
 342 static int
 343 ldap_int_poolq_hash(
 344         struct ldap_int_thread_pool_s *pool,
 345         void *arg )
 346 {
 347         int i = 0, j;
 348         unsigned char *ptr = (unsigned char *)&arg;
 349         /* dumb hash of arg to choose a queue */
 350         for (j=0; j<sizeof(arg); j++)
 351                 i += *ptr++;
 352         i %= pool->ltp_numqs;
 353         return i;
 354 }
 355
 356 /* Submit a task to be performed by the thread pool */
 357 int
 358 ldap_pvt_thread_pool_submit (
 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         struct ldap_int_thread_poolq_s *pq;
 364         ldap_int_thread_task_t *task;
 365         ldap_pvt_thread_t thr;
 366         int i, j;
 367
 368         if (tpool == NULL)
 369                 return(-1);
 370
 371         pool = *tpool;
 372
 373         if (pool == NULL)
 374                 return(-1);
 375
 376         if ( pool->ltp_numqs > 1 )
 377                 i = ldap_int_poolq_hash( pool, arg );
 378         else
 379                 i = 0;
 380
 381         j = i;
 382         while(1) {
 383                 ldap_pvt_thread_mutex_lock(&pool->ltp_wqs[i]->ltp_mutex);
 384                 if (pool->ltp_wqs[i]->ltp_pending_count < pool->ltp_wqs[i]->ltp_max_pending) {
 385                         break;
 386                 }
 387                 ldap_pvt_thread_mutex_unlock(&pool->ltp_wqs[i]->ltp_mutex);
 388                 i++;
 389                 i %= pool->ltp_numqs;
 390                 if ( i == j )
 391                         return -1;
 392         }
 393
 394         pq = pool->ltp_wqs[i];
 395         task = LDAP_SLIST_FIRST(&pq->ltp_free_list);
 396         if (task) {
 397                 LDAP_SLIST_REMOVE_HEAD(&pq->ltp_free_list, ltt_next.l);
 398         } else {
 399                 task = (ldap_int_thread_task_t *) LDAP_MALLOC(sizeof(*task));
 400                 if (task == NULL)
 401                         goto failed;
 402         }
 403
 404         task->ltt_start_routine = start_routine;
 405         task->ltt_arg = arg;
 406
 407         pq->ltp_pending_count++;
 408         LDAP_STAILQ_INSERT_TAIL(&pq->ltp_pending_list, task, ltt_next.q);
 409
 410         if (pool->ltp_pause)
 411                 goto done;
 412
 413         /* should we open (create) a thread? */
 414         if (pq->ltp_open_count < pq->ltp_active_count+pq->ltp_pending_count &&
 415                 pq->ltp_open_count < pq->ltp_max_count)
 416         {
 417                 pq->ltp_starting++;
 418                 pq->ltp_open_count++;
 419
 420                 if (0 != ldap_pvt_thread_create(
 421                         &thr, 1, ldap_int_thread_pool_wrapper, pq))
 422                 {
 423                         /* couldn't create thread.  back out of
 424                          * ltp_open_count and check for even worse things.
 425                          */
 426                         pq->ltp_starting--;
 427                         pq->ltp_open_count--;
 428
 429                         if (pq->ltp_open_count == 0) {
 430                                 /* no open threads at all?!?
 431                                  */
 432                                 ldap_int_thread_task_t *ptr;
 433
 434                                 /* let pool_destroy know there are no more threads */
 435                                 ldap_pvt_thread_cond_signal(&pq->ltp_cond);
 436
 437                                 LDAP_STAILQ_FOREACH(ptr, &pq->ltp_pending_list, ltt_next.q)
 438                                         if (ptr == task) break;
 439                                 if (ptr == task) {
 440                                         /* no open threads, task not handled, so
 441                                          * back out of ltp_pending_count, free the task,
 442                                          * report the error.
 443                                          */
 444                                         pq->ltp_pending_count--;
 445                                         LDAP_STAILQ_REMOVE(&pq->ltp_pending_list, task,
 446                                                 ldap_int_thread_task_s, ltt_next.q);
 447                                         LDAP_SLIST_INSERT_HEAD(&pq->ltp_free_list, task,
 448                                                 ltt_next.l);
 449                                         goto failed;
 450                                 }
 451                         }
 452                         /* there is another open thread, so this
 453                          * task will be handled eventually.
 454                          */
 455                 }
 456         }
 457         ldap_pvt_thread_cond_signal(&pq->ltp_cond);
 458
 459  done:
 460         ldap_pvt_thread_mutex_unlock(&pq->ltp_mutex);
 461         return(0);
 462
 463  failed:
 464         ldap_pvt_thread_mutex_unlock(&pq->ltp_mutex);
 465         return(-1);
 466 }
 467
 468 static void *
 469 no_task( void *ctx, void *arg )
 470 {
 471         return NULL;
 472 }
 473
 474 /* Cancel a pending task that was previously submitted.
 475  * Return 1 if the task was successfully cancelled, 0 if
 476  * not found, -1 for invalid parameters
 477  */
 478 int
 479 ldap_pvt_thread_pool_retract (
 480         ldap_pvt_thread_pool_t *tpool,
 481         ldap_pvt_thread_start_t *start_routine, void *arg )
 482 {
 483         struct ldap_int_thread_pool_s *pool;
 484         struct ldap_int_thread_poolq_s *pq;
 485         ldap_int_thread_task_t *task;
 486         int i;
 487
 488         if (tpool == NULL)
 489                 return(-1);
 490
 491         pool = *tpool;
 492
 493         if (pool == NULL)
 494                 return(-1);
 495
 496         i = ldap_int_poolq_hash( pool, arg );
 497         pq = pool->ltp_wqs[i];
 498
 499         ldap_pvt_thread_mutex_lock(&pq->ltp_mutex);
 500         LDAP_STAILQ_FOREACH(task, &pq->ltp_pending_list, ltt_next.q)
 501                 if (task->ltt_start_routine == start_routine &&
 502                         task->ltt_arg == arg) {
 503                         /* Could LDAP_STAILQ_REMOVE the task, but that
 504                          * walks ltp_pending_list again to find it.
 505                          */
 506                         task->ltt_start_routine = no_task;
 507                         task->ltt_arg = NULL;
 508                         break;
 509                 }
 510         ldap_pvt_thread_mutex_unlock(&pq->ltp_mutex);
 511         return task != NULL;
 512 }
 513
 514 /* Set number of work queues in this pool. Should not be
 515  * more than the number of CPUs. */
 516 int
 517 ldap_pvt_thread_pool_queues(
 518         ldap_pvt_thread_pool_t *tpool,
 519         int numqs )
 520 {
 521         struct ldap_int_thread_pool_s *pool;
 522         struct ldap_int_thread_poolq_s *pq;
 523         int i, rc, rem_thr, rem_pend;
 524
 525         if (numqs < 1 || tpool == NULL)
 526                 return(-1);
 527
 528         pool = *tpool;
 529
 530         if (pool == NULL)
 531                 return(-1);
 532
 533         if (numqs < pool->ltp_numqs) {
 534                 for (i=numqs; i<pool->ltp_numqs; i++)
 535                         pool->ltp_wqs[i]->ltp_max_count = 0;
 536         } else if (numqs > pool->ltp_numqs) {
 537                 struct ldap_int_thread_poolq_s **wqs;
 538                 wqs = LDAP_REALLOC(pool->ltp_wqs, numqs * sizeof(struct ldap_int_thread_poolq_s *));
 539                 if (wqs == NULL)
 540                         return(-1);
 541                 pool->ltp_wqs = wqs;
 542                 for (i=pool->ltp_numqs; i<numqs; i++) {
 543                         char *ptr = LDAP_CALLOC(1, sizeof(struct ldap_int_thread_poolq_s) + CACHELINE-1);
 544                         if (ptr == NULL) {
 545                                 for (; i<numqs; i++)
 546                                         pool->ltp_wqs[i] = NULL;
 547                                 return(-1);
 548                         }
 549                         pq = (struct ldap_int_thread_poolq_s *)(((size_t)ptr + CACHELINE-1) & ~(CACHELINE-1));
 550                         pq->ltp_free = ptr;
 551                         pool->ltp_wqs[i] = pq;
 552                         pq->ltp_pool = pool;
 553                         rc = ldap_pvt_thread_mutex_init(&pq->ltp_mutex);
 554                         if (rc != 0)
 555                                 return(rc);
 556                         rc = ldap_pvt_thread_cond_init(&pq->ltp_cond);
 557                         if (rc != 0)
 558                                 return(rc);
 559                         LDAP_STAILQ_INIT(&pq->ltp_pending_list);
 560                         pq->ltp_work_list = &pq->ltp_pending_list;
 561                         LDAP_SLIST_INIT(&pq->ltp_free_list);
 562                 }
 563         }
 564         rem_thr = pool->ltp_max_count % numqs;
 565         rem_pend = pool->ltp_max_pending % numqs;
 566         for ( i=0; i<numqs; i++ ) {
 567                 pq = pool->ltp_wqs[i];
 568                 pq->ltp_max_count = pool->ltp_max_count / numqs;
 569                 if ( rem_thr ) {
 570                         pq->ltp_max_count++;
 571                         rem_thr--;
 572                 }
 573                 pq->ltp_max_pending = pool->ltp_max_pending / numqs;
 574                 if ( rem_pend ) {
 575                         pq->ltp_max_pending++;
 576                         rem_pend--;
 577                 }
 578         }
 579         pool->ltp_numqs = numqs;
 580         return 0;
 581 }
 582
 583 /* Set max #threads.  value <= 0 means max supported #threads (LDAP_MAXTHR) */
 584 int
 585 ldap_pvt_thread_pool_maxthreads(
 586         ldap_pvt_thread_pool_t *tpool,
 587         int max_threads )
 588 {
 589         struct ldap_int_thread_pool_s *pool;
 590         struct ldap_int_thread_poolq_s *pq;
 591         int remthr, i;
 592
 593         if (! (0 <= max_threads && max_threads <= LDAP_MAXTHR))
 594                 max_threads = 0;
 595
 596         if (tpool == NULL)
 597                 return(-1);
 598
 599         pool = *tpool;
 600
 601         if (pool == NULL)
 602                 return(-1);
 603
 604         pool->ltp_conf_max_count = max_threads;
 605         if ( !max_threads )
 606                 max_threads = LDAP_MAXTHR;
 607         pool->ltp_max_count = max_threads;
 608
 609         remthr = max_threads % pool->ltp_numqs;
 610         max_threads /= pool->ltp_numqs;
 611
 612         for (i=0; i<pool->ltp_numqs; i++) {
 613                 pq = pool->ltp_wqs[i];
 614                 pq->ltp_max_count = max_threads;
 615                 if (remthr) {
 616                         pq->ltp_max_count++;
 617                         remthr--;
 618                 }
 619         }
 620         return(0);
 621 }
 622
 623 /* Inspect the pool */
 624 int
 625 ldap_pvt_thread_pool_query(
 626         ldap_pvt_thread_pool_t *tpool,
 627         ldap_pvt_thread_pool_param_t param,
 628         void *value )
 629 {
 630         struct ldap_int_thread_pool_s   *pool;
 631         int                             count = -1;
 632
 633         if ( tpool == NULL || value == NULL ) {
 634                 return -1;
 635         }
 636
 637         pool = *tpool;
 638
 639         if ( pool == NULL ) {
 640                 return 0;
 641         }
 642
 643         switch ( param ) {
 644         case LDAP_PVT_THREAD_POOL_PARAM_MAX:
 645                 count = pool->ltp_conf_max_count;
 646                 break;
 647
 648         case LDAP_PVT_THREAD_POOL_PARAM_MAX_PENDING:
 649                 count = pool->ltp_max_pending;
 650                 if (count < 0)
 651                         count = -count;
 652                 if (count == MAX_PENDING)
 653                         count = 0;
 654                 break;
 655
 656         case LDAP_PVT_THREAD_POOL_PARAM_PAUSING:
 657                 ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 658                 count = (pool->ltp_pause != 0);
 659                 ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 660                 break;
 661
 662         case LDAP_PVT_THREAD_POOL_PARAM_OPEN:
 663         case LDAP_PVT_THREAD_POOL_PARAM_STARTING:
 664         case LDAP_PVT_THREAD_POOL_PARAM_ACTIVE:
 665         case LDAP_PVT_THREAD_POOL_PARAM_PENDING:
 666         case LDAP_PVT_THREAD_POOL_PARAM_BACKLOAD:
 667                 {
 668                         int i;
 669                         count = 0;
 670                         for (i=0; i<pool->ltp_numqs; i++) {
 671                                 struct ldap_int_thread_poolq_s *pq = pool->ltp_wqs[i];
 672                                 ldap_pvt_thread_mutex_lock(&pq->ltp_mutex);
 673                                 switch(param) {
 674                                         case LDAP_PVT_THREAD_POOL_PARAM_OPEN:
 675                                                 count += pq->ltp_open_count;
 676                                                 break;
 677                                         case LDAP_PVT_THREAD_POOL_PARAM_STARTING:
 678                                                 count += pq->ltp_starting;
 679                                                 break;
 680                                         case LDAP_PVT_THREAD_POOL_PARAM_ACTIVE:
 681                                                 count += pq->ltp_active_count;
 682                                                 break;
 683                                         case LDAP_PVT_THREAD_POOL_PARAM_PENDING:
 684                                                 count += pq->ltp_pending_count;
 685                                                 break;
 686                                         case LDAP_PVT_THREAD_POOL_PARAM_BACKLOAD:
 687                                                 count += pq->ltp_pending_count + pq->ltp_active_count;
 688                                                 break;
 689                                 }
 690                                 ldap_pvt_thread_mutex_unlock(&pq->ltp_mutex);
 691                         }
 692                         if (count < 0)
 693                                 count = -count;
 694                 }
 695                 break;
 696
 697         case LDAP_PVT_THREAD_POOL_PARAM_ACTIVE_MAX:
 698                 break;
 699
 700         case LDAP_PVT_THREAD_POOL_PARAM_PENDING_MAX:
 701                 break;
 702
 703         case LDAP_PVT_THREAD_POOL_PARAM_BACKLOAD_MAX:
 704                 break;
 705
 706         case LDAP_PVT_THREAD_POOL_PARAM_STATE:
 707                 if (pool->ltp_pause)
 708                         *((char **)value) = "pausing";
 709                 else if (!pool->ltp_finishing)
 710                         *((char **)value) = "running";
 711                 else {
 712                         int i;
 713                         for (i=0; i<pool->ltp_numqs; i++)
 714                                 if (pool->ltp_wqs[i]->ltp_pending_count) break;
 715                         if (i<pool->ltp_numqs)
 716                                 *((char **)value) = "finishing";
 717                         else
 718                                 *((char **)value) = "stopping";
 719                 }
 720                 break;
 721
 722         case LDAP_PVT_THREAD_POOL_PARAM_UNKNOWN:
 723                 break;
 724         }
 725
 726         if ( count > -1 ) {
 727                 *((int *)value) = count;
 728         }
 729
 730         return ( count == -1 ? -1 : 0 );
 731 }
 732
 733 /*
 734  * true if pool is pausing; does not lock any mutex to check.
 735  * 0 if not pause, 1 if pause, -1 if error or no pool.
 736  */
 737 int
 738 ldap_pvt_thread_pool_pausing( ldap_pvt_thread_pool_t *tpool )
 739 {
 740         int rc = -1;
 741         struct ldap_int_thread_pool_s *pool;
 742
 743         if ( tpool != NULL && (pool = *tpool) != NULL ) {
 744                 rc = (pool->ltp_pause != 0);
 745         }
 746
 747         return rc;
 748 }
 749
 750 /*
 751  * wrapper for ldap_pvt_thread_pool_query(), left around
 752  * for backwards compatibility
 753  */
 754 int
 755 ldap_pvt_thread_pool_backload ( ldap_pvt_thread_pool_t *tpool )
 756 {
 757         int     rc, count;
 758
 759         rc = ldap_pvt_thread_pool_query( tpool,
 760                 LDAP_PVT_THREAD_POOL_PARAM_BACKLOAD, (void *)&count );
 761
 762         if ( rc == 0 ) {
 763                 return count;
 764         }
 765
 766         return rc;
 767 }
 768
 769 /* Destroy the pool after making its threads finish */
 770 int
 771 ldap_pvt_thread_pool_destroy ( ldap_pvt_thread_pool_t *tpool, int run_pending )
 772 {
 773         struct ldap_int_thread_pool_s *pool, *pptr;
 774         struct ldap_int_thread_poolq_s *pq;
 775         ldap_int_thread_task_t *task;
 776         int i;
 777
 778         if (tpool == NULL)
 779                 return(-1);
 780
 781         pool = *tpool;
 782
 783         if (pool == NULL) return(-1);
 784
 785         ldap_pvt_thread_mutex_lock(&ldap_pvt_thread_pool_mutex);
 786         LDAP_STAILQ_FOREACH(pptr, &ldap_int_thread_pool_list, ltp_next)
 787                 if (pptr == pool) break;
 788         if (pptr == pool)
 789                 LDAP_STAILQ_REMOVE(&ldap_int_thread_pool_list, pool,
 790                         ldap_int_thread_pool_s, ltp_next);
 791         ldap_pvt_thread_mutex_unlock(&ldap_pvt_thread_pool_mutex);
 792
 793         if (pool != pptr) return(-1);
 794
 795         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 796
 797         pool->ltp_finishing = 1;
 798         if (pool->ltp_max_pending > 0)
 799                 pool->ltp_max_pending = -pool->ltp_max_pending;
 800
 801         ldap_pvt_thread_cond_broadcast(&pool->ltp_cond);
 802         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 803
 804         for (i=0; i<pool->ltp_numqs; i++) {
 805                 pq = pool->ltp_wqs[i];
 806                 ldap_pvt_thread_mutex_lock(&pq->ltp_mutex);
 807                 if (pq->ltp_max_pending > 0)
 808                         pq->ltp_max_pending = -pq->ltp_max_pending;
 809                 if (!run_pending) {
 810                         while ((task = LDAP_STAILQ_FIRST(&pq->ltp_pending_list)) != NULL) {
 811                                 LDAP_STAILQ_REMOVE_HEAD(&pq->ltp_pending_list, ltt_next.q);
 812                                 LDAP_FREE(task);
 813                         }
 814                         pq->ltp_pending_count = 0;
 815                 }
 816
 817                 while (pq->ltp_open_count) {
 818                         ldap_pvt_thread_cond_broadcast(&pq->ltp_cond);
 819                         ldap_pvt_thread_cond_wait(&pq->ltp_cond, &pq->ltp_mutex);
 820                 }
 821
 822                 while ((task = LDAP_SLIST_FIRST(&pq->ltp_free_list)) != NULL)
 823                 {
 824                         LDAP_SLIST_REMOVE_HEAD(&pq->ltp_free_list, ltt_next.l);
 825                         LDAP_FREE(task);
 826                 }
 827                 ldap_pvt_thread_mutex_unlock(&pq->ltp_mutex);
 828                 ldap_pvt_thread_cond_destroy(&pq->ltp_cond);
 829                 ldap_pvt_thread_mutex_destroy(&pq->ltp_mutex);
 830         }
 831
 832         ldap_pvt_thread_cond_destroy(&pool->ltp_pcond);
 833         ldap_pvt_thread_cond_destroy(&pool->ltp_cond);
 834         ldap_pvt_thread_mutex_destroy(&pool->ltp_mutex);
 835         for (i=0; i<pool->ltp_numqs; i++) {
 836                 pq = pool->ltp_wqs[i];
 837                 if (pq->ltp_free) {
 838                         LDAP_FREE(pq->ltp_free);
 839                 }
 840         }
 841         LDAP_FREE(pool->ltp_wqs);
 842         LDAP_FREE(pool);
 843         *tpool = NULL;
 844         ldap_int_has_thread_pool = 0;
 845         return(0);
 846 }
 847
 848 /* Thread loop.  Accept and handle submitted tasks. */
 849 static void *
 850 ldap_int_thread_pool_wrapper (
 851         void *xpool )
 852 {
 853         struct ldap_int_thread_poolq_s *pq = xpool;
 854         struct ldap_int_thread_pool_s *pool = pq->ltp_pool;
 855         ldap_int_thread_task_t *task;
 856         ldap_int_tpool_plist_t *work_list;
 857         ldap_int_thread_userctx_t ctx, *kctx;
 858         unsigned i, keyslot, hash;
 859         int pool_lock = 0, freeme = 0;
 860
 861         assert(pool != NULL);
 862
 863         for ( i=0; i<MAXKEYS; i++ ) {
 864                 ctx.ltu_key[i].ltk_key = NULL;
 865         }
 866
 867         ctx.ltu_pq = pq;
 868         ctx.ltu_id = ldap_pvt_thread_self();
 869         TID_HASH(ctx.ltu_id, hash);
 870
 871         ldap_pvt_thread_key_setdata( ldap_tpool_key, &ctx );
 872
 873         if (pool->ltp_pause) {
 874                 ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 875                 /* thread_keys[] is read-only when paused */
 876                 while (pool->ltp_pause)
 877                         ldap_pvt_thread_cond_wait(&pool->ltp_cond, &pool->ltp_mutex);
 878                 ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 879         }
 880
 881         /* find a key slot to give this thread ID and store a
 882          * pointer to our keys there; start at the thread ID
 883          * itself (mod LDAP_MAXTHR) and look for an empty slot.
 884          */
 885         ldap_pvt_thread_mutex_lock(&ldap_pvt_thread_pool_mutex);
 886         for (keyslot = hash & (LDAP_MAXTHR-1);
 887                 (kctx = thread_keys[keyslot].ctx) && kctx != DELETED_THREAD_CTX;
 888                 keyslot = (keyslot+1) & (LDAP_MAXTHR-1));
 889         thread_keys[keyslot].ctx = &ctx;
 890         ldap_pvt_thread_mutex_unlock(&ldap_pvt_thread_pool_mutex);
 891
 892         ldap_pvt_thread_mutex_lock(&pq->ltp_mutex);
 893         pq->ltp_starting--;
 894         pq->ltp_active_count++;
 895
 896         for (;;) {
 897                 work_list = pq->ltp_work_list; /* help the compiler a bit */
 898                 task = LDAP_STAILQ_FIRST(work_list);
 899                 if (task == NULL) {     /* paused or no pending tasks */
 900                         if (--(pq->ltp_active_count) < 1) {
 901                                 if (pool->ltp_pause) {
 902                                         ldap_pvt_thread_mutex_unlock(&pq->ltp_mutex);
 903                                         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 904                                         pool_lock = 1;
 905                                         if (--(pool->ltp_active_queues) < 1) {
 906                                                 /* Notify pool_pause it is the sole active thread. */
 907                                                 ldap_pvt_thread_cond_signal(&pool->ltp_pcond);
 908                                         }
 909                                 }
 910                         }
 911
 912                         do {
 913                                 if (pool->ltp_finishing || pq->ltp_open_count > pq->ltp_max_count) {
 914                                         /* Not paused, and either finishing or too many
 915                                          * threads running (can happen if ltp_max_count
 916                                          * was reduced).  Let this thread die.
 917                                          */
 918                                         goto done;
 919                                 }
 920
 921                                 /* We could check an idle timer here, and let the
 922                                  * thread die if it has been inactive for a while.
 923                                  * Only die if there are other open threads (i.e.,
 924                                  * always have at least one thread open).
 925                                  * The check should be like this:
 926                                  *   if (pool->ltp_open_count>1 && pool->ltp_starting==0)
 927                                  *       check timer, wait if ltp_pause, leave thread;
 928                                  *
 929                                  * Just use pthread_cond_timedwait() if we want to
 930                                  * check idle time.
 931                                  */
 932                                 if (pool_lock) {
 933                                         ldap_pvt_thread_cond_wait(&pool->ltp_cond, &pool->ltp_mutex);
 934                                         if (!pool->ltp_pause) {
 935                                                 ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 936                                                 ldap_pvt_thread_mutex_lock(&pq->ltp_mutex);
 937                                                 pool_lock = 0;
 938                                         }
 939                                 } else
 940                                         ldap_pvt_thread_cond_wait(&pq->ltp_cond, &pq->ltp_mutex);
 941
 942                                 work_list = pq->ltp_work_list;
 943                                 task = LDAP_STAILQ_FIRST(work_list);
 944                         } while (task == NULL);
 945
 946                         if (pool_lock) {
 947                                 ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 948                                 ldap_pvt_thread_mutex_lock(&pq->ltp_mutex);
 949                                 pool_lock = 0;
 950                         }
 951                         pq->ltp_active_count++;
 952                 }
 953
 954                 LDAP_STAILQ_REMOVE_HEAD(work_list, ltt_next.q);
 955                 pq->ltp_pending_count--;
 956                 ldap_pvt_thread_mutex_unlock(&pq->ltp_mutex);
 957
 958                 task->ltt_start_routine(&ctx, task->ltt_arg);
 959
 960                 ldap_pvt_thread_mutex_lock(&pq->ltp_mutex);
 961                 LDAP_SLIST_INSERT_HEAD(&pq->ltp_free_list, task, ltt_next.l);
 962         }
 963  done:
 964
 965         ldap_pvt_thread_mutex_lock(&ldap_pvt_thread_pool_mutex);
 966
 967         /* The pool_mutex lock protects ctx->ltu_key from pool_purgekey()
 968          * during this call, since it prevents new pauses. */
 969         ldap_pvt_thread_pool_context_reset(&ctx);
 970
 971         thread_keys[keyslot].ctx = DELETED_THREAD_CTX;
 972         ldap_pvt_thread_mutex_unlock(&ldap_pvt_thread_pool_mutex);
 973
 974         pq->ltp_open_count--;
 975         if (pq->ltp_open_count == 0) {
 976                 if (pool->ltp_finishing)
 977                         /* let pool_destroy know we're all done */
 978                         ldap_pvt_thread_cond_signal(&pq->ltp_cond);
 979                 else
 980                         freeme = 1;
 981         }
 982
 983         if (pool_lock)
 984                 ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 985         else
 986                 ldap_pvt_thread_mutex_unlock(&pq->ltp_mutex);
 987
 988         if (freeme) {
 989                 ldap_pvt_thread_cond_destroy(&pq->ltp_cond);
 990                 ldap_pvt_thread_mutex_destroy(&pq->ltp_mutex);
 991                 LDAP_FREE(pq->ltp_free);
 992                 pq->ltp_free = NULL;
 993         }
 994         ldap_pvt_thread_exit(NULL);
 995         return(NULL);
 996 }
 997
 998 /* Arguments > ltp_pause to handle_pause(,PAUSE_ARG()).  arg=PAUSE_ARG
 999  * ensures (arg-ltp_pause) sets GO_* at need and keeps DO_PAUSE/GO_*.
1000  */
1001 #define GO_IDLE         8
1002 #define GO_UNIDLE       16
1003 #define CHECK_PAUSE     32      /* if ltp_pause: GO_IDLE; wait; GO_UNIDLE */
1004 #define DO_PAUSE        64      /* CHECK_PAUSE; pause the pool */
1005 #define PAUSE_ARG(a) \
1006                 ((a) | ((a) & (GO_IDLE|GO_UNIDLE) ? GO_IDLE-1 : CHECK_PAUSE))
1007
1008 static int
1009 handle_pause( ldap_pvt_thread_pool_t *tpool, int pause_type )
1010 {
1011         struct ldap_int_thread_pool_s *pool;
1012         struct ldap_int_thread_poolq_s *pq;
1013         int ret = 0, pause, max_ltp_pause;
1014
1015         if (tpool == NULL)
1016                 return(-1);
1017
1018         pool = *tpool;
1019
1020         if (pool == NULL)
1021                 return(0);
1022
1023         if (pause_type == CHECK_PAUSE && !pool->ltp_pause)
1024                 return(0);
1025
1026         {
1027                 ldap_int_thread_userctx_t *ctx = ldap_pvt_thread_pool_context();
1028                 pq = ctx->ltu_pq;
1029         }
1030
1031         /* Let pool_unidle() ignore requests for new pauses */
1032         max_ltp_pause = pause_type==PAUSE_ARG(GO_UNIDLE) ? WANT_PAUSE : NOT_PAUSED;
1033
1034         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
1035
1036         pause = pool->ltp_pause;        /* NOT_PAUSED, WANT_PAUSE or PAUSED */
1037
1038         /* If ltp_pause and not GO_IDLE|GO_UNIDLE: Set GO_IDLE,GO_UNIDLE */
1039         pause_type -= pause;
1040
1041         if (pause_type & GO_IDLE) {
1042                 int do_pool = 0;
1043                 ldap_pvt_thread_mutex_lock(&pq->ltp_mutex);
1044                 pq->ltp_pending_count++;
1045                 pq->ltp_active_count--;
1046                 if (pause && pq->ltp_active_count < 1) {
1047                         do_pool = 1;
1048                 }
1049                 ldap_pvt_thread_mutex_unlock(&pq->ltp_mutex);
1050                 if (do_pool) {
1051                         pool->ltp_active_queues--;
1052                         if (pool->ltp_active_queues < 1)
1053                         /* Tell the task waiting to DO_PAUSE it can proceed */
1054                                 ldap_pvt_thread_cond_signal(&pool->ltp_pcond);
1055                 }
1056         }
1057
1058         if (pause_type & GO_UNIDLE) {
1059                 /* Wait out pause if any, then cancel GO_IDLE */
1060                 if (pause > max_ltp_pause) {
1061                         ret = 1;
1062                         do {
1063                                 ldap_pvt_thread_cond_wait(&pool->ltp_cond, &pool->ltp_mutex);
1064                         } while (pool->ltp_pause > max_ltp_pause);
1065                 }
1066                 ldap_pvt_thread_mutex_lock(&pq->ltp_mutex);
1067                 pq->ltp_pending_count--;
1068                 pq->ltp_active_count++;
1069                 ldap_pvt_thread_mutex_unlock(&pq->ltp_mutex);
1070         }
1071
1072         if (pause_type & DO_PAUSE) {
1073                 int i, j;
1074                 /* Tell everyone else to pause or finish, then await that */
1075                 ret = 0;
1076                 assert(!pool->ltp_pause);
1077                 pool->ltp_pause = WANT_PAUSE;
1078                 pool->ltp_active_queues = 0;
1079
1080                 for (i=0; i<pool->ltp_numqs; i++)
1081                         if (pool->ltp_wqs[i] == pq) break;
1082
1083                 ldap_pvt_thread_mutex_lock(&pq->ltp_mutex);
1084                 /* temporarily remove ourself from active count */
1085                 pq->ltp_active_count--;
1086
1087                 j=i;
1088                 do {
1089                         pq = pool->ltp_wqs[j];
1090                         if (j != i)
1091                                 ldap_pvt_thread_mutex_lock(&pq->ltp_mutex);
1092
1093                         /* Let ldap_pvt_thread_pool_submit() through to its ltp_pause test,
1094                          * and do not finish threads in ldap_pvt_thread_pool_wrapper() */
1095                         pq->ltp_open_count = -pq->ltp_open_count;
1096                         /* Hide pending tasks from ldap_pvt_thread_pool_wrapper() */
1097                         pq->ltp_work_list = &empty_pending_list;
1098
1099                         if (pq->ltp_active_count > 0)
1100                                 pool->ltp_active_queues++;
1101
1102                         ldap_pvt_thread_mutex_unlock(&pq->ltp_mutex);
1103                         if (pool->ltp_numqs > 1) {
1104                                 j++;
1105                                 j %= pool->ltp_numqs;
1106                         }
1107                 } while (j != i);
1108
1109                 /* Wait for this task to become the sole active task */
1110                 while (pool->ltp_active_queues > 0)
1111                         ldap_pvt_thread_cond_wait(&pool->ltp_pcond, &pool->ltp_mutex);
1112
1113                 /* restore us to active count */
1114                 pool->ltp_wqs[i]->ltp_active_count++;
1115
1116                 assert(pool->ltp_pause == WANT_PAUSE);
1117                 pool->ltp_pause = PAUSED;
1118         }
1119         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
1120
1121         return(ret);
1122 }
1123
1124 /* Consider this task idle: It will not block pool_pause() in other tasks. */
1125 void
1126 ldap_pvt_thread_pool_idle( ldap_pvt_thread_pool_t *tpool )
1127 {
1128         handle_pause(tpool, PAUSE_ARG(GO_IDLE));
1129 }
1130
1131 /* Cancel pool_idle(). If the pool is paused, wait it out first. */
1132 void
1133 ldap_pvt_thread_pool_unidle( ldap_pvt_thread_pool_t *tpool )
1134 {
1135         handle_pause(tpool, PAUSE_ARG(GO_UNIDLE));
1136 }
1137
1138 /*
1139  * If a pause was requested, wait for it.  If several threads
1140  * are waiting to pause, let through one or more pauses.
1141  * The calling task must be active, not idle.
1142  * Return 1 if we waited, 0 if not, -1 at parameter error.
1143  */
1144 int
1145 ldap_pvt_thread_pool_pausecheck( ldap_pvt_thread_pool_t *tpool )
1146 {
1147         return handle_pause(tpool, PAUSE_ARG(CHECK_PAUSE));
1148 }
1149
1150 /*
1151  * Pause the pool.  The calling task must be active, not idle.
1152  * Return when all other tasks are paused or idle.
1153  */
1154 int
1155 ldap_pvt_thread_pool_pause( ldap_pvt_thread_pool_t *tpool )
1156 {
1157         return handle_pause(tpool, PAUSE_ARG(DO_PAUSE));
1158 }
1159
1160 /* End a pause */
1161 int
1162 ldap_pvt_thread_pool_resume (
1163         ldap_pvt_thread_pool_t *tpool )
1164 {
1165         struct ldap_int_thread_pool_s *pool;
1166         struct ldap_int_thread_poolq_s *pq;
1167         int i;
1168
1169         if (tpool == NULL)
1170                 return(-1);
1171
1172         pool = *tpool;
1173
1174         if (pool == NULL)
1175                 return(0);
1176
1177         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
1178         assert(pool->ltp_pause == PAUSED);
1179         pool->ltp_pause = 0;
1180         for (i=0; i<pool->ltp_numqs; i++) {
1181                 pq = pool->ltp_wqs[i];
1182                 if (pq->ltp_open_count <= 0) /* true when paused, but be paranoid */
1183                         pq->ltp_open_count = -pq->ltp_open_count;
1184                 pq->ltp_work_list = &pq->ltp_pending_list;
1185                 ldap_pvt_thread_cond_broadcast(&pq->ltp_cond);
1186         }
1187         ldap_pvt_thread_cond_broadcast(&pool->ltp_cond);
1188         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
1189         return(0);
1190 }
1191
1192 /*
1193  * Get the key's data and optionally free function in the given context.
1194  */
1195 int ldap_pvt_thread_pool_getkey(
1196         void *xctx,
1197         void *key,
1198         void **data,
1199         ldap_pvt_thread_pool_keyfree_t **kfree )
1200 {
1201         ldap_int_thread_userctx_t *ctx = xctx;
1202         int i;
1203
1204         if ( !ctx || !key || !data ) return EINVAL;
1205
1206         for ( i=0; i<MAXKEYS && ctx->ltu_key[i].ltk_key; i++ ) {
1207                 if ( ctx->ltu_key[i].ltk_key == key ) {
1208                         *data = ctx->ltu_key[i].ltk_data;
1209                         if ( kfree ) *kfree = ctx->ltu_key[i].ltk_free;
1210                         return 0;
1211                 }
1212         }
1213         return ENOENT;
1214 }
1215
1216 static void
1217 clear_key_idx( ldap_int_thread_userctx_t *ctx, int i )
1218 {
1219         for ( ; i < MAXKEYS-1 && ctx->ltu_key[i+1].ltk_key; i++ )
1220                 ctx->ltu_key[i] = ctx->ltu_key[i+1];
1221         ctx->ltu_key[i].ltk_key = NULL;
1222 }
1223
1224 /*
1225  * Set or remove data for the key in the given context.
1226  * key can be any unique pointer.
1227  * kfree() is an optional function to free the data (but not the key):
1228  *   pool_context_reset() and pool_purgekey() call kfree(key, data),
1229  *   but pool_setkey() does not.  For pool_setkey() it is the caller's
1230  *   responsibility to free any existing data with the same key.
1231  *   kfree() must not call functions taking a tpool argument.
1232  */
1233 int ldap_pvt_thread_pool_setkey(
1234         void *xctx,
1235         void *key,
1236         void *data,
1237         ldap_pvt_thread_pool_keyfree_t *kfree,
1238         void **olddatap,
1239         ldap_pvt_thread_pool_keyfree_t **oldkfreep )
1240 {
1241         ldap_int_thread_userctx_t *ctx = xctx;
1242         int i, found;
1243
1244         if ( !ctx || !key ) return EINVAL;
1245
1246         for ( i=found=0; i<MAXKEYS; i++ ) {
1247                 if ( ctx->ltu_key[i].ltk_key == key ) {
1248                         found = 1;
1249                         break;
1250                 } else if ( !ctx->ltu_key[i].ltk_key ) {
1251                         break;
1252                 }
1253         }
1254
1255         if ( olddatap ) {
1256                 if ( found ) {
1257                         *olddatap = ctx->ltu_key[i].ltk_data;
1258                 } else {
1259                         *olddatap = NULL;
1260                 }
1261         }
1262
1263         if ( oldkfreep ) {
1264                 if ( found ) {
1265                         *oldkfreep = ctx->ltu_key[i].ltk_free;
1266                 } else {
1267                         *oldkfreep = 0;
1268                 }
1269         }
1270
1271         if ( data || kfree ) {
1272                 if ( i>=MAXKEYS )
1273                         return ENOMEM;
1274                 ctx->ltu_key[i].ltk_key = key;
1275                 ctx->ltu_key[i].ltk_data = data;
1276                 ctx->ltu_key[i].ltk_free = kfree;
1277         } else if ( found ) {
1278                 clear_key_idx( ctx, i );
1279         }
1280
1281         return 0;
1282 }
1283
1284 /* Free all elements with this key, no matter which thread they're in.
1285  * May only be called while the pool is paused.
1286  */
1287 void ldap_pvt_thread_pool_purgekey( void *key )
1288 {
1289         int i, j;
1290         ldap_int_thread_userctx_t *ctx;
1291
1292         assert ( key != NULL );
1293
1294         ldap_pvt_thread_mutex_lock(&ldap_pvt_thread_pool_mutex);
1295         for ( i=0; i<LDAP_MAXTHR; i++ ) {
1296                 ctx = thread_keys[i].ctx;
1297                 if ( ctx && ctx != DELETED_THREAD_CTX ) {
1298                         for ( j=0; j<MAXKEYS && ctx->ltu_key[j].ltk_key; j++ ) {
1299                                 if ( ctx->ltu_key[j].ltk_key == key ) {
1300                                         if (ctx->ltu_key[j].ltk_free)
1301                                                 ctx->ltu_key[j].ltk_free( ctx->ltu_key[j].ltk_key,
1302                                                 ctx->ltu_key[j].ltk_data );
1303                                         clear_key_idx( ctx, j );
1304                                         break;
1305                                 }
1306                         }
1307                 }
1308         }
1309         ldap_pvt_thread_mutex_unlock(&ldap_pvt_thread_pool_mutex);
1310 }
1311
1312 /*
1313  * Find the context of the current thread.
1314  * This is necessary if the caller does not have access to the
1315  * thread context handle (for example, a slapd plugin calling
1316  * slapi_search_internal()). No doubt it is more efficient
1317  * for the application to keep track of the thread context
1318  * handles itself.
1319  */
1320 void *ldap_pvt_thread_pool_context( )
1321 {
1322         void *ctx = NULL;
1323
1324         ldap_pvt_thread_key_getdata( ldap_tpool_key, &ctx );
1325         return ctx ? ctx : (void *) &ldap_int_main_thrctx;
1326 }
1327
1328 /*
1329  * Free the context's keys.
1330  * Must not call functions taking a tpool argument (because this
1331  * thread already holds ltp_mutex when called from pool_wrapper()).
1332  */
1333 void ldap_pvt_thread_pool_context_reset( void *vctx )
1334 {
1335         ldap_int_thread_userctx_t *ctx = vctx;
1336         int i;
1337
1338         for ( i=MAXKEYS-1; i>=0; i--) {
1339                 if ( !ctx->ltu_key[i].ltk_key )
1340                         continue;
1341                 if ( ctx->ltu_key[i].ltk_free )
1342                         ctx->ltu_key[i].ltk_free( ctx->ltu_key[i].ltk_key,
1343                         ctx->ltu_key[i].ltk_data );
1344                 ctx->ltu_key[i].ltk_key = NULL;
1345         }
1346 }
1347
1348 ldap_pvt_thread_t ldap_pvt_thread_pool_tid( void *vctx )
1349 {
1350         ldap_int_thread_userctx_t *ctx = vctx;
1351
1352         return ctx->ltu_id;
1353 }
1354 #endif /* LDAP_THREAD_HAVE_TPOOL */