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