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