Linux线程池的实现

Posted 2020-11-06 长风破浪会有时，直挂云帆济沧海

• 线程池的实现
  • 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\n",(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\n");
                      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\n",(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\n",(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 }