哈喽,我是老吴,我又来分享学习心得了。另外,为了更好地体现公众号的核心价值观,从本文开始,我会在文末新增分享一些非技术相关的内容,欢迎大家参与讨论。

 

一、简介

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

 

thpool_init() 的实现思路:

 

  1. 分配 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;  /* function's argument */
} job;

 

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

 

thpool_add_work 的实现思路:

 

  1. 分配 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:测试堆栈内有垃圾数据时的情况;

 

思考技术,也思考人生

要学习技术,更要学习如何生活。

 

最近在看的书:

《精要主义》

 

点击查看大图

 

收获了什么?

身体是一种资产:

 

身体是用来达到个人贡献峰值的珍贵资产,而最常见用来破坏这种资产的方式是:缺乏睡眠。

 

有的人可以很轻易地让自己拼命工作,并且认为自己强大到可以通过减少睡眠来尽快实现自己的目标。而事实是,他们绝大多数人只是习惯了疲惫状态,以至于已经忘记充分休息后的高效学习、工作是什么感觉。

 

正确的做法是:系统地、有意识地为睡眠留下一席之地,为每天的生活保留一部分精力和创造力、解决问题的能力,以应对一些意外情况。

 

你和我各有一个苹果,如果我们交换苹果的话,我们还是只有一个苹果。但当你和我各有一个想法,我们交换想法的话,我们就都有两个想法了。

 

觉得文章对你有价值,不妨点个 在看和赞。