线程池

1 while (pool->cur_queue_size == 0)
2 {
3       pthread_cond_wait (&(pool->queue_ready),&(pool->queue_lock));
4 }

  1 #include <stdio.h>
  2 #include <stdlib.h>
  3 #include <unistd.h>
  4 #include <sys/types.h>
  5 #include <pthread.h>
  6 #include <assert.h>
  7  
  8 /*
  9 *线程池里所有运行和等待的任务都是一个CThread_worker
 10 *由于所有任务都在链表里，所以是一个链表结构
 11 */
 12 typedef struct worker
 13 {
 14     /*回调函数，任务运行时会调用此函数，注意也可声明成其它形式*/
 15     void *(*process) (void *arg);
 16     void *arg;/*回调函数的参数*/
 17     struct worker *next;
 18  
 19 } CThread_worker;
 20 
 21  
 22  
 23 /*线程池结构*/
 24 typedef struct
 25 {
 26     pthread_mutex_t queue_lock;
 27     pthread_cond_t queue_ready;
 28  
 29     /*链表结构，线程池中所有等待任务*/
 30     CThread_worker *queue_head;
 31  
 32     /*是否销毁线程池*/
 33     int shutdown;
 34     pthread_t *threadid;
 35     /*线程池中允许的活动线程数目*/
 36     int max_thread_num;
 37     /*当前等待队列的任务数目*/
 38     int cur_queue_size;
 39  
 40 } CThread_pool;
 41 
 42  
 43  
 44 int pool_add_worker (void *(*process) (void *arg), void *arg);
 45 void *thread_routine (void *arg);
 46 
 47  
 48  
 49 static CThread_pool *pool = NULL;
 50 void
 51 pool_init (int max_thread_num)
 52 {
 53     pool = (CThread_pool *) malloc (sizeof (CThread_pool));
 54  
 55     pthread_mutex_init (&(pool->queue_lock), NULL);
 56     pthread_cond_init (&(pool->queue_ready), NULL);
 57  
 58     pool->queue_head = NULL;
 59  
 60     pool->max_thread_num = max_thread_num;
 61     pool->cur_queue_size = 0;
 62  
 63     pool->shutdown = 0;
 64  
 65     pool->threadid =
 66         (pthread_t *) malloc (max_thread_num * sizeof (pthread_t));
 67     int i = 0;
 68     for (i = 0; i < max_thread_num; i++)
 69     {
 70         pthread_create (&(pool->threadid[i]), NULL, thread_routine,
 71                 NULL);
 72     }
 73 }
 74 
 75  
 76  
 77 /*向线程池中加入任务*/
 78 int
 79 pool_add_worker (void *(*process) (void *arg), void *arg)
 80 {
 81     /*构造一个新任务*/
 82     CThread_worker *newworker =
 83         (CThread_worker *) malloc (sizeof (CThread_worker));
 84     newworker->process = process;
 85     newworker->arg = arg;
 86     newworker->next = NULL;/*别忘置空*/
 87  
 88     pthread_mutex_lock (&(pool->queue_lock));
 89     /*将任务加入到等待队列中*/
 90     CThread_worker *member = pool->queue_head;
 91     if (member != NULL)
 92     {
 93         while (member->next != NULL)
 94             member = member->next;
 95         member->next = newworker;
 96     }
 97     else
 98     {
 99         pool->queue_head = newworker;
100     }
101  
102     assert (pool->queue_head != NULL);
103  
104     pool->cur_queue_size++;
105     pthread_mutex_unlock (&(pool->queue_lock));
106     /*好了，等待队列中有任务了，唤醒一个等待线程；
107     注意如果所有线程都在忙碌，这句没有任何作用*/
108     pthread_cond_signal (&(pool->queue_ready));
109     return 0;
110 }
111 
112  
113  
114 /*销毁线程池，等待队列中的任务不会再被执行，但是正在运行的线程会一直
115 把任务运行完后再退出*/
116 int
117 pool_destroy ()
118 {
119     if (pool->shutdown)
120         return -1;/*防止两次调用*/
121     pool->shutdown = 1;
122  
123     /*唤醒所有等待线程，线程池要销毁了*/
124     pthread_cond_broadcast (&(pool->queue_ready));
125  
126     /*阻塞等待线程退出，否则就成僵尸了*/
127     int i;
128     for (i = 0; i < pool->max_thread_num; i++)
129         pthread_join (pool->threadid[i], NULL);
130     free (pool->threadid);
131  
132     /*销毁等待队列*/
133     CThread_worker *head = NULL;
134     while (pool->queue_head != NULL)
135     {
136         head = pool->queue_head;
137         pool->queue_head = pool->queue_head->next;
138         free (head);
139     }
140     /*条件变量和互斥量也别忘了销毁*/
141     pthread_mutex_destroy(&(pool->queue_lock));
142     pthread_cond_destroy(&(pool->queue_ready));
143    
144     free (pool);
145     /*销毁后指针置空是个好习惯*/
146     pool=NULL;
147     return 0;
148 }
149 
150  
151  
152 void *
153 thread_routine (void *arg)
154 {
155     printf ("starting thread 0x%x/n", pthread_self ());
156     while (1)
157     {
158         pthread_mutex_lock (&(pool->queue_lock));
159         /*如果等待队列为0并且不销毁线程池，则处于阻塞状态; 注意
160         pthread_cond_wait是一个原子操作，等待前会解锁，唤醒后会加锁*/
161         while (pool->cur_queue_size == 0 && !pool->shutdown)
162         {
163             printf ("thread 0x%x is waiting/n", pthread_self ());
164             pthread_cond_wait (&(pool->queue_ready), &(pool->queue_lock));
165         }
166  
167         /*线程池要销毁了*/
168         if (pool->shutdown)
169         {
170             /*遇到break,continue,return等跳转语句，千万不要忘记先解锁*/
171             pthread_mutex_unlock (&(pool->queue_lock));
172             printf ("thread 0x%x will exit/n", pthread_self ());
173             pthread_exit (NULL);
174         }
175  
176         printf ("thread 0x%x is starting to work/n", pthread_self ());
177  
178         /*assert是调试的好帮手*/
179         assert (pool->cur_queue_size != 0);
180         assert (pool->queue_head != NULL);
181        
182         /*等待队列长度减去1，并取出链表中的头元素*/
183         pool->cur_queue_size--;
184         CThread_worker *worker = pool->queue_head;
185         pool->queue_head = worker->next;
186         pthread_mutex_unlock (&(pool->queue_lock));
187  
188         /*调用回调函数，执行任务*/
189         (*(worker->process)) (worker->arg);
190         free (worker);
191         worker = NULL;
192     }
193     /*这一句应该是不可达的*/
194     pthread_exit (NULL);
195 }

 1 void *
 2 myprocess (void *arg)
 3 {
 4     printf ("threadid is 0x%x, working on task %d/n", pthread_self (),*(int *) arg);
 5     sleep (1);/*休息一秒，延长任务的执行时间*/
 6     return NULL;
 7 }
 8  
 9 int
10 main (int argc, char **argv)
11 {
12     pool_init (3);/*线程池中最多三个活动线程*/
13    
14     /*连续向池中投入10个任务*/
15     int *workingnum = (int *) malloc (sizeof (int) * 10);
16     int i;
17     for (i = 0; i < 10; i++)
18     {
19         workingnum[i] = i;
20         pool_add_worker (myprocess, &workingnum[i]);
21     }
22     /*等待所有任务完成*/
23     sleep (5);
24     /*销毁线程池*/
25     pool_destroy ();
26  
27     free (workingnum);
28     return 0;
29 }