C语言线程池的实现方案-c语言pthread线程池

一、简介

https://github.com/Pithikos/C-Thread-Pool

这是一个简单小巧的C语言线程池实现，在 Github 上有 1.1K 的 star，很适合用来学习 Linux 的多线程编程。

另外，里面还涉及到了信号、队列、同步等知识点，代码读起来还是挺过瘾的。

特点：

符合 ANCI C and POSIX; 支持暂停/恢复/等待功能; 简洁的 API; 经过严格的测试，附带了丰富的测试用例; 二、使用

快速上手

example.c：

复制#include “thpool.h” void task(void *arg){ printf(“Thread #%u working on %d “, (int)pthread_self(), (int) arg); } int main(){ puts(“Making threadpool with 4 threads”); threadpool thpool = thpool_init(4); puts(“Adding 10 tasks to threadpool”); int i; for (i=0; i<8; i++){ thpool_add_work(thpool, task, (void*)(uintptr_t)i); }; thpool_wait(thpool); puts(“Killing threadpool”); thpool_destroy(thpool); return 0; }

运行效果：

复制$ gcc example.c thpool.c -D THPOOL_DEBUG -pthread -o example $ ./example Making threadpool with 4 threads THPOOL_DEBUG: Created thread 0 in pool THPOOL_DEBUG: Created thread 1 in pool THPOOL_DEBUG: Created thread 2 in pool THPOOL_DEBUG: Created thread 3 in pool Adding 10 tasks to threadpool Thread #1509455616 working on 0 Thread #1509455616 working on 4 Thread #1509455616 working on 5 Thread #1492670208 working on 2 Thread #1492670208 working on 7 Thread #1509455616 working on 6 Thread #1501062912 working on 1 Thread #1517848320 working on 3 Killing threadpool

代码分析：

复制threadpool thpool = thpool_init(4) 创建了一个含有 4 个线程的线程池; 然后调用 复制thpool_add_work(thpool, …) 往线程池里放入了 8 个任务; 从结果来看： 复制线程5616 抢到了任务 0 / 4 / 5 / 6; 复制线程0208 抢到了任务 2 / 7; 复制线程2919 抢到了任务 1; 复制线程8320 抢到了任务 3;

API 简介

示例 作用 thpool_init(4) 创建一个含有 4 个线程的线程池。 * thpool_add_work(thpool, (void)function_p, (void )arg_p)* ** 添加任务， function_p 是任务要执行的函数，arg_p 是 function_p 的参数。 thpool_wait(thpool) 等待所有任务完成。 thpool_destroy(thpool) 销毁线程池，如果还有任务在执行，则会先等待其完成。 thpool_pause(thpool) 让所有的线程都停止工作，进入睡眠状态。 thpool_resume(thpool) 让所有的线程都恢复工作。 thpool_num_threads_working(thpool) 返回当前正在工作的线程数。 三、内部实现

整体把握

核心代码就是 2 个文件：thpool.c 和 thpool.h。

分解 thpool.c

7 个公共函数：

复制struct thpool_* thpool_init(int num_threads) int thpool_add_work(thpool_* thpool_p, void (*function_p)(void*), void* arg_p) void thpool_wait(thpool_* thpool_p) void thpool_destroy(thpool_* thpool_p) void thpool_pause(thpool_* thpool_p) void thpool_resume(thpool_* thpool_p) int thpool_num_threads_working(thpool_* thpool_p)

正好就是前面说过的 7 个 API，稍后重点分析。

5 个自定义的数据结构：

复制// 描述一个信号量 typedef struct bsem {…} bsem; // 描述一个任务 typedef struct job {…} job; // 描述一个任务队列 typedef struct jobqueue {…} jobqueue; // 描述一个线程 typedef struct thread {…} thread; // 描述一个线程池 typedef struct thpool_ {…} thpool_;

14 个私有函数：

复制// 构造 struct thread，并调用 pthread_create() 创建线程 static int thread_init (thpool_* thpool_p, struct thread** thread_p, int id) // 当线程被暂停时会在这里休眠 static void thread_hold(int sig_id) // 线程在此函数中执行任务 static void* thread_do(struct thread* thread_p) // 销毁 struct thread static void thread_destroy (thread* thread_p) // 任务队列相关的操作集合 static int jobqueue_init(jobqueue* jobqueue_p) static void jobqueue_clear(jobqueue* jobqueue_p) static void jobqueue_push(jobqueue* jobqueue_p, struct job* newjob) static struct job* jobqueue_pull(jobqueue* jobqueue_p) static void jobqueue_destroy(jobqueue* jobqueue_p) // 信号量相关的操作集合 static void bsem_init(bsem *bsem_p, int value) static void bsem_reset(bsem *bsem_p) static void bsem_post(bsem *bsem_p) static void bsem_post_all(bsem *bsem_p) static void bsem_wait(bsem* bsem_p)

核心 API 的实现

1. thpool_init()

该函数用于创建一个线程池，先明确线程池的定义：

复制typedef struct thpool_{ thread** threads; /* pointer to threads */ volatile int num_threads_alive; /* threads currently alive */ volatile int num_threads_working; /* threads currently working */ pthread_mutex_t thcount_lock; /* used for thread count etc */ pthread_cond_t threads_all_idle; /* signal to thpool_wait */ jobqueue jobqueue; /* job queue */ } thpool_;

thpool_init() 的实现思路：

分配 struct thpool_： malloc(sizeof(struct thpool_)) 初始化 struct thpool_; malloc(num_threads * sizeof(struct thread *)) thread_init(thpool_p, &thpool_p->threads[n], n); jobqueue_init(&thpool_p->jobqueue) 初始化 jobqueue： 创建用户指定数目的线程，用一个二级指针来指向这一组线程; 返回 struct thpool_ *;

2. thpool_add_work()

该函数用于往线程池里添加一个任务，先明确任务的定义：

复制typedef struct job{ struct job* prev; /* pointer to previous job */ void (*function)(void* arg); /* function pointer */ void* arg; /* functions argument */ } job;

程序里是用队列来管理任务的，这里的 job 首先是一个队列节点，携带的数据是 function + arg。

thpool_add_work 的实现思路：

分配 struct job： malloc(sizeof(struct job)) 初始化 struct job; newjob->function=function_p; newjob->arg=arg_p; 添加到队列中： jobqueue_push(&thpool_p->jobqueue, newjob);

3. thpool_pause() 和 thpool_resume()

thpool_pause() 用于暂停所有的线程，通过信号机制来实现：

复制void thpool_pause(thpool_* thpool_p) { int n; for (n=0; n < thpool_p->num_threads_alive; n++){ pthread_kill(thpool_p->threads[n]->pthread, SIGUSR1); } }

给所有工作线程发送 复制SIGUSR1，该信号的处理行为就是让线程休眠：

复制static void thread_hold(int sig_id) { (void)sig_id; threads_on_hold = 1; while (threads_on_hold){ sleep(1); } }

只需要 thpool_resume() 中，将 threads_on_hold = 0，就可以让线程返回到原来被中止时的工作状态。

4. thpool_wait()

wait 的实现比较简单，只要还有任务或者还有线程处于工作状态，就执行 pthread 的 wait 操作：

复制while (thpool_p->jobqueue.len || thpool_p->num_threads_working) { pthread_cond_wait(&thpool_p->threads_all_idle, &thpool_p->thcount_lock); }

到此，我感觉已经没有太多难点了，感兴趣的小伙伴们可以自行查阅源码。

四、测试用例

优秀的开源项目通常会附带丰富的测试用例，此项目也不例外：

memleaks.sh：测试是否发生内存泄露; threadpool.sh: 测试线程池是否能正确地执行任务; pause_resume.sh：测试 pause 和 resume 是否正常; wait.sh：测试 wait 功能是否正常; heap_stack_garbage：测试堆栈内有垃圾数据时的情况;