add thread pool implemention

Author: selfboot

Coding/Linux_OS_ThreadPool.h

@@ -0,0 +1,159 @@
+#ifndef THREAD_POOL_H
+#define THREAD_POOL_H
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <pthread.h>
+#include <assert.h>
+
+
+// 线程池里所有运行和等待的任务都是一个CThread_worker, 由于所有任务都在链表里，所以是一个链表结构
+typedef struct worker
+{
+
+    void *(*process) (void *arg);   // 回调函数，任务运行时会调用此函数，注意也可声明成其它形式
+    void *arg;                      // 回调函数的参数
+    struct worker *next;
+} CThread_worker;
+
+// 线程池结构
+typedef struct
+{
+    pthread_mutex_t queue_lock;
+    pthread_cond_t queue_ready;
+    CThread_worker *queue_head;   // 链表结构，线程池中所有等待任务
+    bool shutdown;                // 是否销毁线程池
+    pthread_t *threadid;          // 维持线程链表
+    int max_thread_num;           // 线程池中允许的活动线程数目
+    int cur_queue_size;           // 当前等待队列的任务数目
+} CThread_pool;
+
+int pool_add_worker (void *(*process) (void *arg), void *arg);
+void *thread_routine (void *arg);
+static CThread_pool *pool = NULL;
+
+void pool_init (int max_thread_num)
+{
+    pool = (CThread_pool *) malloc (sizeof (CThread_pool));
+    pthread_mutex_init (&(pool->queue_lock), NULL);
+    pthread_cond_init (&(pool->queue_ready), NULL);
+    pool->queue_head = NULL;
+    pool->max_thread_num = max_thread_num;
+    pool->cur_queue_size = 0;
+    pool->shutdown = false;
+    pool->threadid =
+            (pthread_t *) malloc (max_thread_num * sizeof (pthread_t));
+    int i = 0;
+    for (i = 0; i < max_thread_num; i++)
+    {
+        pthread_create (&(pool->threadid[i]), NULL, thread_routine,
+                        NULL);
+    }
+}
+
+//向线程池中加入任务
+int pool_add_worker (void *(*process) (void *arg), void *arg)
+{
+    // 构造一个新任务
+    CThread_worker *new_worker =
+            (CThread_worker *) malloc (sizeof (CThread_worker));
+    new_worker->process = process;
+    new_worker->arg = arg;
+    new_worker->next = NULL;        // new_worker 为最后一个任务
+    pthread_mutex_lock (&(pool->queue_lock));
+
+    // 将任务加入到等待队列中
+    CThread_worker *member = pool->queue_head;
+    if (member != NULL)
+    {
+        while (member->next != NULL)
+            member = member->next;
+        member->next = new_worker;
+    }
+    else
+    {
+        pool->queue_head = new_worker;
+    }
+    assert (pool->queue_head != NULL);
+    pool->cur_queue_size++;
+    pthread_mutex_unlock (&(pool->queue_lock));
+    // 等待队列中有任务了，唤醒一个等待线程.  注意如果所有线程都在忙碌，这句没有任何作用
+    pthread_cond_signal(&(pool->queue_ready));
+    return 0;
+}
+
+/*销毁线程池，等待队列中的任务不会再被执行，但是正在运行的线程会一直
+把任务运行完后再退出*/
+int pool_destroy ()
+{
+    if (pool->shutdown)
+        return -1;          // 防止两次调用
+    pool->shutdown = true;
+    pthread_cond_broadcast (&(pool->queue_ready));   // 唤醒所有等待线程，线程池要销毁了
+    // 阻塞等待线程退出，否则就成僵尸了
+    int i;
+    for (i = 0; i < pool->max_thread_num; i++)
+        pthread_join (pool->threadid[i], NULL);
+    free (pool->threadid);
+
+    // 销毁等待任务队列
+    CThread_worker *head = NULL;
+    while (pool->queue_head != NULL)
+    {
+        head = pool->queue_head;
+        pool->queue_head = pool->queue_head->next;
+        free (head);
+    }
+    // 销毁条件变量和互斥量
+    pthread_mutex_destroy(&(pool->queue_lock));
+    pthread_cond_destroy(&(pool->queue_ready));
+
+    free (pool);
+    // 销毁后指针置空
+    pool=NULL;
+    return 0;
+}
+
+void *thread_routine (void *arg)
+{
+    printf ("starting thread 0x%x\n", pthread_self());
+    while (1)
+    {
+        pthread_mutex_lock(&(pool->queue_lock));
+        /*
+         * 如果等待队列为0并且不销毁线程池，则处于阻塞状态;
+         * 注意pthread_cond_wait是一个原子操作，等待前会解锁，唤醒后会加锁
+         */
+        while (pool->cur_queue_size == 0 && !pool->shutdown)
+        {
+            printf ("thread 0x%x is waiting\n", pthread_self ());
+            pthread_cond_wait(&(pool->queue_ready), &(pool->queue_lock));
+        }
+        // 将要销毁线程池
+        if (pool->shutdown)
+        {
+            // 遇到break,continue,return等跳转语句，千万不要忘记先解锁
+            pthread_mutex_unlock(&(pool->queue_lock));
+            printf ("thread 0x%x is exiting\n", pthread_self());
+            pthread_exit(NULL);
+        }
+        printf ("thread 0x%x is starting to work\n", pthread_self());
+        // assert是调试的好帮手
+        assert (pool->cur_queue_size != 0);
+        assert (pool->queue_head != NULL);
+
+        /*等待队列长度减去1，并取出链表中的头元素*/
+        pool->cur_queue_size--;
+        CThread_worker *worker = pool->queue_head;
+        pool->queue_head = worker->next;
+        pthread_mutex_unlock (&(pool->queue_lock));
+        // 调用回调函数，执行任务
+        (*(worker->process)) (worker->arg);
+        free (worker);
+        worker = NULL;
+    }
+}
+
+#endif

Coding/Linux_OS_ThreadPoolTest.cpp

@@ -0,0 +1,32 @@
+#include <iostream>
+#include "Linux_OS_ThreadPool.h"
+using namespace std;
+
+void *my_task (void *arg)
+{
+    printf ("threadid is 0x%x, working on task %d\n", pthread_self(),*(int *) arg);
+    sleep (3);              //休息3秒，延长任务的执行时间
+    return NULL;
+}
+
+int main (int argc, char **argv)
+{
+    pool_init(3);           //线程池中最多三个活动线程
+
+    int job_num = 5;
+    // 每隔一定时间, 向池中投入任务, 一共 5 个任务
+    int *jobs = (int *) malloc (sizeof (int) * job_num);
+    int i;
+    for (i = 0; i < job_num; i++)
+    {
+        jobs[i] = i;
+        sleep(i);
+        pool_add_worker(my_task, &jobs[i]);
+    }
+    // 等待所有任务完成
+    sleep(10);
+    // 销毁线程池
+    pool_destroy();
+    free(jobs);
+    return 0;
+}

Linux_OS/ThreadPools.md

@@ -53,14 +53,19 @@ while(1)
 
 当然，如果线程创建和销毁时间相比任务执行时间可以忽略不计，则没有必要使用线程池了。以上面例子来说，如果一个医生每天只能看一个病人，那么就没有必要坐诊了。
 
-要实现一个可伸缩的线程池，那么至少要包括：**n个执行任务的线程，一个任务队列，一个管理线程**，然后按照下面的方式工作：
+要实现一个简单的线程池，只需要设计**n个执行任务的线程，一个任务队列**，然后按照下面的方式工作：
 
 1. 预先启动一些线程，线程负责执行任务队列中的任务，当队列空时，线程挂起。
 2. 调用的时候，直接往任务队列添加任务，并发信号通知线程队列非空。
-3. 管理线程负责监控任务队列和系统中的线程状态，当任务队列为空，线程数目多且很多处于空闲的时候，便通知一些线程退出以节约系统资源；当任务队列排队任务多且线程都在忙，便负责再多启动一些线程来执行任务，以确保任务执行效率。
 
+要实现一个可伸缩的线程池，还需要添加一个管理线程，来负责监控任务队列和系统中的线程状态，当任务队列为空，线程数目多且很多处于空闲的时候，便通知一些线程退出以节约系统资源；当任务队列排队任务多且线程都在忙，便负责再多启动一些线程来执行任务，以确保任务执行效率。
 
+[Linux_OS_ThreadPool](../Coding/Linux_OS_ThreadPool.h) 是一个简单的 C 实现的线程池的例子，[Test.cpp](../Coding/Linux_OS_ThreadPoolTest.cpp) 是一个简单的测试例子， 其中：
 
+* pool_add_worker()：线程池的任务链表中加入一个任务，加入后通过调用 pthread_cond_signal(&(pool->queue_ready)) 唤醒一个出于阻塞状态的线程(如果有的话)。
+* pool_destroy()：销毁线程池，线程池任务链表中的任务不会再被执行，但是正在运行的线程会一直把任务运行完后再退出。
+
+一个更加完善的 C 实现可以在 [C-Thread-Pool](https://github.com/Pithikos/C-Thread-Pool) 找到。
 
 # 更多阅读