Linux C线程池简单实现实例

linux  /  管理员 发布于 7年前   163

Linux C线程池

三个文件 

1 tpool.h

typedef struct tpool_work {   void        (*routine)(void *);   void        *arg;   struct tpool_work  *next; } tpool_work_t;  typedef struct tpool {   /* pool characteristics */   int         num_threads;   int         max_queue_size;   /* pool state */   pthread_t      *tpid;   tpool_work_t    *queue;   int         front, rear;   /* 剩下的任务可以做完, 但不能再加新的任务 */   int         queue_closed;     /* 剩下的任务都不做了, 直接关闭 */   int         shutdown;       /* pool synchronization */   pthread_mutex_t   queue_lock;   pthread_cond_t   queue_has_task;   pthread_cond_t   queue_has_space;   pthread_cond_t   queue_empty; } *tpool_t;  void tpool_init(tpool_t *tpoolp,int num_threads, int max_queue_size);  int tpool_add_work(tpool_t tpool,void(*routine)(void *), void *arg);  int tpool_destroy(tpool_t tpool,int finish); 

 2 tpool.c

#include <stdlib.h> #include <stdio.h> #include <unistd.h> #include <sys/types.h> #include <string.h> #include <pthread.h> #include "tpool.h"  #define DEBUG  #if defined(DEBUG) #define debug(...) do { \   flockfile(stdout); \   printf("###%p.%s: ", (void *)pthread_self(), __func__); \   printf(__VA_ARGS__); \   putchar('\n'); \   fflush(stdout); \   funlockfile(stdout); \ } while (0) #else #define debug(...) #endif  void *tpool_thread(void *);  void tpool_init(tpool_t *tpoolp, int num_worker_threads, int max_queue_size) {   int i;   tpool_t pool;    pool = (tpool_t)malloc(sizeof(struct tpool));   if (pool == NULL) {     perror("malloc");     exit(0);   }    pool->num_threads = 0;   pool->max_queue_size = max_queue_size + 1;   pool->num_threads = num_worker_threads;   pool->tpid = NULL;   pool->front = 0;   pool->rear = 0;   pool->queue_closed = 0;   pool->shutdown = 0;    if (pthread_mutex_init(&pool->queue_lock, NULL) == -1) {     perror("pthread_mutex_init");     free(pool);     exit(0);   }   if (pthread_cond_init(&pool->queue_has_space, NULL) == -1) {     perror("pthread_mutex_init");     free(pool);     exit(0);   }   if (pthread_cond_init(&pool->queue_has_task, NULL) == -1) {     perror("pthread_mutex_init");     free(pool);     exit(0);   }   if (pthread_cond_init(&pool->queue_empty, NULL) == -1) {     perror("pthread_mutex_init");     free(pool);     exit(0);   }    if ((pool->queue = malloc(sizeof(struct tpool_work) *pool->max_queue_size)) == NULL) {     perror("malloc");     free(pool);     exit(0);   }    if ((pool->tpid = malloc(sizeof(pthread_t) * num_worker_threads)) == NULL) {     perror("malloc");     free(pool);     free(pool->queue);     exit(0);   }    for (i = 0; i < num_worker_threads; i++) {     if (pthread_create(&pool->tpid[i], NULL, tpool_thread,(void *)pool) != 0) {       perror("pthread_create");       exit(0);     }   }    *tpoolp = pool; }   int empty(tpool_t pool) {   return pool->front == pool->rear; }  int full(tpool_t pool) {   return ((pool->rear + 1) % pool->max_queue_size == pool->front); }  int size(tpool_t pool) {   return (pool->rear + pool->max_queue_size -         pool->front) % pool->max_queue_size; }  int tpool_add_work(tpool_t tpool, void(*routine)(void *), void *arg) {   tpool_work_t *temp;    pthread_mutex_lock(&tpool->queue_lock);    while (full(tpool) && !tpool->shutdown && !tpool->queue_closed) {     pthread_cond_wait(&tpool->queue_has_space, &tpool->queue_lock);   }    if (tpool->shutdown || tpool->queue_closed) {     pthread_mutex_unlock(&tpool->queue_lock);     return -1;   }    int is_empty = empty(tpool);    temp = tpool->queue + tpool->rear;   temp->routine = routine;   temp->arg = arg;   tpool->rear = (tpool->rear + 1) % tpool->max_queue_size;    if (is_empty) {     debug("signal has task");     pthread_cond_broadcast(&tpool->queue_has_task);   }    pthread_mutex_unlock(&tpool->queue_lock);      return 0; }  void *tpool_thread(void *arg) {   tpool_t pool = (tpool_t)(arg);   tpool_work_t *work;    for (;;) {     pthread_mutex_lock(&pool->queue_lock);      while (empty(pool) && !pool->shutdown) {       debug("I'm sleep");       pthread_cond_wait(&pool->queue_has_task, &pool->queue_lock);     }     debug("I'm awake");      if (pool->shutdown == 1) {       debug("exit");       pthread_mutex_unlock(&pool->queue_lock);       pthread_exit(NULL);     }      int is_full = full(pool);     work = pool->queue + pool->front;     pool->front = (pool->front + 1) % pool->max_queue_size;      if (is_full) {       pthread_cond_broadcast(&pool->queue_has_space);     }      if (empty(pool)) {       pthread_cond_signal(&pool->queue_empty);     }      pthread_mutex_unlock(&pool->queue_lock);        (*(work->routine))(work->arg);   } }  int tpool_destroy(tpool_t tpool, int finish) {   int   i;    pthread_mutex_lock(&tpool->queue_lock);    tpool->queue_closed = 1;    if (finish == 1) {     debug("wait all work done");     while (!empty(tpool)) {       pthread_cond_wait(&tpool->queue_empty, &tpool->queue_lock);     }   }   tpool->shutdown = 1;    pthread_mutex_unlock(&tpool->queue_lock);    pthread_cond_broadcast(&tpool->queue_has_task);    debug("wait worker thread exit");   for (i = 0; i < tpool->num_threads; i++) {     pthread_join(tpool->tpid[i], NULL);   }    debug("free thread pool");   free(tpool->tpid);   free(tpool->queue);   free(tpool); }  

3 tpooltest.c

#include <stdio.h> #include <pthread.h> #include "tpool.h"  char *str[]={"string 0", "string 1", "string 2",          "string 3", "string 4", "string 5"};  void job(void * jobstr) {   long i, x;    for (i = 0; i < 100000000; i++) {     x = x +i;   }   printf("%s\n", (char *)jobstr); }  int main(void) {   int i;    tpool_t test_pool;    tpool_init(&test_pool, 8, 20);    for ( i = 0; i < 5; i++) {     tpool_add_work(test_pool, job, str[i]);   }    tpool_destroy(test_pool, 1);    return 0; } 

感谢阅读，希望能帮助到大家，谢谢大家对本站的支持！