2 * Unix SMB/CIFS implementation.
3 * thread pool implementation
4 * Copyright (C) Volker Lendecke 2009
6 * This program is free software; you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License as published by
8 * the Free Software Foundation; either version 3 of the License, or
9 * (at your option) any later version.
11 * This program is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 * GNU General Public License for more details.
16 * You should have received a copy of the GNU General Public License
17 * along with this program. If not, see <http://www.gnu.org/licenses/>.
31 #include "pthreadpool.h"
32 #include "lib/util/dlinklist.h"
34 struct pthreadpool_job
{
35 struct pthreadpool_job
*next
;
37 void (*fn
)(void *private_data
);
43 * List pthreadpools for fork safety
45 struct pthreadpool
*prev
, *next
;
48 * Control access to this struct
50 pthread_mutex_t mutex
;
53 * Threads waiting for work do so here
55 pthread_cond_t condvar
;
60 struct pthreadpool_job
*jobs
, *last_job
;
68 * indicator to worker threads that they should shut down
73 * maximum number of threads
83 * Number of idle threads
88 * An array of threads that require joining, the array has
89 * "max_threads" elements. It contains "num_exited" ids.
92 pthread_t exited
[1]; /* We alloc more */
95 static pthread_mutex_t pthreadpools_mutex
= PTHREAD_MUTEX_INITIALIZER
;
96 static struct pthreadpool
*pthreadpools
= NULL
;
97 static pthread_once_t pthreadpool_atfork_initialized
= PTHREAD_ONCE_INIT
;
99 static void pthreadpool_prep_atfork(void);
102 * Initialize a thread pool
105 int pthreadpool_init(unsigned max_threads
, struct pthreadpool
**presult
)
107 struct pthreadpool
*pool
;
111 size
= sizeof(struct pthreadpool
)
112 + (max_threads
-1) * sizeof(pthread_t
);
114 pool
= (struct pthreadpool
*)malloc(size
);
119 ret
= pipe(pool
->sig_pipe
);
126 ret
= pthread_mutex_init(&pool
->mutex
, NULL
);
132 ret
= pthread_cond_init(&pool
->condvar
, NULL
);
134 pthread_mutex_destroy(&pool
->mutex
);
140 pool
->jobs
= pool
->last_job
= NULL
;
141 pool
->num_threads
= 0;
142 pool
->num_exited
= 0;
143 pool
->max_threads
= max_threads
;
146 ret
= pthread_mutex_lock(&pthreadpools_mutex
);
148 pthread_cond_destroy(&pool
->condvar
);
149 pthread_mutex_destroy(&pool
->mutex
);
153 DLIST_ADD(pthreadpools
, pool
);
155 ret
= pthread_mutex_unlock(&pthreadpools_mutex
);
158 pthread_once(&pthreadpool_atfork_initialized
, pthreadpool_prep_atfork
);
165 static void pthreadpool_prepare(void)
168 struct pthreadpool
*pool
;
170 ret
= pthread_mutex_lock(&pthreadpools_mutex
);
175 while (pool
!= NULL
) {
176 ret
= pthread_mutex_lock(&pool
->mutex
);
182 static void pthreadpool_parent(void)
185 struct pthreadpool
*pool
;
187 pool
= DLIST_TAIL(pthreadpools
);
190 ret
= pthread_mutex_unlock(&pool
->mutex
);
193 if (pool
== pthreadpools
) {
199 ret
= pthread_mutex_unlock(&pthreadpools_mutex
);
203 static void pthreadpool_child(void)
206 struct pthreadpool
*pool
;
208 pool
= DLIST_TAIL(pthreadpools
);
211 close(pool
->sig_pipe
[0]);
212 close(pool
->sig_pipe
[1]);
214 ret
= pipe(pool
->sig_pipe
);
217 pool
->num_threads
= 0;
218 pool
->num_exited
= 0;
221 while (pool
->jobs
!= NULL
) {
222 struct pthreadpool_job
*job
;
224 pool
->jobs
= job
->next
;
227 pool
->last_job
= NULL
;
229 ret
= pthread_mutex_unlock(&pool
->mutex
);
232 if (pool
== pthreadpools
) {
238 ret
= pthread_mutex_unlock(&pthreadpools_mutex
);
242 static void pthreadpool_prep_atfork(void)
244 pthread_atfork(pthreadpool_prepare
, pthreadpool_parent
,
249 * Return the file descriptor which becomes readable when a job has
253 int pthreadpool_sig_fd(struct pthreadpool
*pool
)
255 return pool
->sig_pipe
[0];
259 * Do a pthread_join() on all children that have exited, pool->mutex must be
262 static void pthreadpool_join_children(struct pthreadpool
*pool
)
266 for (i
=0; i
<pool
->num_exited
; i
++) {
267 pthread_join(pool
->exited
[i
], NULL
);
269 pool
->num_exited
= 0;
273 * Fetch a finished job number from the signal pipe
276 int pthreadpool_finished_job(struct pthreadpool
*pool
)
284 while ((nread
== -1) && (errno
== EINTR
)) {
285 nread
= read(pool
->sig_pipe
[0], &result
, sizeof(int));
290 if (nread
!= sizeof(int)) {
297 * Destroy a thread pool, finishing all threads working for it
300 int pthreadpool_destroy(struct pthreadpool
*pool
)
304 ret
= pthread_mutex_lock(&pool
->mutex
);
309 if ((pool
->jobs
!= NULL
) || pool
->shutdown
) {
310 ret
= pthread_mutex_unlock(&pool
->mutex
);
315 if (pool
->num_threads
> 0) {
317 * We have active threads, tell them to finish, wait for that.
322 if (pool
->num_idle
> 0) {
324 * Wake the idle threads. They will find pool->quit to
325 * be set and exit themselves
327 ret
= pthread_cond_broadcast(&pool
->condvar
);
329 pthread_mutex_unlock(&pool
->mutex
);
334 while ((pool
->num_threads
> 0) || (pool
->num_exited
> 0)) {
336 if (pool
->num_exited
> 0) {
337 pthreadpool_join_children(pool
);
341 * A thread that shuts down will also signal
344 ret
= pthread_cond_wait(&pool
->condvar
, &pool
->mutex
);
346 pthread_mutex_unlock(&pool
->mutex
);
352 ret
= pthread_mutex_unlock(&pool
->mutex
);
356 ret
= pthread_mutex_destroy(&pool
->mutex
);
357 ret1
= pthread_cond_destroy(&pool
->condvar
);
366 ret
= pthread_mutex_lock(&pthreadpools_mutex
);
370 DLIST_REMOVE(pthreadpools
, pool
);
371 ret
= pthread_mutex_unlock(&pthreadpools_mutex
);
374 close(pool
->sig_pipe
[0]);
375 pool
->sig_pipe
[0] = -1;
377 close(pool
->sig_pipe
[1]);
378 pool
->sig_pipe
[1] = -1;
386 * Prepare for pthread_exit(), pool->mutex must be locked
388 static void pthreadpool_server_exit(struct pthreadpool
*pool
)
390 pool
->num_threads
-= 1;
391 pool
->exited
[pool
->num_exited
] = pthread_self();
392 pool
->num_exited
+= 1;
395 static void *pthreadpool_server(void *arg
)
397 struct pthreadpool
*pool
= (struct pthreadpool
*)arg
;
400 res
= pthread_mutex_lock(&pool
->mutex
);
408 struct pthreadpool_job
*job
;
411 * idle-wait at most 1 second. If nothing happens in that
412 * time, exit this thread.
415 gettimeofday(&tv
, NULL
);
416 ts
.tv_sec
= tv
.tv_sec
+ 1;
417 ts
.tv_nsec
= tv
.tv_usec
*1000;
419 while ((pool
->jobs
== NULL
) && (pool
->shutdown
== 0)) {
422 res
= pthread_cond_timedwait(
423 &pool
->condvar
, &pool
->mutex
, &ts
);
426 if (res
== ETIMEDOUT
) {
428 if (pool
->jobs
== NULL
) {
430 * we timed out and still no work for
433 pthreadpool_server_exit(pool
);
434 pthread_mutex_unlock(&pool
->mutex
);
449 * Ok, there's work for us to do, remove the job from
450 * the pthreadpool list
452 pool
->jobs
= job
->next
;
453 if (pool
->last_job
== job
) {
454 pool
->last_job
= NULL
;
458 * Do the work with the mutex unlocked
461 res
= pthread_mutex_unlock(&pool
->mutex
);
464 job
->fn(job
->private_data
);
466 written
= write(pool
->sig_pipe
[1], &job
->id
,
471 res
= pthread_mutex_lock(&pool
->mutex
);
474 if (written
!= sizeof(int)) {
475 pthreadpool_server_exit(pool
);
476 pthread_mutex_unlock(&pool
->mutex
);
481 if ((pool
->jobs
== NULL
) && (pool
->shutdown
!= 0)) {
483 * No more work to do and we're asked to shut down, so
486 pthreadpool_server_exit(pool
);
488 if (pool
->num_threads
== 0) {
490 * Ping the main thread waiting for all of us
491 * workers to have quit.
493 pthread_cond_broadcast(&pool
->condvar
);
496 pthread_mutex_unlock(&pool
->mutex
);
502 int pthreadpool_add_job(struct pthreadpool
*pool
, int job_id
,
503 void (*fn
)(void *private_data
), void *private_data
)
505 struct pthreadpool_job
*job
;
508 sigset_t mask
, omask
;
510 job
= (struct pthreadpool_job
*)malloc(sizeof(struct pthreadpool_job
));
516 job
->private_data
= private_data
;
520 res
= pthread_mutex_lock(&pool
->mutex
);
526 if (pool
->shutdown
) {
528 * Protect against the pool being shut down while
529 * trying to add a job
531 res
= pthread_mutex_unlock(&pool
->mutex
);
538 * Just some cleanup under the mutex
540 pthreadpool_join_children(pool
);
543 * Add job to the end of the queue
545 if (pool
->jobs
== NULL
) {
549 pool
->last_job
->next
= job
;
551 pool
->last_job
= job
;
553 if (pool
->num_idle
> 0) {
555 * We have idle threads, wake one.
557 res
= pthread_cond_signal(&pool
->condvar
);
558 pthread_mutex_unlock(&pool
->mutex
);
562 if (pool
->num_threads
>= pool
->max_threads
) {
564 * No more new threads, we just queue the request
566 pthread_mutex_unlock(&pool
->mutex
);
571 * Create a new worker thread. It should not receive any signals.
576 res
= pthread_sigmask(SIG_BLOCK
, &mask
, &omask
);
578 pthread_mutex_unlock(&pool
->mutex
);
582 res
= pthread_create(&thread_id
, NULL
, pthreadpool_server
,
585 pool
->num_threads
+= 1;
588 assert(pthread_sigmask(SIG_SETMASK
, &omask
, NULL
) == 0);
590 pthread_mutex_unlock(&pool
->mutex
);