一、简介

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\n", (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_;