一个c语言实现的线程池

/******************************************************** * An example source module to accompany... * * "Using POSIX Threads: Programming with Pthreads" * by Brad nichols, Dick Buttlar, Jackie Farrell * O'Reilly & Associates, Inc. * ******************************************************** * tpool.c -- * * Example thread pooling library */ #include <stdlib.h> #include <stdio.h> #include <unistd.h> #include <sys/types.h> #include <string.h> #include <pthread.h> #include "tpool.h" void *tpool_thread(void *); void tpool_init(tpool_t *tpoolp, int num_worker_threads, int max_queue_size, int do_not_block_when_full) { int i, rtn; tpool_t tpool; /* allocate a pool data structure */ if ((tpool = (tpool_t )malloc(sizeof(struct tpool))) == NULL) perror("malloc"), exit(1); /* initialize th fields */ tpool->num_threads = num_worker_threads; tpool->max_queue_size = max_queue_size; tpool->do_not_block_when_full = do_not_block_when_full; if ((tpool->threads = (pthread_t *)malloc(sizeof(pthread_t)*num_worker_threads)) == NULL) perror("malloc"), exit(1); tpool->cur_queue_size = 0; tpool->queue_head = NULL; tpool->queue_tail = NULL; tpool->queue_closed = 0; tpool->shutdown = 0; if ((rtn = pthread_mutex_init(&(tpool->queue_lock), NULL)) != 0) fprintf(stderr,"pthread_mutex_init %s",strerror(rtn)), exit(1); if ((rtn = pthread_cond_init(&(tpool->queue_not_empty), NULL)) != 0) fprintf(stderr,"pthread_cond_init %s",strerror(rtn)), exit(1); if ((rtn = pthread_cond_init(&(tpool->queue_not_full), NULL)) != 0) fprintf(stderr,"pthread_cond_init %s",strerror(rtn)), exit(1); if ((rtn = pthread_cond_init(&(tpool->queue_empty), NULL)) != 0) fprintf(stderr,"pthread_cond_init %s",strerror(rtn)), exit(1); /* create threads */ for (i = 0; i != num_worker_threads; i++) { if ((rtn = pthread_create( &(tpool->threads[i]), NULL, tpool_thread, (void *)tpool)) != 0) fprintf(stderr,"pthread_create %d",rtn), exit(1); } *tpoolp = tpool; } int tpool_add_work( tpool_t tpool, void (*routine)(void *), void *arg) { int rtn; tpool_work_t *workp; if ((rtn = pthread_mutex_lock(&(tpool->queue_lock))) != 0) fprintf(stderr,"pthread_mutex_lock %d",rtn), exit(1); /* no space and this caller doesn't want to wait */ if ((tpool->cur_queue_size == tpool->max_queue_size) && tpool->do_not_block_when_full) { if ((rtn = pthread_mutex_unlock(&(tpool->queue_lock))) != 0) fprintf(stderr,"pthread_mutex_unlock %d",rtn), exit(1); return -1; } while( (tpool->cur_queue_size == tpool->max_queue_size) && (!(tpool->shutdown || tpool->queue_closed)) ) { if ((rtn = pthread_cond_wait(&(tpool->queue_not_full), &(tpool->queue_lock))) != 0) fprintf(stderr,"pthread_cond_wait %d",rtn), exit(1); } /* the pool is in the process of being destroyed */ if (tpool->shutdown || tpool->queue_closed) { if ((rtn = pthread_mutex_unlock(&(tpool->queue_lock))) != 0) fprintf(stderr,"pthread_mutex_unlock %d",rtn), exit(1); return -1; } /* allocate work structure */ if ((workp = (tpool_work_t *)malloc(sizeof(tpool_work_t))) == NULL) perror("malloc"), exit(1); workp->routine = routine; workp->arg = arg; workp->next = NULL; printf("adder: adding an item %d\n", workp->routine); if (tpool->cur_queue_size == 0) { tpool->queue_tail = tpool->queue_head = workp; printf("adder: queue == 0, waking all workers\n"); if ((rtn = pthread_cond_broadcast(&(tpool->queue_not_empty))) != 0) fprintf(stderr,"pthread_cond_signal %d",rtn), exit(1);; } else { tpool->queue_tail->next = workp; tpool->queue_tail = workp; } tpool->cur_queue_size++; if ((rtn = pthread_mutex_unlock(&(tpool->queue_lock))) != 0) fprintf(stderr,"pthread_mutex_unlock %d",rtn), exit(1); return 1; } int tpool_destroy(tpool_t tpool, int finish) { int i,rtn; tpool_work_t *cur_nodep; if ((rtn = pthread_mutex_lock(&(tpool->queue_lock))) != 0) fprintf(stderr,"pthread_mutex_lock %d",rtn), exit(1); /* Is a shutdown already in progress? */ if (tpool->queue_closed || tpool->shutdown) { if ((rtn = pthread_mutex_unlock(&(tpool->queue_lock))) != 0) fprintf(stderr,"pthread_mutex_unlock %d",rtn), exit(1); return 0; } tpool->queue_closed = 1; /* If the finish flag is set, wait for workers to drain queue */ if (finish == 1) { while (tpool->cur_queue_size != 0) { if ((rtn = pthread_cond_wait(&(tpool->queue_empty), &(tpool->queue_lock))) != 0) fprintf(stderr,"pthread_cond_wait %d",rtn), exit(1); } } tpool->shutdown = 1; if ((rtn = pthread_mutex_unlock(&(tpool->queue_lock))) != 0) fprintf(stderr,"pthread_mutex_unlock %d",rtn), exit(1); /* Wake up any workers so they recheck shutdown flag */ if ((rtn = pthread_cond_broadcast(&(tpool->queue_not_empty))) != 0) fprintf(stderr,"pthread_cond_broadcast %d",rtn), exit(1); if ((rtn = pthread_cond_broadcast(&(tpool->queue_not_full))) != 0) fprintf(stderr,"pthread_cond_broadcast %d",rtn), exit(1); /* Wait for workers to exit */ for(i=0; i < tpool->num_threads; i++) { if ((rtn = pthread_join(tpool->threads[i],NULL)) != 0) fprintf(stderr,"pthread_join %d",rtn), exit(1); } /* Now free pool structures */ free(tpool->threads); while(tpool->queue_head != NULL) { cur_nodep = tpool->queue_head->next; tpool->queue_head = tpool->queue_head->next; free(cur_nodep); } free(tpool); } void *tpool_thread(void *arg) { tpool_t tpool = (tpool_t)arg; int rtn; tpool_work_t *my_workp; for(;;) { /* Check queue for work */ if ((rtn = pthread_mutex_lock(&(tpool->queue_lock))) != 0) fprintf(stderr,"pthread_mutex_lock %d",rtn), exit(1); while ((tpool->cur_queue_size == 0) && (!tpool->shutdown)) { printf("worker %d: I'm sleeping again\n", pthread_self()); if ((rtn = pthread_cond_wait(&(tpool->queue_not_empty), &(tpool->queue_lock))) != 0) fprintf(stderr,"pthread_cond_wait %d",rtn), exit(1); } sleep(5); printf("worker %d: I'm awake\n", pthread_self()); /* Has a shutdown started while i was sleeping? */ if (tpool->shutdown == 1) { if ((rtn = pthread_mutex_unlock(&(tpool->queue_lock))) != 0) fprintf(stderr,"pthread_mutex_unlock %d",rtn), exit(1); pthread_exit(NULL); } /* Get to work, dequeue the next item */ my_workp = tpool->queue_head; tpool->cur_queue_size--; if (tpool->cur_queue_size == 0) tpool->queue_head = tpool->queue_tail = NULL; else tpool->queue_head = my_workp->next; printf("worker %d: dequeing item %d\n", pthread_self(), my_workp->next); /* Handle waiting add_work threads */ if ((!tpool->do_not_block_when_full) && (tpool->cur_queue_size == (tpool->max_queue_size - 1))) if ((rtn = pthread_cond_broadcast(&(tpool->queue_not_full))) != 0) fprintf(stderr,"pthread_cond_broadcast %d",rtn), exit(1); /* Handle waiting destroyer threads */ if (tpool->cur_queue_size == 0) if ((rtn = pthread_cond_signal(&(tpool->queue_empty))) != 0) fprintf(stderr,"pthread_cond_signal %d",rtn), exit(1); if ((rtn = pthread_mutex_unlock(&(tpool->queue_lock))) != 0) fprintf(stderr,"pthread_mutex_unlock %d",rtn), exit(1); /* Do this work item */ (*(my_workp->routine))(my_workp->arg); free(my_workp); } return(NULL); }