/**

* 线程池

* @author jimmy

* @date 2016-5-14*/#include#include#include#include#include"pd_log.h"#include"pd_log.c"#include"pd_pool.h"

/*tsd*/pthread_key_t key;void *pd_worker_dispatch(void *argv){ushort exit_flag = 0;

pd_task_t*a_task;

pd_pool_t*a_pool = (pd_pool_t *)argv;if(pthread_setspecific(key, (void *)&exit_flag) != 0){returnNULL;

}/*动态从任务列表中获取任务执行*/

while(!exit_flag){

pthread_mutex_lock(&a_pool->mutex);/*如果此时任务链表为空，则需要等待条件变量为真*/

while(a_pool->queue.head ==NULL){

pthread_cond_wait(&a_pool->cond, &a_pool->mutex);

}/*从任务链表中任务开支执行*/a_task= a_pool->queue.head;

a_pool->queue.head = a_task->next;

a_pool->queue.cur_task_num--;if(a_pool->queue.head ==NULL){

a_pool->queue.tail = &a_pool->queue.head;

}/*解锁*/pthread_mutex_unlock(&a_pool->mutex);/*执行任务*/a_task->routine(a_task->argv);//core

free(a_task);

a_task=NULL;

}

pthread_exit(0);

}/**

* 根据线程数创建所有的线程*/

static int pd_pool_create(pd_pool_t *a_pool){inti;

pthread_t tid;for(i = 0; i < a_pool->thread_num; i++){

pthread_create(&tid, NULL, pd_worker_dispatch, a_pool);

}return 0;

}/**

* 线程退出函数*/

void pd_pool_exit_cb(void *argv){

unsignedint *lock =argv;ushort *exit_flag_ptr =pthread_getspecific(key);*exit_flag_ptr = 1;

pthread_setspecific(key, (void *)exit_flag_ptr);*lock = 0;

}/**

* 线程池初始化*/pd_pool_t*pd_pool_init(size_t thread_num, size_t thread_max_num){

pd_pool_t*a_pool =NULL;

a_pool= calloc(1, sizeof(pd_pool_t));if(!a_pool){

error("pool_init calloc fail: %s", strerror(errno));returnNULL;

}

a_pool->thread_num =thread_num;//初始化队列参数

a_pool->queue.max_task_num =thread_max_num;

a_pool->queue.cur_task_num = 0;

a_pool->queue.head =NULL;

a_pool->queue.tail = &a_pool->queue.head;//初始化tsd

if(pthread_key_create(&key, NULL) != 0){

error("pthread_key_create fail: %s", strerror(errno));gotoerr;

}//初始化互斥锁

if(pthread_mutex_init(&a_pool->mutex, NULL) != 0){

error("pthread_mutex_init fail: %s", strerror(errno));

pthread_key_delete(key);gotoerr;

}//初始化条件变量

if(pthread_cond_init(&a_pool->cond, NULL) != 0){

error("pthread_cond_init fail: %s", strerror(errno));

pthread_mutex_destroy(&a_pool->mutex);gotoerr;

}//创建线程池

if(pd_pool_create(a_pool) != 0){

error("pd_pool_create fail: %s", strerror(errno));

pthread_mutex_unlock(&a_pool->mutex);

pthread_cond_destroy(&a_pool->cond);gotoerr;

}returna_pool;

err:free(a_pool);returnNULL;

}/**

* 向线程池中添加任务..*/

int pd_pool_add_task(pd_pool_t *a_pool, void (*routine)(void *), void *argv){

pd_task_t*a_task =NULL;

a_task= (pd_task_t *)calloc(1, sizeof(pd_task_t));if(!a_task){

error("add task calloc faile: %s", strerror(errno));return -1;

}

a_task->routine =routine;

a_task->argv =argv;

a_task->next =NULL;/*加锁*/pthread_mutex_lock(&a_pool->mutex);if(a_pool->queue.cur_task_num >= a_pool->queue.max_task_num){

error("cur_task_num >= max_task_num");gotoerr;

}/*将任务放到末尾*/

*(a_pool->queue.tail) =a_task;

a_pool->queue.tail = &a_task->next;

a_pool->queue.cur_task_num++;/*通知堵塞的线程*/pthread_cond_signal(&a_pool->cond);/*解锁*/pthread_mutex_unlock(&a_pool->mutex);return 0;

err:

pthread_mutex_unlock(&a_pool->mutex);free(a_task);return -1;

}void pd_pool_destroy(pd_pool_t *a_pool){

unsignedintn;

unsignedint lock;for(n = 0; n < a_pool->thread_num; n++){lock = 1;if(pd_pool_add_task(a_pool, pd_pool_exit_cb, &lock) != 0){

error("pd_pool_destroy fail: add_task fail");return;

}while(lock){

usleep(1);

}

}

pthread_mutex_destroy(&a_pool->mutex);

pthread_cond_destroy(&a_pool->cond);

pthread_key_delete(key);free(a_pool);

}/******************************************************************************************/

void testfun(void *argv){

printf("testfun\n");

sleep(1);

}intmain(){

pd_pool_t*a_pool = pd_pool_init(9, 5);

pd_pool_add_task(a_pool, testfun, NULL);

pd_pool_add_task(a_pool, testfun, NULL);

pd_pool_add_task(a_pool, testfun, NULL);

pd_pool_destroy(a_pool);

}