Linux C线程池简单实现

​ 线程池常用于高性能服务器中。为了减小线程创建、销毁时的开销，引入了池式结构。比如：内存池、数据库连接池等等。

学习线程池也只要是因为最近在学网络编程的东西。

线程池结构

从上图可以看到线程池主要包括任务队列、线程数组、管理者线程。为了简单起见，没有加入管理者线程。

调用顺序

//TODO

线程池数据结构设计

线程池结构定义

主要包括线程池的一些属性。线程池状态信息、任务队列信息、互斥锁

struct threadpool_t{
    pthread_mutex_t lock;     //用于内部工作的互斥锁
    pthread_cond_t notify;    //线程间通知的条件变量
    pthread_t *threads;       //线程数组
    threadpool_task_t *queue; //存储任务的数组,即任务队列
    int thread_count;         //线程数量
    int queue_size;           //任务队列大小
    int head;                 //任务队列中首个任务位置（注：任务队列中所有任务都是未开始运行的）
    int tail;                 //任务队列中最后一个任务的下一个位置（注：队列以数组存储，head 和 tail 指示队列位置）
    int count;                //任务队列里的任务数量，即等待运行的任务数
    int shutdown;             //表示线程池是否关闭
    int started;              //开始的线程数   
};

任务队列结构定义

可以放置多种不同任务函数的函数指针，参数为*void**型

typedef struct
{
    void (*function)(void *);	//函数指针，参数为void*。任意线程函数
    void *argument;				//线程函数参数
} threadpool_task_t;

其他一些枚举类型

错误码

typedef enum
{
    threadpool_invalid = -1,
    threadpool_lock_failure = -2,
    threadpool_queue_full = -3,
    threadpool_shutdown = -4,
    threadpool_thread_failure = -5
} threadpool_terror_t;

线程池关闭方式

typedef enum {
    immediate_shutdown = 1, //自动关闭
    graceful_shutdown  = 2	//立即关闭
} threadpool_shutdown_t;

线程池销毁的枚举

typedef enum
{
    threadpool_graceful = 1
} threadpool_destroy_flags_t;

线程池API设计

线程池对外开放了三个API：

/**
 * @function 			threadpool_create
 * @brief	 			创建线程池
 * @param thread_count 	工作线程数量
 * @param thread_queue 	任务队列大小
 * @param flags 	   	未使用的参数
 * @return 				返回一个新创建的线程池或者NULL
**/
threadpool_t *threadpool_create(int thread_count, int queue_size, int flags);

/**
 * @function 			threadpool_add
 * @brief	 			添加新任务到线程池队列中
 * @param pool 			线程池指针
 * @param routine 		函数指针
 * @param arg       	传入函数的参数
 * @param flags 		未使用的参数
 * @return 				成功返回0，失败返回错误码
**/
int threadpool_add(threadpool_t *pool, void (*routine)(void *),void *arg, int flags);

/**
 * @function 			threadpool_destroy
 * @brief	 			停止并且摧毁线程池
 * @param pool 			线程池指针
 * @param flags 		指定关闭方式
**/
int threadpool_destroy(threadpool_t *pool, int flags);

此外还有一个工作线程未对外开放：

static void *threadpool_thread(void *threadpool);

为了避免内存越界：

int threadpool_free(threadpool_t *pool);

创建线程池

创建和运行线程池

threadpool_t *threadpool_create(int thread_count, int queue_size, int flags)
{
    //对传入参数进行判断，MAX_THREADS、MAX_QUEUE为宏定义，自行定义
    if (thread_count <= 0 || thread_count > MAX_THREADS || 
        queue_size <= 0 || queue_size > MAX_QUEUE) 
    {
        return NULL;
    }
	
    //声明线程池变量
    threadpool_t *pool;
	
    //动态申请线程池空间，失败goto err
    if ((pool = (threadpool_t *)malloc(sizeof(threadpool_t))) == NULL)
    {
        goto err;
    }
	
    //对线程池结构体参数进行初始化
    pool->thread_count = 0;
    pool->queue_size = queue_size;
    pool->head = pool->tail = pool->count = 0;
    pool->shutdown = pool->started = 0;
	
    //线程数组动态申请
    pool->threads = (pthread_t *)malloc(sizeof(pthread_t) * thread_count);
    //任务队列动态申请
    pool->queue = (threadpool_task_t *)malloc(sizeof(threadpool_task_t) * queue_size);
	
    //初始化互斥锁、条件变量并判断线程数组、任务队列等是否申请(初始化成功)
    if ((pthread_mutex_init(&pool->lock, NULL) != 0) ||
        (pthread_cond_init(&pool->notify, NULL) != 0) ||
        (pool->threads == NULL) || (pool->queue == NULL))
    {
        goto err;
    }
	
    //创建指定的线程并开始运行
    for (size_t i = 0; i < thread_count; i++)
    {
        if (pthread_create(&pool->threads[i], NULL, threadpool_thread, (void *)pool) != 0)
        {
            threadpool_destroy(pool, 0);
            return NULL;
        }
        pool->thread_count++;	//每成功创建一个线程，工作线程数量++
        pool->started++;		//每运行一个线程，运行中线程数量++
    }
    return pool;

err:
    if (pool)
    {
        threadpool_free(pool);
    }
    return NULL;
}

添加任务到线程池

int threadpool_add(threadpool_t *pool, void (*function)(void *), void *argument, int flags)
{
    int err = 0;
    int next;
	//参数判断
    if (pool == NULL || function == NULL)
    {
        return threadpool_invalid;
    }
	//对资源进行上锁。若上锁失败，会阻塞。
    if (pthread_mutex_lock(&(pool->lock)) != 0)
    {
        return threadpool_lock_failure;
    }
	
    //计算下一个可以存储任务的位置
    next = pool->tail + 1;
    //判断next到没到任务队列的上限，没有就next = next
    next = (next == pool->queue_size) ? 0 : next;
	
    //此处使用do {...}while(0)结构是为了保证至少运行一次。如果有异常，直接break掉，不运行后边的语句
    do
    {	
        //判断线程池中任务数量是否小于任务队列上限
        if (pool->count == pool->queue_size)
        {
            err = threadpool_queue_full;
            break;
        }
		
        //添加任务到任务队列。queue实际大小是0~255，而next是从1开始，1~256
        pool->queue[pool->tail].function = function;
        pool->queue[pool->tail].argument = argument;
		
        //这句就相当于pool->tail++
        pool->tail = next;
        //任务数量++
        pool->count += 1;
		
        //条件变量。任务添加到队列以后，发出信号通知空闲线程取任务。
        if (pthread_cond_signal(&(pool->notify)) != 0)
        {
            err = threadpool_lock_failure;
            break;
        }
    } while (0);
	
    //释放资源
    if (pthread_mutex_unlock(&(pool->lock)) != 0)
    {
        err = threadpool_lock_failure;
    }

    return err;
}

线程池销毁

线程池的销毁比较简单。

int threadpool_destroy(threadpool_t *pool, int flags)
{
    int i, err = 0;

    if (pool == NULL)
        return threadpool_invalid;
    //对资源进行上锁(取得互斥锁资源)
    if (pthread_mutex_lock(&(pool->lock)) != 0)
        return threadpool_lock_failure;
    
	//此处使用do {...}while(0)结构是为了保证至少运行一次。如果有异常，直接break掉，不运行后边的语句
    do
    {
        //判断是否已经关闭了线程池
        if (pool->shutdown)
        {
            err = threadpool_shutdown;
            break;
        }
        
        //指定线程池关闭方式;graceful_shutdown为自动关闭，即等待所有线程完成任务后退出
        //immediate_shutdown，立即退出，不等待
        pool->shutdown = (flags & threadpool_graceful) ? graceful_shutdown : immediate_shutdown;
		
        //唤醒所有因条件变脸阻塞的线程，并释放互斥锁
        if ((pthread_cond_broadcast(&(pool->notify)) != 0) || (pthread_mutex_unlock(&(pool->lock)) != 0))
        {
            err = threadpool_lock_failure;
            break;
        }
		
        //等待子线程退出、并回收线程资源
        for (size_t i = 0; i < pool->thread_count; i++)
        {
            if (pthread_join(pool->threads[i], NULL) != 0)
            {
                err = threadpool_thread_failure;
            }
        }
    } while (0);

    if (!err)
    {
        threadpool_free(pool);
    }

    return err;
}

释放互斥锁、条件变量等资源

int threadpool_free(threadpool_t *pool)
{
    //判断参数是否合法
    if (pool == NULL || pool->started > 0)
    {
        return -1;
    }
	
    //释放互斥锁、条件变量、malloc出的条件队列、线程数组等
    if (pool->threads)
    {
        free(pool->threads);
        free(pool->queue);
		
        //以防万一，加锁后再destroy。不写应该没有啥问题
        pthread_mutex_lock(&(pool->lock));
        pthread_mutex_destroy(&(pool->lock));
        pthread_cond_destroy(&(pool->notify));
    }
    free(pool);
    return 0;
}

工作线程

static void *threadpool_thread(void *threadpool)
{
    threadpool_t *pool = (threadpool_t *)threadpool;
    threadpool_task_t task;

    for (;;)
    {
        /* 取得互斥资源 */
        pthread_mutex_lock(&(pool->lock));
		//使用while是为了在唤醒时重新检查条件
        while ((pool->count == 0) && (!pool->shutdown))
        {
            //任务队列为空，且线程池没有关闭时，阻塞在这里。
            pthread_cond_wait(&(pool->notify), &(pool->lock));
        }
		
        //判断线程池关闭方式，immediate_shutdown时，立即break，并退出
        //graceful_shutdown时，要等待线程池中任务被全部取走，即count == 0时
        if ((pool->shutdown == immediate_shutdown) || ((pool->shutdown == graceful_shutdown) && (pool->count == 0)))
        {
            break;
        }
		
        //取得任务队列第一个任务
        task.function = pool->queue[pool->head].function;
        task.argument = pool->queue[pool->head].argument;
		
        //更新head、cout
        pool->head += 1;
        pool->head = (pool->head == pool->queue_size) ? 0 : pool->head;
        pool->count -= 1;
		
        //释放互斥资源
        pthread_mutex_unlock(&(pool->lock));
		
        //运行任务
        (*(task.function))(task.argument);
    }
	
    //线程将结束、更新运行线程数
    pool->started--;

    pthread_mutex_unlock(&(pool->lock));
    pthread_exit(NULL);
    return NULL;
}

测试用例

#include "threadpool.h"
#include <stdio.h>
#include <pthread.h>
#include <assert.h>
#include <unistd.h>

#define THREAD 32
#define QUEUE 256

int tasks = 0, done = 0;
pthread_mutex_t lock;

void dummy_task(void *arg)
{
    usleep(10000);
    pthread_mutex_lock(&lock);
    /* 记录成功完成的任务数 */
    done++;

    pthread_mutex_unlock(&lock);
}

int main(int argc, char **argv)
{
    threadpool_t *pool;
	
    /* 初始化互斥锁 */
    pthread_mutex_init(&lock, NULL);

    /* 断言线程池创建成功 */
    assert((pool = threadpool_create(THREAD, QUEUE, 0)) != NULL);
    fprintf(stderr, "Pool started with %d threads and queue size of %d\n", THREAD, QUEUE);

    /* 只要任务队列还没满，就一直添加 */
    while (threadpool_add(pool, &dummy_task, NULL, 0) == 0)
    {
        pthread_mutex_lock(&lock);
        tasks++;
        pthread_mutex_unlock(&lock);
    }

    fprintf(stderr, "added %d tasks\n", tasks);
    
    /* 不断检查任务数是否完成一半以上，没有则继续休眠 */
    while ((tasks / 2) > done)
    {
        usleep(10000);
    }
    
	/* 这时候销毁线程池,0 代表 immediate_shutdown 1 代表 graceful_shutdown*/
    assert(threadpool_destroy(pool, 1) == 0);
    fprintf(stderr, "did %d tasks\n", done);

    return 0;
}

pthread库一些函数

创建线程

/**
 * @function 				pthread_create
 * @brief	 				创建线程
 * @param thread	 		线程标识符地址
 * @param attr		 		线程属性结构体地址
 * @param start_routine 	线程函数的入口地址
 * @param arg				传给线程函数的参数
 * @return 					成功 返回0；失败 返回 非0 
**/
int pthread_create(pthread_t *thread,const pthread_attr_t *attr, void *(*start_routine)(void *),void *arg);

初始化互斥锁

int pthread_mutex_init(pthread_mutex_t *__mutex, const pthread_mutexattr_t *__mutexattr)

初始化条件变量

int pthread_cond_init(pthread_cond_t *__restrict__ __cond, const pthread_condattr_t *__restrict__ __cond_attr)

//TODO