- 线程池的实现
- 1：自定义封装的条件变量
-
1 //condition.h 2 #ifndef _CONDITION_H_ 3 #define _CONDITION_H_ 4 5 #include <pthread.h> 6 7 typedef struct condition 8 { 9 pthread_mutex_t pmutex; 10 pthread_cond_t pcond; 11 }condition_t; 12 13 int condition_init(condition_t *cond); 14 int condition_lock(condition_t *cond); 15 int condition_unlock(condition_t *cond); 16 int condition_wait(condition_t *cond); 17 int condition_timewait(condition_t *cond,const struct timespec *abstime); 18 int condition_signal(condition_t *cond); 19 int condition_broadcast(condition_t *cond); 20 int condition_destroy(condition_t *cond); 21 22 #endif
-
1 //condition.c 2 #include "condition.h" 3 4 int condition_init(condition_t *cond) 5 { 6 int status; 7 if((status = pthread_mutex_init(&cond->pmutex,NULL))) 8 return status; 9 if((status = pthread_cond_init(&cond->pcond,NULL))) 10 return status; 11 return 0; 12 } 13 14 int condition_lock(condition_t *cond) 15 { 16 return pthread_mutex_lock(&cond->pmutex); 17 } 18 int condition_unlock(condition_t *cond) 19 { 20 return pthread_mutex_unlock(&cond->pmutex); 21 } 22 int condition_wait(condition_t *cond) 23 { 24 return pthread_cond_wait(&cond->pcond,&cond->pmutex); 25 } 26 int condition_timewait(condition_t *cond,const struct timespec *abstime) 27 { 28 return pthread_cond_timedwait(&cond->pcond,&cond->pmutex,abstime); 29 } 30 int condition_signal(condition_t *cond) 31 { 32 return pthread_cond_signal(&cond->pcond); 33 } 34 int condition_broadcast(condition_t *cond) 35 { 36 return pthread_cond_broadcast(&cond->pcond); 37 } 38 int condition_destroy(condition_t *cond) 39 { 40 int status; 41 if((status = pthread_mutex_destroy(&cond->pmutex))) 42 return status; 43 if((status = pthread_cond_destroy(&cond->pcond))) 44 return status; 45 return 0; 46 47 }
-
- 2:线程池逻辑
-
1 //threadpool.h 2 #include "condition.h" 3 4 typedef struct task 5 { 6 void *(*run)(void *arg); 7 void *arg; 8 struct task *next; 9 }task_t; 10 11 typedef struct threadpool 12 { 13 condition_t ready; 14 task_t *first; 15 task_t *last; 16 int counter; 17 int idle; 18 int max_threads; 19 int quit; 20 }threadpool_t; 21 22 void threadpool_init(threadpool_t *pool, int threads); 23 24 void threadpool_add_task(threadpool_t *pool,void *(*run)(void *arg),void *arg); 25 26 void threadpool_destroy(threadpool_t *pool);
-
//threadpool.c #include "threadpool.h" #include <string.h> #include <errno.h> #include <time.h> void *thread_routine(void *arg) { printf("thread 0x%0x is starting ",(int)pthread_self()); struct timespec abstime; int timeout; threadpool_t *pool = (threadpool_t*)arg; while(1) { timeout = 0; condition_lock(&pool->ready); pool->idle++; while(pool->first == NULL && pool->quit) { // condition_wait(&pool->ready); clock_gettime(CLOCK_REALTIME, &abstime); abstime.tv_sec += 2; int status = condition_timewait(&pool->ready,&abstime); if(status == ETIMEDOUT) { printf("thread time out "); timeout = 1; break; } } pool->idle--; if(pool->first != NULL) { task_t *t = pool->first; pool->first = t->next; //由于执行下线程工作需要时间，所以先解锁以让其他线程能过获取资源 condition_unlock(&pool->ready); t->run(t->arg); free(t); condition_lock(&pool->ready); } if(pool->quit && pool->first == NULL) { pool->counter--; if(pool->counter == 0) { condition_signal(&pool->ready); } condition_unlock(&pool->ready); break; } if(timeout && pool->first == NULL) { pool->counter--; condition_unlock(&pool->ready); break; } condition_unlock(&pool->ready); } printf("thead 0x%0x is exting ",(int)pthread_self()); return NULL; } void threadpool_init(threadpool_t *pool, int threads) { // condition_t ready; // task_t *first; // task_t *last; // int counter; // int idle; // int max_threads; // int quit; condition_init(&pool->ready); pool->first = NULL; pool->last = NULL; pool->counter= 0; pool->idle = 0; pool->max_threads = threads; pool->quit = 0; } void threadpool_add_task(threadpool_t *pool,void *(*run)(void *arg),void *arg) { task_t* newtask = malloc(sizeof(task_t)); newtask->run = run; newtask->arg = arg; newtask->next = NULL; condition_lock(&pool->ready); //add task to tasklist tail if(pool->first ==NULL) pool->first = newtask; else pool->last->next = newtask; pool->last = newtask; // if have wait thread,wake to do work if(pool->idle > 0) { condition_signal(&pool->ready); } else if(pool->counter < pool->max_threads) { pthread_t tid; pthread_create(&tid, NULL, thread_routine, pool); pool->counter++; } condition_unlock(&pool->ready); } void threadpool_destroy(threadpool_t *pool) { if(pool->quit) { return; } condition_lock(&pool->ready); pool->quit = 1; if(pool->counter > 0) { if(pool->idle > 0) { condition_broadcast(&pool->ready); } while(pool->counter > 0) { condition_wait(&pool->ready); } } condition_unlock(&pool->ready); condition_destroy(&pool->ready); }
-
-
3:main
-
1 #include "threadpool.h" 2 #include <unistd.h> 3 #include <stdio.h> 4 #include <stdlib.h> 5 6 void* mytask(void* arg) 7 { 8 printf("thread 0x%0x is working on task %d ",(int)pthread_self(),*(int*)arg); 9 sleep(1); 10 free(arg); 11 return NULL; 12 } 13 14 int main(void) 15 { 16 threadpool_t pool; 17 threadpool_init(&pool,3); 18 19 int i; 20 for(i = 0; i < 10; i++) 21 { 22 int *arg = (int*)malloc(sizeof(int)); 23 *arg = i; 24 threadpool_add_task(&pool,mytask,arg); 25 } 26 // sleep(15); 27 threadpool_destroy(&pool); 28 return 0; 29 }
-
- 1：自定义封装的条件变量