Linux-C編程 多線(xiàn)程 一個(gè)簡(jiǎn)潔可靠的線(xiàn)程池實(shí)現(xiàn)

哈嘍，我是老吳，我又來(lái)分享學(xué)習(xí)心得了。另外，為了更好地體現(xiàn)公眾號(hào)的核心價(jià)值觀(guān)，從本文開(kāi)始，我會(huì)在文末新增分享一些非技術(shù)相關(guān)的內(nèi)容，歡迎大家參與討論。

一、簡(jiǎn)介

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

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

另外，里面還涉及到了信號(hào)、隊(duì)列、同步等知識(shí)點(diǎn)，代碼讀起來(lái)還是挺過(guò)癮的。

特點(diǎn)：

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

二、使用

快速上手

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

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

API 簡(jiǎn)介

三、內(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)?

正好就是前面說(shuō)過(guò)的 7 個(gè) API，稍后重點(diǎn)分析。

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

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

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

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

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

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

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

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

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

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

//?銷(xiāo)毀?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)?

//?信號(hào)量相關(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è)線(xiàn)程池，先明確線(xiàn)程池的定義：

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

   初始化 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)建用戶(hù)指定數(shù)目的線(xiàn)程，用一個(gè)二級(jí)指針來(lái)指向這一組線(xiàn)程;

   返回 struct thpool_ *;

2. thpool_add_work()

該函數(shù)用于往線(xiàn)程池里添加一個(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ì)列來(lái)管理任務(wù)的，這里的 job 首先是一個(gè)隊(duì)列節(jié)點(diǎn)，攜帶的數(shù)據(jù)是 function + arg。

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

1. 分配 struct job：
   • malloc(sizeof(struct job))

   初始化 struct job;

   • newjob->function=function_p;newjob->arg=arg_p;

   添加到隊(duì)列中：

   • jobqueue_push(&thpool_p->jobqueue, newjob);

3. thpool_pause() 和 thpool_resume()

thpool_pause() 用于暫停所有的線(xiàn)程，通過(guò)信號(hào)機(jī)制來(lái)實(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);
?}
}

給所有工作線(xiàn)程發(fā)送 SIGUSR1，該信號(hào)的處理行為就是讓線(xiàn)程休眠：

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，就可以讓線(xiàn)程返回到原來(lái)被中止時(shí)的工作狀態(tài)。

4. thpool_wait()

wait 的實(shí)現(xiàn)比較簡(jiǎn)單，只要還有任務(wù)或者還有線(xiàn)程處于工作狀態(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);
?}

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

四、測(cè)試用例

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

memleaks.sh：測(cè)試是否發(fā)生內(nèi)存泄露;

threadpool.sh: 測(cè)試線(xiàn)程池是否能正確地執(zhí)行任務(wù);

pause_resume.sh：測(cè)試 pause 和 resume 是否正常;

wait.sh：測(cè)試 wait 功能是否正常;

heap_stack_garbage：測(cè)試堆棧內(nèi)有垃圾數(shù)據(jù)時(shí)的情況;

思考技術(shù)，也思考人生

要學(xué)習(xí)技術(shù)，更要學(xué)習(xí)如何生活。

最近在看的書(shū)：

《精要主義》

點(diǎn)擊查看大圖

收獲了什么？

身體是一種資產(chǎn)：

身體是用來(lái)達(dá)到個(gè)人貢獻(xiàn)峰值的珍貴資產(chǎn)，而最常見(jiàn)用來(lái)破壞這種資產(chǎn)的方式是：缺乏睡眠。

有的人可以很輕易地讓自己拼命工作，并且認(rèn)為自己強(qiáng)大到可以通過(guò)減少睡眠來(lái)盡快實(shí)現(xiàn)自己的目標(biāo)。而事實(shí)是，他們絕大多數(shù)人只是習(xí)慣了疲憊狀態(tài)，以至于已經(jīng)忘記充分休息后的高效學(xué)習(xí)、工作是什么感覺(jué)。

正確的做法是：系統(tǒng)地、有意識(shí)地為睡眠留下一席之地，為每天的生活保留一部分精力和創(chuàng)造力、解決問(wèn)題的能力，以應(yīng)對(duì)一些意外情況。

你和我各有一個(gè)蘋(píng)果，如果我們交換蘋(píng)果的話(huà)，我們還是只有一個(gè)蘋(píng)果。但當(dāng)你和我各有一個(gè)想法，我們交換想法的話(huà)，我們就都有兩個(gè)想法了。

覺(jué)得文章對(duì)你有價(jià)值，不妨點(diǎn)個(gè) 在看和贊。