https://github.com/Pithikos/C-Thread-Pool
这是一个简单小巧的C语言线程池实现,在 Github 上有 1.1K 的 star,很适合用来学习 Linux 的多线程编程。
另外,里面还涉及到了信号、队列、同步等知识点,代码读起来还是挺过瘾的。
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;示例 | 作用 |
---|---|
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。
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)
该函数用于创建一个线程池,先明确线程池的定义:
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() 的实现思路:
该函数用于往线程池里添加一个任务,先明确任务的定义:
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 的实现思路:
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,就可以让线程返回到原来被中止时的工作状态。
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);
}
到此,我感觉已经没有太多难点了,感兴趣的小伙伴们可以自行查阅源码。
优秀的开源项目通常会附带丰富的测试用例,此项目也不例外:
全部0条评论
快来发表一下你的评论吧 !