記一次node協程模組開發
早前通過swoole瞭解到了協程的概念，正值當時看到了JS的GeneratorFunction，於是激動的心顫抖的手，敲著程式碼往前走，就用js寫下了一個協程模組 ofollow,noindex">node-coroutine-js .當然這個模組比較簡單，基本就是利用node本身的能力實現的，其中為了避免主線的阻塞所以使用了setImmediate的方法來執行方法。後來在瞭解協程方面資訊的時候，瞭解到了libco以後，又產生了通過libco的方式來做node協程模組，因為libco這個庫真的太厲害了，我接下來為大家分析一下其中我用到的swap相關的核心邏輯。至於libco的loop，在node的libuv面前卻顯得不太出眾。
libco核心邏輯
libco是通過一個stCoRoutine_t結構體來管理一個協程單元的，這個結構中主要包括了該協程單元的被交換時的暫存器值，以及stack的空間，以及協程執行的方法和傳入的引數值，通過co_create方法來初始化，初始化的過程也不算複雜，有興趣的可以自己瞭解一下。然後在初始化了stCoRoutine_t以後，通過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 );
}
複製程式碼
其中stCoRoutineEnv_t結構體是每個執行緒中管理整個協程操作排程的結構體，其中通過pCallStack陣列來維持一個呼叫棧，其中總是包含一個儲存主線資訊的stCoRoutine_t，該陣列預設有128的深度來用於一些巢狀的呼叫，不過通常情況下不會出現太深的巢狀，基本就是主線和當前正在呼叫的協程方法兩個值。從函式中我們可以看出，如果方法是第一次執行會首先通過coctx_make來做一次對coctx_t的初始化。結構體coctx_t是呼叫中的核心，上面我們說過stCoRoutine_t儲存了暫存器值就是通過該結構體來儲存的，我們來看下這個結構體的程式碼：
struct coctx_t
{
#if defined(__i386__)
void *regs[ 8 ];
#else
void *regs[ 14 ];
#endif
size_t ss_size;
char *ss_sp;
};
複製程式碼
從結構體的屬性我麼可以看出，32位儲存7個暫存器的值和返回地址的值，而64位儲存13個暫存器的值和返回地址的值，ss_sp則是儲存的是初始化stCoRoutine_t時所分配的stack在記憶體中的起始位置。瞭解了coctx_t以後我們來看看它是如何初始化的：
#if defined(__i386__)
int coctx_make( coctx_t *ctx,coctx_pfn_t pfn,const void *s,const void *s1 )
{
//make room for coctx_param
char *sp = ctx->ss_sp + ctx->ss_size - sizeof(coctx_param_t);
sp = (char*)((unsigned long)sp & -16L);
coctx_param_t* param = (coctx_param_t*)sp ;
param->s1 = s;
param->s2 = s1;
memset(ctx->regs, 0, sizeof(ctx->regs));
ctx->regs[ kESP ] = (char*)(sp) - sizeof(void*);
ctx->regs[ kEIP ] = (char*)pfn;
return 0;
}
#elif defined(__x86_64__)
int coctx_make( coctx_t *ctx,coctx_pfn_t pfn,const void *s,const void *s1 )
{
char *sp = ctx->ss_sp + ctx->ss_size;
sp = (char*) ((unsigned long)sp & -16LL);
memset(ctx->regs, 0, sizeof(ctx->regs));
ctx->regs[ kRSP ] = sp - 8;
ctx->regs[ kRETAddr] = (char*)pfn;
ctx->regs[ kRDI ] = (char*)s;
ctx->regs[ kRSI ] = (char*)s1;
return 0;
}
#endif
複製程式碼
接下來我們來了解一下上面的程式碼，雖然上面的程式碼有32位架構的也有64位架構的，但是其實做的都是同樣三件事:
1.指定sp的起始地址，因為coctx_t的ss_sp指標指向的位置在低位，而我們知道，esp所存放的棧上地址是由高到低的，所以需要通過 ctx->ss_sp + ctx->ss_size; 的方式找到記憶體的高位，再通過align方式將記憶體對齊，然後通過減去一個 sizeof(void*) 的記憶體用來放置fp的指標，而在32位的架構中還要多留兩個指標的位置存放傳入引數，就得到了協程方法的sp起始地址。
2.指定傳入的引數，我們可以看到在32位架構中，引數是通過push入棧的形式傳入的，後面的引數先進棧，前面的引數後進棧（所以s1在低位，s2在高位），而在64位的架構中則是通過rdi暫存器傳入第一個引數，rsi傳入第二個引數，這個對我們後面討論彙編語句的時候很重要，當然對於熟悉彙編的朋友來說，這些提不提都行。
3.將需要執行的方法的地址置為返回地址中，即是CoRoutineFunc。
在完成了coctx_t的初始化以後，則是檔案中最核心的呼叫了 co_swap 函式。在主線中呼叫co_resume函式時，lpCurrRoutine必然儲存的是主線的資訊，我們就不討論深入巢狀的情況了，就只看主線是如何呼叫協程函式的。 co_swap 中有很多記錄資訊的東西，我們可以拋開不看，其中最主要的就是嵌入的彙編函式 void coctx_swap( coctx_t *,coctx_t* ) asm("coctx_swap"); ，我們來看一下他的程式碼，很簡短，但是真的讓人驚歎：
#if defined(__i386__) leal 4(%esp), %eax //sp movl 4(%esp), %esp leal 32(%esp), %esp //parm a : ®s[7] + sizeof(void*) pushl %eax //esp ->parm a pushl %ebp pushl %esi pushl %edi pushl %edx pushl %ecx pushl %ebx pushl -4(%eax) movl 4(%eax), %esp //parm b -> ®s[0] popl %eax//ret func addr popl %ebx popl %ecx popl %edx popl %edi popl %esi popl %ebp popl %esp pushl %eax //set ret func addr xorl %eax, %eax ret #elif defined(__x86_64__) leaq 8(%rsp),%rax leaq 112(%rdi),%rsp pushq %rax pushq %rbx pushq %rcx pushq %rdx pushq -8(%rax) //ret func addr pushq %rsi pushq %rdi pushq %rbp pushq %r8 pushq %r9 pushq %r12 pushq %r13 pushq %r14 pushq %r15 movq %rsi, %rsp popq %r15 popq %r14 popq %r13 popq %r12 popq %r9 popq %r8 popq %rbp popq %rdi popq %rsi popq %rax //ret func addr popq %rdx popq %rcx popq %rbx popq %rsp pushq %rax xorl %eax, %eax ret #endif 複製程式碼
我們著重來分析一下32位的操作，剛剛我們說過，在32位系統中後面的引數先入棧，而前面的引數後入棧，所以 leal 4(%esp), %eax 首先將傳入的第一個引數的棧地址賦值給eax暫存器,第一個引數當然就是我們主線的上下文，接著通過 movl 4(%esp), %esp 將取第一個引數棧地址賦值給esp暫存器，也就是第一個引數本身，指向主線上下文的指標，為什麼要這麼做呢，我們接下來看這句 leal 32(%esp), %esp ，也就是會將指向上下文的指標加32以後的值賦值給esp，就等於這樣 current_ctx_ptr + 32 = esp 。剛剛說到我們的ctx指標指向的起始值也是低位，而esp的值是從高到低變化的，所以先將esp的值指向ctx.reg陣列結束處，這樣做在每次執行push指令的時候就能將值存入到當前的ctx.reg陣列的位置上，所以我們可以看到這樣的存值方式，eax的指標在ctx.reg[7]，而ebx在ctx.reg[1]，最後一句 pushl -4(%eax) 要著重說一下，eax我們知道是當前上線文的指標值，是通過esp+4得到的，那麼eax-4得到的就是最初的esp值，這個值自然是指向返回地址的，所以在ctx.reg[0]中存放的是返回地址，我們知道主線返回地址自然是呼叫 coctx_swap 方法的 co_swap 方法。 movl 4(%eax), %esp ，通過上面的分析我們就可以知道這是將第二個引數即協程上下文的的地址傳入esp暫存器，這個時候這個指標指向的既是ctx->regs[0]的地址，將之pop到eax，後面的則是將regs陣列中的值pop到對應的暫存器上。從剛剛我們看到的協程上下文的初始化中，我們可以看到我們將CoRoutineFunc的地址放入ctx->regs[0]中，所以 pushl %eax 即是將CoRoutineFunc的地址壓到棧頂，然後通過ret指令則會跳轉到esp指向的地址。而 xorl %eax, %eax 是一個清空eax暫存器的操作。
分析完了這個彙編檔案我們就可知道，在主線中呼叫了 coctx_swap 函式後，即會根據改變了的返回地址跳轉到CoRoutineFunc中執行：
static int CoRoutineFunc( stCoRoutine_t *co,void * )
{
if( co->pfn )
{
co->pfn( co->arg );
}
co->cEnd = 1;
stCoRoutineEnv_t *env = co->env;
co_yield_env( env );
return 0;
}
複製程式碼
在這個函式中，會呼叫我們在開始通過co_create註冊的函式，在執行完成後會呼叫co_yield_env：
stCoRoutine_t *last = env->pCallStack[ env->iCallStackSize - 2 ]; stCoRoutine_t *curr = env->pCallStack[ env->iCallStackSize - 1 ]; env->iCallStackSize--; co_swap( curr, last); 複製程式碼
這個函式將會通過co_swap返回到主線上，通過剛剛的描述我們可以知道即是返回到主線 co_swap 呼叫完成 coctx_swap 處。
到此基本整個libco的交換核心部分就討論完畢了，在瞭解了這個過程以後確實讓人拍案叫絕，以前看深入Linux核心架構的時候看到核心在建立執行緒時候程式碼：
Linux核心版本2.6.24
arch/x86/kernel/process_32.c
asmlinkage int sys_clone(struct pt_regs regs) {
unsigned long clone_flags;
unsigned long newsp;
int __user *parent_tidptr, *child_tidptr;
clone_flags = regs.ebx;
newsp = regs.ecx;
parent_tidptr = (int __user *)regs.edx; child_tidptr = (int __user *)regs.edi; if (!newsp)
newsp = regs.esp;
return do_fork(clone_flags, newsp, ®s, 0, parent_tidptr, child_tidptr);
}
複製程式碼
從上面的程式碼我們可以粗略看出，建立執行緒的newsp通過ecx傳入的，跟libco在堆上分配一塊記憶體然後將協程的esp指向該處頗有相似之處。當然協程的排程中因為沒有核心參與任務的排程，加上是單線操作，避開了一些鎖競爭之類的問題，使效能得到了極大的提升，即是其優勢所在。
node-coroutine
說完了libco的邏輯，說回我自己開發的模組，當然開發這個東西毫無必要，畢竟通過generatorFunction本身就可以實現了，不過我還是想嘗試一下這方面的開發，於是就開始了自坑之路，專案地址： node-coroutine
首先，libco中包括了很多關於多執行緒和非同步的東西，多執行緒在node中沒用，而非同步比起libuv來說確實只是個子集，所以我先砍掉了這兩塊，將env作為全域性的變數在註冊模組的時候就會初始化。另外libco的方法也只保留了跟交換上下文有關的核心方法，如下：
//2.context int coctx_init( coctx_t *ctx ,pfn_co_routine_t pfn,const void *s); //3.coroutine int co_create( stCoRoutine_t **co,const stCoRoutineAttr_t *attr,pfn_co_routine_t pfn,void *arg ); void co_swap(stCoRoutine_t* curr, stCoRoutine_t* pending_co,int main); void co_yield(); void co_free( stCoRoutine_t * co ); //4.func void save_stack_buffer(stCoRoutine_t* occupy_co); stStackMem_t* co_get_stackmem(stShareStack_t* share_stack); stStackMem_t* co_alloc_stackmem(unsigned int stack_size); stShareStack_t* co_alloc_sharestack(int count, int stack_size); 複製程式碼
至於我自己寫的流程相對來說比較簡單，只是在swap的時候沒有當時呼叫而是通過uv_work_t的回撥來實現，因為最後的協程方法都是在迴圈中呼叫，如果都是無限迴圈的方法，很可能造成主線的其他業務無法執行，一直在協程中執行的情況，所以通過uv_work的方式可以讓libuv的loop始終在執行，在從協程回到主線時，可以處理其他業務（跟js版使用setImmediate來執行切換的思路類似）。不過在開發過程中遇到的兩個問題倒是值得跟大家分享：
SetStackLimit的坑
因為協程的棧地址是在堆上執行，所以在第一次跑測試的就報出了這個錯誤：
RangeError: Maximum call stack size exceeded 複製程式碼
這種情況一般在我們的平時的程式設計中多半是無限的遞迴呼叫導致，但是我很確定我的程式碼中沒有遞迴，那麼我就判斷出肯定是v8對記憶體中執行的地址是有限制的，這個時候我就想起了node的v8-options中有一個 stack_size的選項，既然有這個選項我就猜測應該會有一個設定這個值的地方，於是就直接在v8.h檔案中搜索stack_size，但是並沒有，於是我就只搜尋stack，出來的選項有點多，不過沒往下跳幾次，我就找到了一個setStackLimit的方法，然後心中那個高興啊，覺得自己解決這個問題易如反掌，因為需要設定的stack是在協程方法中於是我就在我的協程方法中加入了這樣的語句：
uintptr_t limit = reinterpret_cast<uintptr_t>(&limit) - (co_arg->coroutine->ctx).ss_size; globalIsolate->SetStackLimit(limit); 複製程式碼
結果這樣以後並沒有起到作用，這讓我很詫異，我去V8看了原始碼，這個方法主要就是設定thread_top的c_limit值和js_limit值，然後我在原始碼中列印了設定後的這兩個值，發現設定成功了，但是依然沒解決問題，這讓我很是不解。於是我就在v8中搜索這個錯誤，然後發現該錯誤是從方法Isolate::StackOverflow中爆出來的，於是我在該方法中做了斷言，產生了core檔案，然後用 llnode 做了分析，得到下圖：
從該圖可以看出是在internal的frame中判斷出錯的，這個是v8內部的runtime方法執行而爆出來的錯。於是我就找到了v8/src/runtime/runtime_internal.cc中,這個檔案中有兩個地方都出現了 isolate->StackOverflow() 的呼叫，這個就簡單了，我在這兩個方法下都下個斷言，然後就發現錯誤是這個地方暴出的:
RUNTIME_FUNCTION(Runtime_ThrowStackOverflow) { SealHandleScope shs(isolate); DCHECK_LE(0, args.length()); return isolate->StackOverflow(); } 複製程式碼
你如果搜尋Runtime_ThrowStackOverflow在全域性都是找不到方法的，因為在v8內部通過巨集將Runtime方法都放到了一個數組中，而其索引成為了FunctionId來索引這些方法呼叫，所以通過搜尋kThrowStackOverflow即可找到呼叫呼叫處，發現在builtins-xxx.cc中都是呼叫這個方法，當然我的機器是x64的所以我直接找到了builtins-x64.cc的檔案中，在每個呼叫kThrowStackOverflow的地方打印出標記，然後編譯執行，結果。。。編譯的時候我列印的語句倒是都執行了，但是執行的時候一個都沒執行。然後我就猜測這個地方的可能在編譯過程中都變成了位元組碼，而我的列印語句明顯不是需要轉變成位元組碼的一部分，所以編譯的時候直接執行過了，但是真正執行的時候並沒有什麼用。那怎麼辦呢？當然就是按照他的格式寫啦，於是我在runtime-internal.cc中加入了一個方法
RUNTIME_FUNCTION(Runtime_ThrowStackOverflowDebug) {
SealHandleScope shs(isolate);
DCHECK_LE(0, args.length());
printf("have done!\n");
return isolate->StackOverflow();
}
複製程式碼
當然只在這裡加還是不夠的還要在src/runtime/runtime.h中的 #define FOR_EACH_INTRINSIC_INTERNAL(F) 下面加一條 F(ThrowStackOverflowDebug, 0, 1) 這個時候再編譯就行了，就這樣通過測試我發現這個錯誤主要是builtins-x64.cc中兩個地方報出來的：
static void Generate_CheckStackOverflow(MacroAssembler* masm,IsTagged rax_is_tagged) 複製程式碼
中通過判斷：
__ cmpp(rcx, r11); __ j(greater, &okay);// Signed comparison. // Out of stack space. __ CallRuntime(Runtime::kThrowStackOverflow); 複製程式碼
爆出，以及：
static void Generate_StackOverflowCheck(MacroAssembler* masm, Register num_args, Register scratch,Label* stack_overflow,Label::Distance stack_overflow_distance = Label::kFar) 複製程式碼
中的判斷:
__ cmpp(scratch, num_args); // Signed comparison. __ j(less_equal, stack_overflow, stack_overflow_distance); 複製程式碼
不得不說v8這個偽彙編寫得真是太舒服了，讓我爆破之魂熊熊燃燒，於是 通過將上面方法中的方法 __ j(greater, &okay); 改成 __ j(always, &okay); ，而下面的 __ j(less_equal, stack_overflow, stack_overflow_distance); 改成 __ j(never, stack_overflow, stack_overflow_distance); 重新編譯，然後問題就解決了。不過我總不能讓別人用我這個編譯版本吧，於是還是靜下來心來看，發現這些判斷上面都跟一個變數有密切的關係： Heap::kRealStackLimitRootIndex 於是我搜索了這個變數。好吧又回到了stacklimit上，這次我找到了Heap::SetStackLimits發現是這個方法下會設定上面的引數:
roots_[kRealStackLimitRootIndex] = reinterpret_cast<Object*>((isolate_->stack_guard()->real_jslimit() & ~kSmiTagMask) | kSmiTag); 複製程式碼
我又在下面下了斷言，然後執行測試檔案，果然沒執行，這個時候我很困惑，難道還真的不能設定這個值不成？然後就用試一試的心態又放到主線上執行，結果這次真的執行了。原來 isolate::SetStackLimit 要在正常的stack下去設定才行，通過 isolate::SetStackLimit 打斷點以後發現這個原因其實很簡單，因為設定了這個limit以後要過會兒才通過runtime方法的呼叫來生效，而在協程方法中不會執行這些方法，所以設定了也不會生效。所以我在main.cpp的方法中直接設定了 globalIsolate->SetStackLimit(1) 因為不能判斷到底最低的stack在哪兒。
在這次修改以後終於能執行成功了，但是我還沒來得及高興，又發生了崩潰了。。。
莫名其妙的HeapObject
既然發生了崩潰，第一件事就是用llnode開啟core檔案，原來是新生代回收在做掃描的時候發生了崩潰，既然找到了地址當然是開啟程式碼一看，這是一個很簡單的Inline函式總共只有一句話：
flags_ & kIsInNewSpaceMask) != 0; 複製程式碼
所以只能判斷是這個物件本身發生了問題，於是我列印了掃描的所有object的指標，果然發現在崩潰時會出現一個很莫名其妙的地址，如下圖所示：
這個HeapObject的出現就讓我百思不得其解了。難道是在棧上產生的物件？但新生代和老生代的都是heap上的記憶體，按理說不應該都在一個範圍內，不應該出現如此詭異的記憶體地址啊，所以猜測應該是有其他地址的記憶體溢位導致改寫了這裡的正常值，或者是已經回收了這段記憶體但是在jsframe的棧上還找得到其指標。但是在是什麼造成上述的這些問題或者是其他什麼原因，到現在我還是毫無頭緒，希望有朋友能幫忙給出一些答案。
總結
一次不成功的開發之旅，雖然探索了很多東西，但是最後沒能解決亂地址的問題，確實讓人很沮喪，還希望能有高手幫忙指出。