libco协程库源码解读
2019獨角獸企業(yè)重金招聘Python工程師標(biāo)準(zhǔn)>>>
????協(xié)程,又被稱為用戶級線程,是在應(yīng)用層被調(diào)度,可以減少因為調(diào)用系統(tǒng)調(diào)用而阻塞的線程切換的時間.目前有很多協(xié)程的實現(xiàn),由于微信內(nèi)部大量使用了其直研的的libco協(xié)程庫，所以我選擇了騰訊開源的libco協(xié)程庫進(jìn)行研究,學(xué)習(xí)協(xié)程的基本思想.
1，基本原理
? ? 協(xié)程實質(zhì)上可以看成是子程序、函數(shù)。一個線程上面可以運行多個協(xié)程，但是同一時間只能運行一個協(xié)程，協(xié)程在線程上的切換，是由于遇到阻塞的操作，或者主動讓出線程使用權(quán)。比如，有10個協(xié)程，當(dāng)前線程正在運行協(xié)程1，然后協(xié)程1執(zhí)行一個recv的阻塞操作，協(xié)程的調(diào)度器能夠檢測到這個操作，將協(xié)程1切換出去，將協(xié)程2調(diào)度進(jìn)來執(zhí)行。如果沒有協(xié)程的調(diào)度器，此時協(xié)程1將會由于調(diào)用recv這個系統(tǒng)調(diào)用且數(shù)據(jù)未到達(dá)而阻塞，進(jìn)行休眠，此時操作系統(tǒng)將會發(fā)生線程切換，調(diào)度其他線程執(zhí)行，而線程切換非常耗時，高達(dá)幾十微秒（同事測試是20us），即便新執(zhí)行的線程是用戶任務(wù)相關(guān)的，用戶任務(wù)也會多了幾十微秒的線程切換的消耗。而如果使用協(xié)程，協(xié)程之間的切換只需要幾百納秒（同事測試為0.35us，即350納秒），耗時很少。這就是協(xié)程發(fā)揮優(yōu)勢的地方。
? ? 下面講解libco的源碼部分，有一篇文章：C++開源協(xié)程庫libco-原理與應(yīng)用.pdf，非常深入的講解了libco的原理，而且不枯燥，十分推薦讀者先看看這篇文章。
? ? 由于libco是非對稱的協(xié)程機(jī)制，如果從當(dāng)前協(xié)程A切換到協(xié)程B，而協(xié)程B又沒有切換到下一個協(xié)程，在協(xié)程B執(zhí)行結(jié)束之后，會返回到協(xié)程A執(zhí)行。
2，libco基本框架
? ? libco中的基本框架如下（引自C/C++協(xié)程庫libco：微信怎樣漂亮地完成異步化改造）：
協(xié)程接口層實現(xiàn)了協(xié)程的基本源語。co_create、co_resume等簡單接口負(fù)責(zé)協(xié)程創(chuàng)建于恢復(fù)。co_cond_signal類接口可以在協(xié)程間創(chuàng)建一個協(xié)程信號量，可用于協(xié)程間的同步通信。
系統(tǒng)函數(shù)Hook層負(fù)責(zé)主要負(fù)責(zé)系統(tǒng)中同步API到異步執(zhí)行的轉(zhuǎn)換。對于常用的同步網(wǎng)絡(luò)接口，Hook層會把本次網(wǎng)絡(luò)請求注冊為異步事件，然后等待事件驅(qū)動層的喚醒執(zhí)行。
事件驅(qū)動層實現(xiàn)了一個簡單高效的異步網(wǎng)路框架，里面包含了異步網(wǎng)絡(luò)框架所需要的事件與超時回調(diào)。對于來源于同步系統(tǒng)函數(shù)Hook層的請求，事件注冊與回調(diào)實質(zhì)上是協(xié)程的讓出與恢復(fù)執(zhí)行。
本文通過講解接口層的幾個主要函數(shù)，使讀者對libco協(xié)程的框架和原理有一個大概的認(rèn)識，下一篇文章將會講解libco如何處理事件循環(huán)等。
下面我們從幾個主要的協(xié)程函數(shù)一一分析。
3，主要函數(shù)源碼解析
- co_create?????首先來開一下協(xié)程創(chuàng)建的函數(shù)，源碼如下：
????? ? co_create()的第一行判斷是當(dāng)前線程初始化環(huán)境變量的判斷，如果沒進(jìn)行環(huán)境初始化，那么調(diào)用co_init_curr_thread_env() 進(jìn)行環(huán)境初始化，會生成當(dāng)前環(huán)境g_arrCoEnvPerThread[ GetPid() ]的第一個協(xié)程 env->pCallStack，其?cIsMain 標(biāo)志位 1，iCallStackSize表示協(xié)程層數(shù)，目前只有1層，AllocEpoll()函數(shù)中初始化當(dāng)前環(huán)境env的 pstActiveList，pstTimeoutList 這兩個列表，這兩個列表分別記錄了活動協(xié)程和超時協(xié)程。環(huán)境初始化操作在一個線程中只會進(jìn)行一次。在初始化完成之后，會調(diào)用co_create_env()創(chuàng)建一個新的協(xié)程，新協(xié)程的結(jié)構(gòu)體中的env這個域始終指向當(dāng)前協(xié)程環(huán)境g_arrCoEnvPerThread[ GetPid() ]。新協(xié)程創(chuàng)建之后，并沒有做什么操作。
- co_resume void co_resume( stCoRoutine_t *co ) {stCoRoutineEnv_t *env = co->env;stCoRoutine_t *lpCurrRoutine = env->pCallStack[ env->iCallStackSize - 1 ];if( !co->cStart ){coctx_make( &co->ctx,(coctx_pfn_t)CoRoutineFunc,co,0 );co->cStart = 1;}env->pCallStack[ env->iCallStackSize++ ] = co;co_swap( lpCurrRoutine, co ); } co_resume()函數(shù)是切換協(xié)程的函數(shù)，也可以稱為是啟動協(xié)程的函數(shù)。co_resume()函數(shù)的第一行是獲取當(dāng)前線程的協(xié)程環(huán)境env，第二行獲取當(dāng)前正在執(zhí)行的協(xié)程，也即馬上要被切換出去的協(xié)程。接下來判斷待切換的協(xié)程co是否已經(jīng)被切換過，如果沒有，那么為co準(zhǔn)備上下文,cStart字段設(shè)置為1。這里為co準(zhǔn)備的上下文，就是在coctx_make()函數(shù)里面,這個函數(shù)將函數(shù)指針CoRoutineFunc賦值給co->ctx的reg[0],將來上下文切換的時候,就能切換到reg[0]所指向的地址去執(zhí)行.準(zhǔn)備好co的上下文之后,然后將待切換的協(xié)程co入棧，置于協(xié)程環(huán)境env的協(xié)程棧的頂端，表明當(dāng)前最新的協(xié)程是co。注意，這并不是說協(xié)程棧中只有棧頂才是co，可能棧中某些位置也存了co。最后，調(diào)用co_swap()，該函數(shù)將協(xié)程上下文環(huán)境切換為co的上下文環(huán)境，并進(jìn)入co指定的函數(shù)內(nèi)執(zhí)行,之前被切換出去的協(xié)程被掛起,直到co主動yield,讓出cpu,才會恢復(fù)被切換出去的協(xié)程執(zhí)行.注意,這里的所有的協(xié)程都是在當(dāng)前協(xié)程執(zhí)行的,也就是說,所有的協(xié)程都是串行執(zhí)行的,調(diào)用co_resume()之后,執(zhí)行上下文就跳到co的代碼空間中去了。因為co_swap()要等co主動讓出cpu才會返回，而co的協(xié)程內(nèi)部可能會resume新的協(xié)程繼續(xù)執(zhí)行下去，所以co_swap()函數(shù)調(diào)用可能要等到很長時間才能返回。 void co_swap(stCoRoutine_t* curr, stCoRoutine_t* pending_co) {stCoRoutineEnv_t* env = co_get_curr_thread_env();//get curr stack spchar c;curr->stack_sp= &c;if (!pending_co->cIsShareStack){env->pending_co = NULL;env->occupy_co = NULL;}else {env->pending_co = pending_co;//get last occupy co on the same stack memstCoRoutine_t* occupy_co = pending_co->stack_mem->occupy_co;//set pending co to occupy thest stack mem;pending_co->stack_mem->occupy_co = pending_co;env->occupy_co = occupy_co;if (occupy_co && occupy_co != pending_co){save_stack_buffer(occupy_co);}}//swap contextcoctx_swap(&(curr->ctx),&(pending_co->ctx) );//stack buffer may be overwrite, so get again;stCoRoutineEnv_t* curr_env = co_get_curr_thread_env();stCoRoutine_t* update_occupy_co = curr_env->occupy_co;stCoRoutine_t* update_pending_co = curr_env->pending_co;if (update_occupy_co && update_pending_co && update_occupy_co != update_pending_co){//resume stack bufferif (update_pending_co->save_buffer && update_pending_co->save_size > 0){memcpy(update_pending_co->stack_sp, update_pending_co->save_buffer, update_pending_co->save_size);}} } 在co_swap()函數(shù)代碼中，由于libco不是共享棧的模式，即pending_co->cIsShareStack為0，所以執(zhí)行了if分支，接下來執(zhí)行coctx_swap()，這是一段匯編源碼，內(nèi)容就是從curr的上下文跳轉(zhuǎn)到pending_co的上下文中執(zhí)行,通過回調(diào)CoRoutineFunc()函數(shù)實現(xiàn)，此時當(dāng)前線程的cpu已經(jīng)開始執(zhí)行pending_co協(xié)程中的代碼,直到pending_co主動讓出cpu,才接著執(zhí)行coctx_swap()下面的代碼,由于update_occupy_co為NULL,下面的if語句沒有執(zhí)行,所以相當(dāng)于coctx_swap()下面沒有代碼,直接返回到curr協(xié)程中.
- co_yield
co_yield()與co_yield_ct()的功能是一樣的,都是使得當(dāng)前協(xié)程讓出cpu. void co_yield_env( stCoRoutineEnv_t *env ) {stCoRoutine_t *last = env->pCallStack[ env->iCallStackSize - 2 ];stCoRoutine_t *curr = env->pCallStack[ env->iCallStackSize - 1 ];env->iCallStackSize--;co_swap( curr, last); } co_yield_env()函數(shù)中的第二行獲取當(dāng)前執(zhí)行的協(xié)程,也即當(dāng)前協(xié)程環(huán)境的協(xié)程棧的棧頂,函數(shù)的第一行獲取協(xié)程棧的次頂,也即上一次被切換的協(xié)程last,從這里也可以看出,libco的協(xié)程讓出cpu,只能讓給上一次被切換出去的協(xié)程.最后一行是co_swap()函數(shù),前面講到,該函數(shù)會進(jìn)入last協(xié)程的上下文去執(zhí)行代碼,也就是回到上次co_resume()函數(shù)內(nèi)部的co_swap()的地方,繼續(xù)往下走.
當(dāng)協(xié)程正常結(jié)束的時候,會繼續(xù)執(zhí)行CoRoutineFunc()函數(shù),將協(xié)程的cEnd設(shè)置為1,表示已經(jīng)結(jié)束,并執(zhí)行一次co_yield_env(),讓出cpu,切換回上一次被讓出的協(xié)程繼續(xù)執(zhí)行.
這里有一點我之前不太理解,懷疑會發(fā)生棧溢出的地方,那就是在調(diào)用co_yield_env()，進(jìn)入co_swap()之后，調(diào)用coctx_swap()，切換到上一次的last協(xié)程的上下文，那么當(dāng)前協(xié)程的co_swap()函數(shù)里面的變量，都是在?？臻g上面的，切換到last協(xié)程的上下文之后，那些變量依然在?？臻g上面，不會被銷毀，直到回到了main函數(shù)的協(xié)程，還是沒有被銷毀。其實這是個誤區(qū)，這些變量其實不是在棧空間上面，而是在CPU的通用寄存器里面，當(dāng)調(diào)用coctx_swap()之后，這些寄存器變量就會保存到當(dāng)前協(xié)程的?？臻g中去，其實是我們之前co_create()函數(shù)malloc出來的一片堆空間。這是因為cpu的工作寄存器數(shù)量較多，而局部變量較少，而co_swap()函數(shù)的變量都是局部變量，直接存放在cpu的工作寄存器中，而coctx_swap()的作用就是將CPU的各個通用寄存器保存到coctx_t結(jié)構(gòu)的regs[1] ~ regs[6]的位置，然后將last協(xié)程的coctx_t結(jié)構(gòu)的regs[1]~regs[6]的內(nèi)容加載到當(dāng)前的通用寄存器中，并將執(zhí)行cpu的執(zhí)行順序切換到last協(xié)程中去執(zhí)行。 - co_release
co_release()的功能比較簡單，就是釋放資源 void co_release( stCoRoutine_t *co ) {if( co->cEnd ){free( co );} } - co_self
co_self()函數(shù)是獲取當(dāng)前正在執(zhí)行的協(xié)程，只要獲取到當(dāng)前協(xié)程環(huán)境的線程棧頂?shù)膮f(xié)程即可。 stCoRoutine_t *co_self() {return GetCurrThreadCo(); } stCoRoutine_t *GetCurrThreadCo( ) {stCoRoutineEnv_t *env = co_get_curr_thread_env();if( !env ) return 0;return GetCurrCo(env); } stCoRoutine_t *GetCurrCo( stCoRoutineEnv_t *env ) {return env->pCallStack[ env->iCallStackSize - 1 ]; } - co_enable_hook_sys
libco封裝了系統(tǒng)調(diào)用，在系統(tǒng)調(diào)用，比如send/recv/condition_wait等函數(shù)前面加了一層hook,有了這層hook就可以在系統(tǒng)調(diào)用的時候不讓線程阻塞而產(chǎn)生線程切換，co_enable_hook_sys()函數(shù)允許協(xié)程hook，當(dāng)然也可以不允許hook，直接使用原生的系統(tǒng)調(diào)用。 void co_enable_hook_sys() {stCoRoutine_t *co = GetCurrThreadCo();if( co ){co->cEnableSysHook = 1;} }
?
轉(zhuǎn)載于:https://my.oschina.net/u/2447371/blog/1591005
總結(jié)
以上是生活随笔為你收集整理的libco协程库源码解读的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 汇编学习(一)
- 下一篇： 898A. Rounding#数的舍入