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-2007 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/stdarg.h>
  21 #include <ac/stdlib.h>
  22 #include <ac/string.h>
  23 #include <ac/time.h>
  24 #include <ac/errno.h>
  25
  26 #include "ldap-int.h"
  27 #include "ldap_pvt_thread.h" /* Get the thread interface */
  28 #include "ldap_queue.h"
  29 #define LDAP_THREAD_POOL_IMPLEMENTATION
  30 #include "ldap_thr_debug.h"  /* May rename symbols defined below */
  31
  32 #ifndef LDAP_THREAD_HAVE_TPOOL
  33
  34 typedef enum ldap_int_thread_pool_state_e {
  35         LDAP_INT_THREAD_POOL_RUNNING,
  36         LDAP_INT_THREAD_POOL_FINISHING,
  37         LDAP_INT_THREAD_POOL_STOPPING
  38 } ldap_int_thread_pool_state_t;
  39
  40 typedef struct ldap_int_thread_key_s {
  41         void *ltk_key;
  42         void *ltk_data;
  43         ldap_pvt_thread_pool_keyfree_t *ltk_free;
  44 } ldap_int_thread_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 #define LDAP_MAXTHR     1024    /* must be a power of 2 */
  51
  52 typedef struct ldap_int_thread_userctx_s {
  53         ldap_pvt_thread_t ltu_id;
  54         ldap_int_thread_key_t ltu_key[MAXKEYS];
  55 } ldap_int_thread_userctx_t;
  56
  57 static ldap_pvt_thread_t tid_zero;
  58
  59 /* Thread ID -> context mapping (poor open-addressed hash table).
  60  * Protected by ldap_pvt_thread_pool_mutex except during pauses,
  61  * when it is reserved for ldap_pvt_thread_pool_purgekey().
  62  */
  63 static struct {
  64         ldap_pvt_thread_t id;
  65         ldap_int_thread_userctx_t *ctx;         /* set when id != tid_zero */
  66 #       define DELETED_THREAD_CTX (&ldap_int_main_thrctx + 1) /* dummy addr */
  67 } thread_keys[LDAP_MAXTHR];
  68
  69 #define TID_HASH(tid, hash) do { \
  70         unsigned const char *ptr_ = (unsigned const char *)&(tid); \
  71         unsigned i_; \
  72         for (i_ = 0, (hash) = ptr_[0]; ++i_ < sizeof(tid);) \
  73                 (hash) += ((hash) << 5) ^ ptr_[i_]; \
  74 } while(0)
  75
  76
  77 typedef struct ldap_int_thread_ctx_s {
  78         union {
  79         LDAP_STAILQ_ENTRY(ldap_int_thread_ctx_s) q;
  80         LDAP_SLIST_ENTRY(ldap_int_thread_ctx_s) l;
  81         LDAP_SLIST_ENTRY(ldap_int_thread_ctx_s) al;
  82         } ltc_next;
  83         ldap_pvt_thread_start_t *ltc_start_routine;
  84         void *ltc_arg;
  85 } ldap_int_thread_ctx_t;
  86
  87 struct ldap_int_thread_pool_s {
  88         LDAP_STAILQ_ENTRY(ldap_int_thread_pool_s) ltp_next;
  89         ldap_pvt_thread_mutex_t ltp_mutex;
  90         ldap_pvt_thread_cond_t ltp_cond;
  91         ldap_pvt_thread_cond_t ltp_pcond;
  92         LDAP_STAILQ_HEAD(tcq, ldap_int_thread_ctx_s) ltp_pending_list;
  93         LDAP_SLIST_HEAD(tcl, ldap_int_thread_ctx_s) ltp_free_list;
  94         LDAP_SLIST_HEAD(tclq, ldap_int_thread_ctx_s) ltp_active_list;
  95         ldap_int_thread_pool_state_t ltp_state;
  96         int ltp_pause;
  97         long ltp_max_count;
  98         long ltp_max_pending;
  99         long ltp_pending_count;
 100         long ltp_active_count;
 101         long ltp_open_count;
 102         long ltp_starting;
 103 };
 104
 105 static LDAP_STAILQ_HEAD(tpq, ldap_int_thread_pool_s)
 106         ldap_int_thread_pool_list =
 107         LDAP_STAILQ_HEAD_INITIALIZER(ldap_int_thread_pool_list);
 108
 109 static ldap_pvt_thread_mutex_t ldap_pvt_thread_pool_mutex;
 110
 111 static void *ldap_int_thread_pool_wrapper( void *pool );
 112
 113 static ldap_pvt_thread_t ldap_int_main_tid;
 114
 115 static ldap_int_thread_userctx_t ldap_int_main_thrctx;
 116
 117 int
 118 ldap_int_thread_pool_startup ( void )
 119 {
 120         ldap_int_main_tid = ldap_pvt_thread_self();
 121         ldap_int_main_thrctx.ltu_id = ldap_int_main_tid;
 122
 123         return ldap_pvt_thread_mutex_init(&ldap_pvt_thread_pool_mutex);
 124 }
 125
 126 int
 127 ldap_int_thread_pool_shutdown ( void )
 128 {
 129         struct ldap_int_thread_pool_s *pool;
 130
 131         while ((pool = LDAP_STAILQ_FIRST(&ldap_int_thread_pool_list)) != NULL) {
 132                 (ldap_pvt_thread_pool_destroy)(&pool, 0); /* ignore thr_debug macro */
 133         }
 134         ldap_pvt_thread_mutex_destroy(&ldap_pvt_thread_pool_mutex);
 135         return(0);
 136 }
 137
 138 typedef struct ldap_lazy_sem_t {
 139         ldap_pvt_thread_mutex_t ls_mutex;
 140         ldap_pvt_thread_cond_t  ls_cond;
 141         int ls_sem_value;
 142         /*
 143          * when more than ls_lazy_count number of resources
 144          * becmoes available, the thread wating for the resources will
 145          * be waken up in order to prevent frequent blocking/waking-up
 146          */
 147         unsigned int ls_lazy_count;
 148         /*
 149          * only one thread(listener) will wait on this semaphore
 150          * using a flag instead of a list
 151          */
 152         int ls_wait;
 153 } ldap_lazy_sem_t;
 154
 155 ldap_lazy_sem_t* thread_pool_sem = NULL;
 156
 157 int
 158 ldap_lazy_sem_init( unsigned int value, unsigned int lazyness )
 159 {
 160         thread_pool_sem = (ldap_lazy_sem_t*) LDAP_CALLOC(1,
 161                 sizeof( ldap_lazy_sem_t ));
 162
 163         if( thread_pool_sem == NULL ) return -1;
 164
 165         ldap_pvt_thread_mutex_init( &thread_pool_sem->ls_mutex );
 166         ldap_pvt_thread_cond_init( &thread_pool_sem->ls_cond );
 167         thread_pool_sem->ls_sem_value = value;
 168         thread_pool_sem->ls_lazy_count = lazyness;
 169         thread_pool_sem->ls_wait = 0;
 170
 171         return 0;
 172 }
 173
 174 /* FIXME: move to some approprite header */
 175 int ldap_lazy_sem_dec( ldap_lazy_sem_t* ls );
 176 int ldap_lazy_sem_wait ( ldap_lazy_sem_t* ls );
 177
 178 /*
 179  * ldap_lazy_sem_wait is used if a caller is blockable(listener).
 180  * Otherwise use ldap_lazy_sem_dec (worker)
 181  */
 182 int
 183 ldap_lazy_sem_op_submit( ldap_lazy_sem_t* ls )
 184 {
 185         if ( ls == NULL ) return -1;
 186
 187         /* only worker thread has its thread ctx */
 188         if ( ldap_pvt_thread_pool_context() ) {
 189                 /* worker thread */
 190                 return ldap_lazy_sem_dec( ls );
 191         } else {
 192                 /* listener */
 193                 return ldap_lazy_sem_wait( ls );
 194         }
 195 }
 196
 197 /*
 198  * test if given semaphore's count is zero.
 199  * If 0, the caller is blocked
 200  * If not, the count is decremented.
 201  */
 202 int
 203 ldap_lazy_sem_wait ( ldap_lazy_sem_t* ls )
 204 {
 205         ldap_pvt_thread_mutex_lock( &ls->ls_mutex );
 206
 207 lazy_sem_retry:
 208         if ( ls->ls_sem_value <= 0 ) {
 209                 /* no more avaliable resources */
 210                 ls->ls_wait = 1;
 211                 ldap_pvt_thread_cond_wait( &ls->ls_cond, &ls->ls_mutex );
 212                 goto lazy_sem_retry;
 213         } else {
 214                 /* avaliable resources */
 215                 ls->ls_sem_value--;
 216         }
 217
 218         ldap_pvt_thread_mutex_unlock( &ls->ls_mutex );
 219
 220         return 0;
 221 }
 222
 223 /*
 224  * decrement the count without blocking
 225  * even when the count becomes less than or equal to 0
 226  */
 227 int
 228 ldap_lazy_sem_dec( ldap_lazy_sem_t* ls )
 229 {
 230         ldap_pvt_thread_mutex_lock( &ls->ls_mutex );
 231
 232         ls->ls_sem_value--;
 233
 234         ldap_pvt_thread_mutex_unlock( &ls->ls_mutex );
 235
 236         return 0;
 237 }
 238
 239 /*
 240  * Increment the count by one and test if it is greater or
 241  * equal to lazyness. If it is, wake up a blocked thread.
 242  */
 243 int
 244 ldap_lazy_sem_post( ldap_lazy_sem_t* ls )
 245 {
 246         if( ls == NULL ) return (-1);
 247
 248         ldap_pvt_thread_mutex_lock( &ls->ls_mutex );
 249
 250         ls->ls_sem_value++;
 251         if ( ls->ls_wait ) {
 252                 if ( ls->ls_sem_value >= ls->ls_lazy_count ) {
 253                         ls->ls_wait = 0;
 254                         ldap_pvt_thread_cond_signal( &ls->ls_cond );
 255                 }
 256         }
 257
 258         ldap_pvt_thread_mutex_unlock( &ls->ls_mutex );
 259
 260         return 0;
 261 }
 262
 263 int
 264 ldap_pvt_thread_pool_init (
 265         ldap_pvt_thread_pool_t *tpool,
 266         int max_threads,
 267         int max_pending )
 268 {
 269         ldap_pvt_thread_pool_t pool;
 270         int rc;
 271
 272         *tpool = NULL;
 273         pool = (ldap_pvt_thread_pool_t) LDAP_CALLOC(1,
 274                 sizeof(struct ldap_int_thread_pool_s));
 275
 276         if (pool == NULL) return(-1);
 277
 278         rc = ldap_pvt_thread_mutex_init(&pool->ltp_mutex);
 279         if (rc != 0)
 280                 return(rc);
 281         rc = ldap_pvt_thread_cond_init(&pool->ltp_cond);
 282         if (rc != 0)
 283                 return(rc);
 284         rc = ldap_pvt_thread_cond_init(&pool->ltp_pcond);
 285         if (rc != 0)
 286                 return(rc);
 287         pool->ltp_state = LDAP_INT_THREAD_POOL_RUNNING;
 288         pool->ltp_max_count = max_threads;
 289         pool->ltp_max_pending = max_pending;
 290         LDAP_STAILQ_INIT(&pool->ltp_pending_list);
 291         LDAP_SLIST_INIT(&pool->ltp_free_list);
 292         LDAP_SLIST_INIT(&pool->ltp_active_list);
 293         ldap_pvt_thread_mutex_lock(&ldap_pvt_thread_pool_mutex);
 294         LDAP_STAILQ_INSERT_TAIL(&ldap_int_thread_pool_list, pool, ltp_next);
 295         ldap_pvt_thread_mutex_unlock(&ldap_pvt_thread_pool_mutex);
 296
 297 #if 0
 298         /* THIS WILL NOT WORK on some systems.  If the process
 299          * forks after starting a thread, there is no guarantee
 300          * that the thread will survive the fork.  For example,
 301          * slapd forks in order to daemonize, and does so after
 302          * calling ldap_pvt_thread_pool_init.  On some systems,
 303          * this initial thread does not run in the child process,
 304          * but ltp_open_count == 1, so two things happen:
 305          * 1) the first client connection fails, and 2) when
 306          * slapd is kill'ed, it never terminates since it waits
 307          * for all worker threads to exit. */
 308
 309         /* start up one thread, just so there is one. no need to
 310          * lock the mutex right now, since no threads are running.
 311          */
 312         pool->ltp_open_count++;
 313
 314         ldap_pvt_thread_t thr;
 315         rc = ldap_pvt_thread_create( &thr, 1, ldap_int_thread_pool_wrapper, pool );
 316
 317         if( rc != 0) {
 318                 /* couldn't start one?  then don't start any */
 319                 ldap_pvt_thread_mutex_lock(&ldap_pvt_thread_pool_mutex);
 320                 LDAP_STAILQ_REMOVE(ldap_int_thread_pool_list, pool,
 321                         ldap_int_thread_pool_s, ltp_next);
 322                 ldap_pvt_thread_mutex_unlock(&ldap_pvt_thread_pool_mutex);
 323                 ldap_pvt_thread_cond_destroy(&pool->ltp_pcond);
 324                 ldap_pvt_thread_cond_destroy(&pool->ltp_cond);
 325                 ldap_pvt_thread_mutex_destroy(&pool->ltp_mutex);
 326                 LDAP_FREE(pool);
 327                 return(-1);
 328         }
 329 #endif
 330
 331         *tpool = pool;
 332         return(0);
 333 }
 334
 335
 336 int
 337 ldap_pvt_thread_pool_submit (
 338         ldap_pvt_thread_pool_t *tpool,
 339         ldap_pvt_thread_start_t *start_routine, void *arg )
 340 {
 341         struct ldap_int_thread_pool_s *pool;
 342         ldap_int_thread_ctx_t *ctx;
 343         int need_thread = 0;
 344         ldap_pvt_thread_t thr;
 345
 346         if (tpool == NULL)
 347                 return(-1);
 348
 349         pool = *tpool;
 350
 351         if (pool == NULL)
 352                 return(-1);
 353
 354         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 355         if (pool->ltp_state != LDAP_INT_THREAD_POOL_RUNNING
 356                 || (pool->ltp_max_pending > 0
 357                         && pool->ltp_pending_count >= pool->ltp_max_pending))
 358         {
 359                 ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 360                 return(-1);
 361         }
 362
 363         ctx = LDAP_SLIST_FIRST(&pool->ltp_free_list);
 364         if (ctx) {
 365                 LDAP_SLIST_REMOVE_HEAD(&pool->ltp_free_list, ltc_next.l);
 366         } else {
 367                 ctx = (ldap_int_thread_ctx_t *) LDAP_MALLOC(
 368                         sizeof(ldap_int_thread_ctx_t));
 369                 if (ctx == NULL) {
 370                         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 371                         return(-1);
 372                 }
 373         }
 374
 375         ctx->ltc_start_routine = start_routine;
 376         ctx->ltc_arg = arg;
 377
 378         pool->ltp_pending_count++;
 379         LDAP_STAILQ_INSERT_TAIL(&pool->ltp_pending_list, ctx, ltc_next.q);
 380         if (pool->ltp_pause) {
 381                 ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 382                 return(0);
 383         }
 384         ldap_pvt_thread_cond_signal(&pool->ltp_cond);
 385         if (pool->ltp_open_count < pool->ltp_active_count + pool->ltp_pending_count
 386                 && (pool->ltp_open_count < pool->ltp_max_count ||
 387                         pool->ltp_max_count <= 0 ))
 388         {
 389                 pool->ltp_open_count++;
 390                 pool->ltp_starting++;
 391                 need_thread = 1;
 392         }
 393         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 394
 395 #ifdef LDAP_PVT_THREAD_POOL_SEM_LOAD_CONTROL
 396         ldap_lazy_sem_op_submit( thread_pool_sem );
 397 #endif
 398
 399         if (need_thread) {
 400                 int rc;
 401
 402                 ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 403
 404                 rc = ldap_pvt_thread_create( &thr, 1,
 405                         ldap_int_thread_pool_wrapper, pool );
 406                 pool->ltp_starting--;
 407                 if (rc != 0) {
 408                         /* couldn't create thread.  back out of
 409                          * ltp_open_count and check for even worse things.
 410                          */
 411                         pool->ltp_open_count--;
 412                         if (pool->ltp_open_count == 0) {
 413                                 /* no open threads at all?!?
 414                                  */
 415                                 ldap_int_thread_ctx_t *ptr;
 416
 417                                 /* let pool_destroy know there are no more threads */
 418                                 ldap_pvt_thread_cond_signal(&pool->ltp_cond);
 419
 420                                 LDAP_STAILQ_FOREACH(ptr, &pool->ltp_pending_list, ltc_next.q)
 421                                         if (ptr == ctx) break;
 422                                 if (ptr == ctx) {
 423                                         /* no open threads, context not handled, so
 424                                          * back out of ltp_pending_count, free the context,
 425                                          * report the error.
 426                                          */
 427                                         LDAP_STAILQ_REMOVE(&pool->ltp_pending_list, ctx,
 428                                                 ldap_int_thread_ctx_s, ltc_next.q);
 429                                         pool->ltp_pending_count--;
 430                                         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 431                                         LDAP_FREE(ctx);
 432                                         return(-1);
 433                                 }
 434                         }
 435                         /* there is another open thread, so this
 436                          * context will be handled eventually.
 437                          * continue on and signal that the context
 438                          * is waiting.
 439                          */
 440                 }
 441                 ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 442         }
 443
 444         return(0);
 445 }
 446
 447 int
 448 ldap_pvt_thread_pool_maxthreads ( ldap_pvt_thread_pool_t *tpool, int max_threads )
 449 {
 450         struct ldap_int_thread_pool_s *pool;
 451
 452         if (tpool == NULL)
 453                 return(-1);
 454
 455         pool = *tpool;
 456
 457         if (pool == NULL)
 458                 return(-1);
 459
 460         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 461         pool->ltp_max_count = max_threads;
 462         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 463         return(0);
 464 }
 465
 466 int
 467 ldap_pvt_thread_pool_query ( ldap_pvt_thread_pool_t *tpool, ldap_pvt_thread_pool_param_t param, void *value )
 468 {
 469         struct ldap_int_thread_pool_s   *pool;
 470         int                             count = -1;
 471
 472         if ( tpool == NULL || value == NULL ) {
 473                 return -1;
 474         }
 475
 476         pool = *tpool;
 477
 478         if ( pool == NULL ) {
 479                 return 0;
 480         }
 481
 482         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 483         switch ( param ) {
 484         case LDAP_PVT_THREAD_POOL_PARAM_MAX:
 485                 count = pool->ltp_max_count;
 486                 break;
 487
 488         case LDAP_PVT_THREAD_POOL_PARAM_MAX_PENDING:
 489                 count = pool->ltp_max_pending;
 490                 break;
 491
 492         case LDAP_PVT_THREAD_POOL_PARAM_OPEN:
 493                 count = pool->ltp_open_count;
 494                 break;
 495
 496         case LDAP_PVT_THREAD_POOL_PARAM_STARTING:
 497                 count = pool->ltp_starting;
 498                 break;
 499
 500         case LDAP_PVT_THREAD_POOL_PARAM_ACTIVE:
 501                 count = pool->ltp_active_count;
 502                 break;
 503
 504         case LDAP_PVT_THREAD_POOL_PARAM_PENDING:
 505                 count = pool->ltp_pending_count;
 506                 break;
 507
 508         case LDAP_PVT_THREAD_POOL_PARAM_BACKLOAD:
 509                 count = pool->ltp_pending_count + pool->ltp_active_count;
 510                 break;
 511
 512         case LDAP_PVT_THREAD_POOL_PARAM_ACTIVE_MAX:
 513                 break;
 514
 515         case LDAP_PVT_THREAD_POOL_PARAM_PENDING_MAX:
 516                 break;
 517
 518         case LDAP_PVT_THREAD_POOL_PARAM_BACKLOAD_MAX:
 519                 break;
 520
 521         case LDAP_PVT_THREAD_POOL_PARAM_STATE: {
 522                 static struct {
 523                         char                            *name;
 524                         ldap_int_thread_pool_state_t    state;
 525                 }               str2state[] = {
 526                         { "running",    LDAP_INT_THREAD_POOL_RUNNING },
 527                         { "finishing",  LDAP_INT_THREAD_POOL_FINISHING },
 528                         { "stopping",   LDAP_INT_THREAD_POOL_STOPPING },
 529                         { NULL }
 530                 };
 531                 int             i;
 532
 533                 if ( pool->ltp_pause ) {
 534                         *((char **)value) = "pausing";
 535                 } else {
 536                         for ( i = 0; str2state[ i ].name != NULL; i++ ) {
 537                                 if ( str2state[ i ].state == pool->ltp_state ) {
 538                                         break;
 539                                 }
 540                         }
 541                         *((char **)value) = str2state[ i ].name;
 542                 }
 543                 if ( *((char **)value) != NULL ) {
 544                         count = -2;
 545                 }
 546                 } break;
 547         }
 548         ldap_pvt_thread_mutex_unlock( &pool->ltp_mutex );
 549
 550         if ( count > -1 ) {
 551                 *((int *)value) = count;
 552         }
 553
 554         return ( count == -1 ? -1 : 0 );
 555 }
 556
 557 /*
 558  * wrapper for ldap_pvt_thread_pool_query(), left around
 559  * for backwards compatibility
 560  */
 561 int
 562 ldap_pvt_thread_pool_backload ( ldap_pvt_thread_pool_t *tpool )
 563 {
 564         int     rc, count;
 565
 566         rc = ldap_pvt_thread_pool_query( tpool,
 567                 LDAP_PVT_THREAD_POOL_PARAM_BACKLOAD, (void *)&count );
 568
 569         if ( rc == 0 ) {
 570                 return count;
 571         }
 572
 573         return rc;
 574 }
 575
 576 int
 577 ldap_pvt_thread_pool_destroy ( ldap_pvt_thread_pool_t *tpool, int run_pending )
 578 {
 579         struct ldap_int_thread_pool_s *pool, *pptr;
 580         ldap_int_thread_ctx_t *ctx;
 581
 582         if (tpool == NULL)
 583                 return(-1);
 584
 585         pool = *tpool;
 586
 587         if (pool == NULL) return(-1);
 588
 589         ldap_pvt_thread_mutex_lock(&ldap_pvt_thread_pool_mutex);
 590         LDAP_STAILQ_FOREACH(pptr, &ldap_int_thread_pool_list, ltp_next)
 591                 if (pptr == pool) break;
 592         if (pptr == pool)
 593                 LDAP_STAILQ_REMOVE(&ldap_int_thread_pool_list, pool,
 594                         ldap_int_thread_pool_s, ltp_next);
 595         ldap_pvt_thread_mutex_unlock(&ldap_pvt_thread_pool_mutex);
 596
 597         if (pool != pptr) return(-1);
 598
 599         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 600         pool->ltp_state = run_pending
 601                 ? LDAP_INT_THREAD_POOL_FINISHING
 602                 : LDAP_INT_THREAD_POOL_STOPPING;
 603
 604         while (pool->ltp_open_count) {
 605                 if (!pool->ltp_pause)
 606                         ldap_pvt_thread_cond_broadcast(&pool->ltp_cond);
 607                 ldap_pvt_thread_cond_wait(&pool->ltp_cond, &pool->ltp_mutex);
 608         }
 609         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 610
 611         while ((ctx = LDAP_STAILQ_FIRST(&pool->ltp_pending_list)) != NULL)
 612         {
 613                 LDAP_STAILQ_REMOVE_HEAD(&pool->ltp_pending_list, ltc_next.q);
 614                 LDAP_FREE(ctx);
 615         }
 616
 617         while ((ctx = LDAP_SLIST_FIRST(&pool->ltp_free_list)) != NULL)
 618         {
 619                 LDAP_SLIST_REMOVE_HEAD(&pool->ltp_free_list, ltc_next.l);
 620                 LDAP_FREE(ctx);
 621         }
 622
 623         ldap_pvt_thread_cond_destroy(&pool->ltp_pcond);
 624         ldap_pvt_thread_cond_destroy(&pool->ltp_cond);
 625         ldap_pvt_thread_mutex_destroy(&pool->ltp_mutex);
 626         LDAP_FREE(pool);
 627 #ifdef LDAP_PVT_THREAD_POOL_SEM_LOAD_CONTROL
 628         if ( thread_pool_sem ) {
 629                 LDAP_FREE( thread_pool_sem );
 630         }
 631 #endif
 632         return(0);
 633 }
 634
 635 static void *
 636 ldap_int_thread_pool_wrapper (
 637         void *xpool )
 638 {
 639         struct ldap_int_thread_pool_s *pool = xpool;
 640         ldap_int_thread_ctx_t *ctx;
 641         ldap_int_thread_userctx_t uctx;
 642         unsigned i, keyslot, hash;
 643
 644         if (pool == NULL)
 645                 return NULL;
 646
 647         for ( i=0; i<MAXKEYS; i++ ) {
 648                 uctx.ltu_key[i].ltk_key = NULL;
 649         }
 650
 651         uctx.ltu_id = ldap_pvt_thread_self();
 652         TID_HASH(uctx.ltu_id, hash);
 653
 654         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 655
 656         /* when paused, thread_keys[] is reserved for pool_purgekey() */
 657         while (pool->ltp_pause)
 658                 ldap_pvt_thread_cond_wait(&pool->ltp_cond, &pool->ltp_mutex);
 659
 660         /* find a key slot to give this thread ID and store a
 661          * pointer to our keys there; start at the thread ID
 662          * itself (mod LDAP_MAXTHR) and look for an empty slot.
 663          */
 664         ldap_pvt_thread_mutex_lock(&ldap_pvt_thread_pool_mutex);
 665         for (keyslot = hash & (LDAP_MAXTHR-1);
 666                 !ldap_pvt_thread_equal(thread_keys[keyslot].id, tid_zero);
 667                 keyslot = (keyslot+1) & (LDAP_MAXTHR-1));
 668         thread_keys[keyslot].id = uctx.ltu_id;
 669         thread_keys[keyslot].ctx = &uctx;
 670         ldap_pvt_thread_mutex_unlock(&ldap_pvt_thread_pool_mutex);
 671
 672         for (;;) {
 673                 while (pool->ltp_pause)
 674                         ldap_pvt_thread_cond_wait(&pool->ltp_cond, &pool->ltp_mutex);
 675
 676                 if (pool->ltp_state == LDAP_INT_THREAD_POOL_STOPPING)
 677                         break;
 678
 679                 ctx = LDAP_STAILQ_FIRST(&pool->ltp_pending_list);
 680                 if (ctx == NULL) {
 681                         if (pool->ltp_state == LDAP_INT_THREAD_POOL_FINISHING)
 682                                 break;
 683
 684                         if (pool->ltp_max_count > 0
 685                                 && pool->ltp_open_count > pool->ltp_max_count)
 686                         {
 687                                 /* too many threads running (can happen if the
 688                                  * maximum threads value is set during ongoing
 689                                  * operation using ldap_pvt_thread_pool_maxthreads)
 690                                  * so let this thread die.
 691                                  */
 692                                 break;
 693                         }
 694
 695                         /* we could check an idle timer here, and let the
 696                          * thread die if it has been inactive for a while.
 697                          * only die if there are other open threads (i.e.,
 698                          * always have at least one thread open).  the check
 699                          * should be like this:
 700                          *   if (pool->ltp_open_count > 1 && pool->ltp_starting == 0)
 701                          *       check timer, wait if ltp_pause, leave thread (break;)
 702                          *
 703                          * Just use pthread_cond_timedwait if we want to
 704                          * check idle time.
 705                          */
 706
 707                         assert(pool->ltp_state == LDAP_INT_THREAD_POOL_RUNNING);
 708                         ldap_pvt_thread_cond_wait(&pool->ltp_cond, &pool->ltp_mutex);
 709                         continue;
 710                 }
 711
 712                 LDAP_STAILQ_REMOVE_HEAD(&pool->ltp_pending_list, ltc_next.q);
 713                 pool->ltp_pending_count--;
 714
 715                 LDAP_SLIST_INSERT_HEAD(&pool->ltp_active_list, ctx, ltc_next.al);
 716                 pool->ltp_active_count++;
 717                 ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 718
 719                 ctx->ltc_start_routine(&uctx, ctx->ltc_arg);
 720
 721 #ifdef LDAP_PVT_THREAD_POOL_SEM_LOAD_CONTROL
 722                 ldap_lazy_sem_post( thread_pool_sem );
 723 #endif
 724                 ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 725                 LDAP_SLIST_REMOVE(&pool->ltp_active_list, ctx,
 726                         ldap_int_thread_ctx_s, ltc_next.al);
 727                 LDAP_SLIST_INSERT_HEAD(&pool->ltp_free_list, ctx, ltc_next.l);
 728                 pool->ltp_active_count--;
 729
 730                 /* let pool_pause know when it is the sole active thread */
 731                 if (pool->ltp_active_count < 2)
 732                         ldap_pvt_thread_cond_signal(&pool->ltp_pcond);
 733         }
 734
 735         ldap_pvt_thread_pool_context_reset(&uctx);
 736
 737         /* Needed if context_reset can let another thread request a pause */
 738         while (pool->ltp_pause)
 739                 ldap_pvt_thread_cond_wait(&pool->ltp_cond, &pool->ltp_mutex);
 740
 741         ldap_pvt_thread_mutex_lock(&ldap_pvt_thread_pool_mutex);
 742         thread_keys[keyslot].ctx = DELETED_THREAD_CTX;
 743         thread_keys[keyslot].id = tid_zero;
 744         ldap_pvt_thread_mutex_unlock(&ldap_pvt_thread_pool_mutex);
 745
 746         pool->ltp_open_count--;
 747
 748         /* let pool_destroy know we're all done */
 749         if (pool->ltp_open_count < 1)
 750                 ldap_pvt_thread_cond_signal(&pool->ltp_cond);
 751
 752         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 753
 754         ldap_pvt_thread_exit(NULL);
 755         return(NULL);
 756 }
 757
 758 int
 759 ldap_pvt_thread_pool_pause (
 760         ldap_pvt_thread_pool_t *tpool )
 761 {
 762         struct ldap_int_thread_pool_s *pool;
 763
 764         if (tpool == NULL)
 765                 return(-1);
 766
 767         pool = *tpool;
 768
 769         if (pool == NULL)
 770                 return(0);
 771
 772         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 773
 774         /* If someone else has already requested a pause, we have to wait */
 775         while (pool->ltp_pause) {
 776                 pool->ltp_pending_count++;
 777                 pool->ltp_active_count--;
 778                 /* let the other pool_pause() know when it can proceed */
 779                 if (pool->ltp_active_count < 2)
 780                         ldap_pvt_thread_cond_signal(&pool->ltp_pcond);
 781                 ldap_pvt_thread_cond_wait(&pool->ltp_cond, &pool->ltp_mutex);
 782                 pool->ltp_pending_count--;
 783                 pool->ltp_active_count++;
 784         }
 785
 786         /* Wait for everyone else to pause or finish */
 787         pool->ltp_pause = 1;
 788         while (pool->ltp_active_count > 1) {
 789                 ldap_pvt_thread_cond_wait(&pool->ltp_pcond, &pool->ltp_mutex);
 790         }
 791
 792         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 793         return(0);
 794 }
 795
 796 int
 797 ldap_pvt_thread_pool_resume (
 798         ldap_pvt_thread_pool_t *tpool )
 799 {
 800         struct ldap_int_thread_pool_s *pool;
 801
 802         if (tpool == NULL)
 803                 return(-1);
 804
 805         pool = *tpool;
 806
 807         if (pool == NULL)
 808                 return(0);
 809
 810         ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
 811         pool->ltp_pause = 0;
 812         ldap_pvt_thread_cond_broadcast(&pool->ltp_cond);
 813         ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
 814         return(0);
 815 }
 816
 817 int ldap_pvt_thread_pool_getkey(
 818         void *xctx,
 819         void *key,
 820         void **data,
 821         ldap_pvt_thread_pool_keyfree_t **kfree )
 822 {
 823         ldap_int_thread_userctx_t *ctx = xctx;
 824         int i;
 825
 826         if ( !ctx || !data ) return EINVAL;
 827
 828         for ( i=0; i<MAXKEYS && ctx->ltu_key[i].ltk_key; i++ ) {
 829                 if ( ctx->ltu_key[i].ltk_key == key ) {
 830                         *data = ctx->ltu_key[i].ltk_data;
 831                         if ( kfree ) *kfree = ctx->ltu_key[i].ltk_free;
 832                         return 0;
 833                 }
 834         }
 835         return ENOENT;
 836 }
 837
 838 int ldap_pvt_thread_pool_setkey(
 839         void *xctx,
 840         void *key,
 841         void *data,
 842         ldap_pvt_thread_pool_keyfree_t *kfree )
 843 {
 844         ldap_int_thread_userctx_t *ctx = xctx;
 845         int i;
 846
 847         if ( !ctx || !key ) return EINVAL;
 848
 849         for ( i=0; i<MAXKEYS; i++ ) {
 850                 if (( data && !ctx->ltu_key[i].ltk_key ) || ctx->ltu_key[i].ltk_key == key ) {
 851                         if ( data || kfree ) {
 852                                 ctx->ltu_key[i].ltk_key = key;
 853                                 ctx->ltu_key[i].ltk_data = data;
 854                                 ctx->ltu_key[i].ltk_free = kfree;
 855                         } else {
 856                                 int j;
 857                                 for ( j=i+1; j<MAXKEYS; j++ )
 858                                         if ( !ctx->ltu_key[j].ltk_key ) break;
 859                                 j--;
 860                                 if ( j != i ) {
 861                                         ctx->ltu_key[i].ltk_key = ctx->ltu_key[j].ltk_key;
 862                                         ctx->ltu_key[i].ltk_data = ctx->ltu_key[j].ltk_data;
 863                                         ctx->ltu_key[i].ltk_free = ctx->ltu_key[j].ltk_free;
 864                                 }
 865                                 ctx->ltu_key[j].ltk_key = NULL;
 866                                 ctx->ltu_key[j].ltk_data = NULL;
 867                                 ctx->ltu_key[j].ltk_free = NULL;
 868                         }
 869                         return 0;
 870                 }
 871         }
 872         return ENOMEM;
 873 }
 874
 875 /* Free all elements with this key, no matter which thread they're in.
 876  * May only be called while the pool is paused.
 877  */
 878 void ldap_pvt_thread_pool_purgekey( void *key )
 879 {
 880         int i, j;
 881         ldap_int_thread_userctx_t *ctx;
 882
 883         for ( i=0; i<LDAP_MAXTHR; i++ ) {
 884                 ctx = thread_keys[i].ctx;
 885                 if ( ctx && ctx != DELETED_THREAD_CTX ) {
 886                         for ( j=0; j<MAXKEYS; j++ ) {
 887                                 if ( ctx->ltu_key[j].ltk_key == key ) {
 888                                         if (ctx->ltu_key[j].ltk_free)
 889                                                 ctx->ltu_key[j].ltk_free( ctx->ltu_key[j].ltk_key,
 890                                                 ctx->ltu_key[j].ltk_data );
 891                                         ctx->ltu_key[j].ltk_key = NULL;
 892                                         ctx->ltu_key[j].ltk_free = NULL;
 893                                         break;
 894                                 }
 895                         }
 896                 }
 897         }
 898 }
 899
 900 /*
 901  * This is necessary if the caller does not have access to the
 902  * thread context handle (for example, a slapd plugin calling
 903  * slapi_search_internal()). No doubt it is more efficient
 904  * for the application to keep track of the thread context
 905  * handles itself.
 906  */
 907 void *ldap_pvt_thread_pool_context( )
 908 {
 909         ldap_pvt_thread_t tid;
 910         unsigned i, hash;
 911         ldap_int_thread_userctx_t *ctx;
 912
 913         tid = ldap_pvt_thread_self();
 914         if ( ldap_pvt_thread_equal( tid, ldap_int_main_tid ))
 915                 return &ldap_int_main_thrctx;
 916
 917         TID_HASH( tid, hash );
 918         i = hash &= (LDAP_MAXTHR-1);
 919         ldap_pvt_thread_mutex_lock(&ldap_pvt_thread_pool_mutex);
 920         do {
 921                 ctx = thread_keys[i].ctx;
 922                 if ( ctx != DELETED_THREAD_CTX )
 923                         if ( ldap_pvt_thread_equal(thread_keys[i].id, tid) || !ctx )
 924                                 goto done;
 925         } while ( (i = (i+1) & (LDAP_MAXTHR-1)) != hash );
 926         ctx = NULL;
 927  done:
 928         ldap_pvt_thread_mutex_unlock(&ldap_pvt_thread_pool_mutex);
 929
 930         return ctx;
 931 }
 932
 933 void ldap_pvt_thread_pool_context_reset( void *vctx )
 934 {
 935         ldap_int_thread_userctx_t *ctx = vctx;
 936         int i;
 937
 938         for ( i=MAXKEYS-1; i>=0; i--) {
 939                 if ( !ctx->ltu_key[i].ltk_key )
 940                         continue;
 941                 if ( ctx->ltu_key[i].ltk_free )
 942                         ctx->ltu_key[i].ltk_free( ctx->ltu_key[i].ltk_key,
 943                         ctx->ltu_key[i].ltk_data );
 944                 ctx->ltu_key[i].ltk_key = NULL;
 945         }
 946 }
 947
 948 ldap_pvt_thread_t ldap_pvt_thread_pool_tid( void *vctx )
 949 {
 950         ldap_int_thread_userctx_t *ctx = vctx;
 951
 952         return ctx->ltu_id;
 953 }
 954 #endif /* LDAP_THREAD_HAVE_TPOOL */