記一次node協程模塊開發

開始

早前經過swoole瞭解到了協程的概念，正值當時看到了JS的GeneratorFunction，因而激動的心顫抖的手，敲着代碼往前走，就用js寫下了一個協程模塊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 : &regs[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 -> &regs[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, &regs, 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的棧上還找獲得其指針。可是在是什麼形成上述的這些問題或者是其餘什麼緣由，到如今我仍是毫無頭緒，但願有朋友能幫忙給出一些答案。

總結

一次不成功的開發之旅，雖然探索了不少東西，可是最後沒能解決亂地址的問題，確實讓人很沮喪，還但願能有高手幫忙指出。