tpool.c 37 KB

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