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/>.
29 #include "system/time.h"
30 #include "system/filesys.h"
33 #include "pthreadpool.h"
34 #include "lib/util/dlinklist.h"
36 struct pthreadpool_job
{
37 struct pthreadpool_job
*next
;
39 void (*fn
)(void *private_data
);
45 * List pthreadpools for fork safety
47 struct pthreadpool
*prev
, *next
;
50 * Control access to this struct
52 pthread_mutex_t mutex
;
55 * Threads waiting for work do so here
57 pthread_cond_t condvar
;
62 struct pthreadpool_job
*jobs
, *last_job
;
70 * indicator to worker threads that they should shut down
75 * maximum number of threads
85 * Number of idle threads
90 * An array of threads that require joining.
93 pthread_t
*exited
; /* We alloc more */
96 static pthread_mutex_t pthreadpools_mutex
= PTHREAD_MUTEX_INITIALIZER
;
97 static struct pthreadpool
*pthreadpools
= NULL
;
98 static pthread_once_t pthreadpool_atfork_initialized
= PTHREAD_ONCE_INIT
;
100 static void pthreadpool_prep_atfork(void);
103 * Initialize a thread pool
106 int pthreadpool_init(unsigned max_threads
, struct pthreadpool
**presult
)
108 struct pthreadpool
*pool
;
111 pool
= (struct pthreadpool
*)malloc(sizeof(struct pthreadpool
));
116 ret
= pipe(pool
->sig_pipe
);
123 ret
= pthread_mutex_init(&pool
->mutex
, NULL
);
125 close(pool
->sig_pipe
[0]);
126 close(pool
->sig_pipe
[1]);
131 ret
= pthread_cond_init(&pool
->condvar
, NULL
);
133 pthread_mutex_destroy(&pool
->mutex
);
134 close(pool
->sig_pipe
[0]);
135 close(pool
->sig_pipe
[1]);
141 pool
->jobs
= pool
->last_job
= NULL
;
142 pool
->num_threads
= 0;
143 pool
->num_exited
= 0;
145 pool
->max_threads
= max_threads
;
148 ret
= pthread_mutex_lock(&pthreadpools_mutex
);
150 pthread_cond_destroy(&pool
->condvar
);
151 pthread_mutex_destroy(&pool
->mutex
);
152 close(pool
->sig_pipe
[0]);
153 close(pool
->sig_pipe
[1]);
157 DLIST_ADD(pthreadpools
, pool
);
159 ret
= pthread_mutex_unlock(&pthreadpools_mutex
);
162 pthread_once(&pthreadpool_atfork_initialized
, pthreadpool_prep_atfork
);
169 static void pthreadpool_prepare(void)
172 struct pthreadpool
*pool
;
174 ret
= pthread_mutex_lock(&pthreadpools_mutex
);
179 while (pool
!= NULL
) {
180 ret
= pthread_mutex_lock(&pool
->mutex
);
186 static void pthreadpool_parent(void)
189 struct pthreadpool
*pool
;
191 for (pool
= DLIST_TAIL(pthreadpools
);
193 pool
= DLIST_PREV(pool
)) {
194 ret
= pthread_mutex_unlock(&pool
->mutex
);
198 ret
= pthread_mutex_unlock(&pthreadpools_mutex
);
202 static void pthreadpool_child(void)
205 struct pthreadpool
*pool
;
207 for (pool
= DLIST_TAIL(pthreadpools
);
209 pool
= DLIST_PREV(pool
)) {
211 close(pool
->sig_pipe
[0]);
212 close(pool
->sig_pipe
[1]);
214 ret
= pipe(pool
->sig_pipe
);
217 pool
->num_threads
= 0;
219 pool
->num_exited
= 0;
225 while (pool
->jobs
!= NULL
) {
226 struct pthreadpool_job
*job
;
228 pool
->jobs
= job
->next
;
231 pool
->last_job
= NULL
;
233 ret
= pthread_mutex_unlock(&pool
->mutex
);
237 ret
= pthread_mutex_unlock(&pthreadpools_mutex
);
241 static void pthreadpool_prep_atfork(void)
243 pthread_atfork(pthreadpool_prepare
, pthreadpool_parent
,
248 * Return the file descriptor which becomes readable when a job has
252 int pthreadpool_signal_fd(struct pthreadpool
*pool
)
254 return pool
->sig_pipe
[0];
258 * Do a pthread_join() on all children that have exited, pool->mutex must be
261 static void pthreadpool_join_children(struct pthreadpool
*pool
)
265 for (i
=0; i
<pool
->num_exited
; i
++) {
266 pthread_join(pool
->exited
[i
], NULL
);
268 pool
->num_exited
= 0;
271 * Deliberately not free and NULL pool->exited. That will be
272 * re-used by realloc later.
277 * Fetch a finished job number from the signal pipe
280 int pthreadpool_finished_job(struct pthreadpool
*pool
, int *jobid
)
288 while ((nread
== -1) && (errno
== EINTR
)) {
289 nread
= read(pool
->sig_pipe
[0], &ret_jobid
, sizeof(int));
294 if (nread
!= sizeof(int)) {
302 * Destroy a thread pool, finishing all threads working for it
305 int pthreadpool_destroy(struct pthreadpool
*pool
)
309 ret
= pthread_mutex_lock(&pool
->mutex
);
314 if ((pool
->jobs
!= NULL
) || pool
->shutdown
) {
315 ret
= pthread_mutex_unlock(&pool
->mutex
);
320 if (pool
->num_threads
> 0) {
322 * We have active threads, tell them to finish, wait for that.
327 if (pool
->num_idle
> 0) {
329 * Wake the idle threads. They will find
330 * pool->shutdown to be set and exit themselves
332 ret
= pthread_cond_broadcast(&pool
->condvar
);
334 pthread_mutex_unlock(&pool
->mutex
);
339 while ((pool
->num_threads
> 0) || (pool
->num_exited
> 0)) {
341 if (pool
->num_exited
> 0) {
342 pthreadpool_join_children(pool
);
346 * A thread that shuts down will also signal
349 ret
= pthread_cond_wait(&pool
->condvar
, &pool
->mutex
);
351 pthread_mutex_unlock(&pool
->mutex
);
357 ret
= pthread_mutex_unlock(&pool
->mutex
);
361 ret
= pthread_mutex_destroy(&pool
->mutex
);
362 ret1
= pthread_cond_destroy(&pool
->condvar
);
371 ret
= pthread_mutex_lock(&pthreadpools_mutex
);
375 DLIST_REMOVE(pthreadpools
, pool
);
376 ret
= pthread_mutex_unlock(&pthreadpools_mutex
);
379 close(pool
->sig_pipe
[0]);
380 pool
->sig_pipe
[0] = -1;
382 close(pool
->sig_pipe
[1]);
383 pool
->sig_pipe
[1] = -1;
392 * Prepare for pthread_exit(), pool->mutex must be locked
394 static void pthreadpool_server_exit(struct pthreadpool
*pool
)
398 pool
->num_threads
-= 1;
400 exited
= (pthread_t
*)realloc(
401 pool
->exited
, sizeof(pthread_t
) * (pool
->num_exited
+ 1));
403 if (exited
== NULL
) {
404 /* lost a thread status */
407 pool
->exited
= exited
;
409 pool
->exited
[pool
->num_exited
] = pthread_self();
410 pool
->num_exited
+= 1;
413 static void *pthreadpool_server(void *arg
)
415 struct pthreadpool
*pool
= (struct pthreadpool
*)arg
;
418 res
= pthread_mutex_lock(&pool
->mutex
);
425 struct pthreadpool_job
*job
;
428 * idle-wait at most 1 second. If nothing happens in that
429 * time, exit this thread.
432 clock_gettime(CLOCK_REALTIME
, &ts
);
435 while ((pool
->jobs
== NULL
) && (pool
->shutdown
== 0)) {
438 res
= pthread_cond_timedwait(
439 &pool
->condvar
, &pool
->mutex
, &ts
);
442 if (res
== ETIMEDOUT
) {
444 if (pool
->jobs
== NULL
) {
446 * we timed out and still no work for
449 pthreadpool_server_exit(pool
);
450 pthread_mutex_unlock(&pool
->mutex
);
465 * Ok, there's work for us to do, remove the job from
466 * the pthreadpool list
468 pool
->jobs
= job
->next
;
469 if (pool
->last_job
== job
) {
470 pool
->last_job
= NULL
;
474 * Do the work with the mutex unlocked
477 res
= pthread_mutex_unlock(&pool
->mutex
);
480 job
->fn(job
->private_data
);
482 written
= write(pool
->sig_pipe
[1], &job
->id
,
487 res
= pthread_mutex_lock(&pool
->mutex
);
490 if (written
!= sizeof(int)) {
491 pthreadpool_server_exit(pool
);
492 pthread_mutex_unlock(&pool
->mutex
);
497 if ((pool
->jobs
== NULL
) && (pool
->shutdown
!= 0)) {
499 * No more work to do and we're asked to shut down, so
502 pthreadpool_server_exit(pool
);
504 if (pool
->num_threads
== 0) {
506 * Ping the main thread waiting for all of us
507 * workers to have quit.
509 pthread_cond_broadcast(&pool
->condvar
);
512 pthread_mutex_unlock(&pool
->mutex
);
518 int pthreadpool_add_job(struct pthreadpool
*pool
, int job_id
,
519 void (*fn
)(void *private_data
), void *private_data
)
521 struct pthreadpool_job
*job
;
524 sigset_t mask
, omask
;
526 job
= (struct pthreadpool_job
*)malloc(sizeof(struct pthreadpool_job
));
532 job
->private_data
= private_data
;
536 res
= pthread_mutex_lock(&pool
->mutex
);
542 if (pool
->shutdown
) {
544 * Protect against the pool being shut down while
545 * trying to add a job
547 res
= pthread_mutex_unlock(&pool
->mutex
);
554 * Just some cleanup under the mutex
556 pthreadpool_join_children(pool
);
559 * Add job to the end of the queue
561 if (pool
->jobs
== NULL
) {
565 pool
->last_job
->next
= job
;
567 pool
->last_job
= job
;
569 if (pool
->num_idle
> 0) {
571 * We have idle threads, wake one.
573 res
= pthread_cond_signal(&pool
->condvar
);
574 pthread_mutex_unlock(&pool
->mutex
);
578 if ((pool
->max_threads
!= 0) &&
579 (pool
->num_threads
>= pool
->max_threads
)) {
581 * No more new threads, we just queue the request
583 pthread_mutex_unlock(&pool
->mutex
);
588 * Create a new worker thread. It should not receive any signals.
593 res
= pthread_sigmask(SIG_BLOCK
, &mask
, &omask
);
595 pthread_mutex_unlock(&pool
->mutex
);
599 res
= pthread_create(&thread_id
, NULL
, pthreadpool_server
,
602 pool
->num_threads
+= 1;
605 assert(pthread_sigmask(SIG_SETMASK
, &omask
, NULL
) == 0);
607 pthread_mutex_unlock(&pool
->mutex
);