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