一、簡介

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

這是一個(gè)簡單小巧的C語言線程池實(shí)現(xiàn)，在 Github 上有 1.1K 的 star，很適合用來學(xué)習(xí) Linux 的多線程編程。

另外，里面還涉及到了信號、隊(duì)列、同步等知識點(diǎn)，代碼讀起來還是挺過癮的。

特點(diǎn)：

• 符合 ANCI C and POSIX;
• 支持暫停/恢復(fù)/等待功能;
• 簡潔的 API;
• 經(jīng)過嚴(yán)格的測試，附帶了豐富的測試用例;

二、使用

快速上手

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;
}

運(yùn)行效果：

$ 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) 創(chuàng)建了一個(gè)含有 4 個(gè)線程的線程池;
• 然后調(diào)用 thpool_add_work(thpool, ...) 往線程池里放入了 8 個(gè)任務(wù);
• 從結(jié)果來看：
  • 線程5616 搶到了任務(wù) 0 / 4 / 5 / 6;
  • 線程0208 搶到了任務(wù) 2 / 7;
  • 線程2919 搶到了任務(wù) 1;
  • 線程8320 搶到了任務(wù) 3;

API 簡介

三、內(nèi)部實(shí)現(xiàn)

整體把握

核心代碼就是 2 個(gè)文件：thpool.c 和 thpool.h。

分解 thpool.c

7 個(gè)公共函數(shù)：

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 個(gè) API，稍后重點(diǎn)分析。

5 個(gè)自定義的數(shù)據(jù)結(jié)構(gòu)：

// 描述一個(gè)信號量
typedef struct bsem {...} bsem;

// 描述一個(gè)任務(wù)
typedef struct job {...} job;

// 描述一個(gè)任務(wù)隊(duì)列
typedef struct jobqueue {...} jobqueue;

// 描述一個(gè)線程
typedef struct thread {...} thread;

// 描述一個(gè)線程池
typedef struct thpool_ {...} thpool_;

14 個(gè)私有函數(shù)：

// 構(gòu)造 struct thread，并調(diào)用 pthread_create() 創(chuàng)建線程
static int thread_init (thpool_* thpool_p, struct thread** thread_p, int id) 

// 當(dāng)線程被暫停時(shí)會(huì)在這里休眠
static void thread_hold(int sig_id) 

// 線程在此函數(shù)中執(zhí)行任務(wù)
static void* thread_do(struct thread* thread_p) 

// 銷毀 struct thread
static void thread_destroy (thread* thread_p) 

// 任務(wù)隊(duì)列相關(guān)的操作集合
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) 

// 信號量相關(guān)的操作集合
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 的實(shí)現(xiàn)

1. thpool_init()

該函數(shù)用于創(chuàng)建一個(gè)線程池，先明確線程池的定義：

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() 的實(shí)現(xiàn)思路：

1. 分配 struct thpool_：
   • malloc(sizeof(struct thpool_))
2. 初始化 struct thpool_;
   • malloc(num_threads * sizeof(struct thread *))
   • thread_init(thpool_p, &thpool_p->threads[n], n);
   • jobqueue_init(&thpool_p->jobqueue)
   • 初始化 jobqueue：
   • 創(chuàng)建用戶指定數(shù)目的線程，用一個(gè)二級指針來指向這一組線程;
3. 返回 struct thpool_ *;

2. thpool_add_work()

該函數(shù)用于往線程池里添加一個(gè)任務(wù)，先明確任務(wù)的定義：

typedef struct job{
 struct job*  prev; /* pointer to previous job */
 void   (*function)(void* arg);  /* function pointer */
 void*  arg;  /* function's argument */
} job;

程序里是用隊(duì)列來管理任務(wù)的，這里的 job 首先是一個(gè)隊(duì)列節(jié)點(diǎn)，攜帶的數(shù)據(jù)是 function + arg。

thpool_add_work 的實(shí)現(xiàn)思路：

1. 分配 struct job：
   • malloc(sizeof(struct job))
2. 初始化 struct job;
   • newjob->function=function_p;
   • newjob->arg=arg_p;
3. 添加到隊(duì)列中：
   • jobqueue_push(&thpool_p->jobqueue, newjob);

3. thpool_pause() 和 thpool_resume()

thpool_pause() 用于暫停所有的線程，通過信號機(jī)制來實(shí)現(xiàn)：

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);
 }
}

給所有工作線程發(fā)送 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，就可以讓線程返回到原來被中止時(shí)的工作狀態(tài)。

4. thpool_wait()

wait 的實(shí)現(xiàn)比較簡單，只要還有任務(wù)或者還有線程處于工作狀態(tài)，就執(zhí)行 pthread 的 wait 操作：

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

到此，我感覺已經(jīng)沒有太多難點(diǎn)了，感興趣的小伙伴們可以自行查閱源碼。

四、測試用例

優(yōu)秀的開源項(xiàng)目通常會(huì)附帶豐富的測試用例，此項(xiàng)目也不例外：

• memleaks.sh：測試是否發(fā)生內(nèi)存泄露;
• threadpool.sh: 測試線程池是否能正確地執(zhí)行任務(wù);
• pause_resume.sh：測試 pause 和 resume 是否正常;
• wait.sh：測試 wait 功能是否正常;
• heap_stack_garbage：測試堆棧內(nèi)有垃圾數(shù)據(jù)時(shí)的情況;