libco协程库学习笔记

Posted on | In | Visitors:

前言

对于协程具体的概念谷歌百度可以搜出一堆，我简单说一下我的理解：协程=函数调用+线程=用户级线程。当然这个公式适用于libco，或者说适用于非对称协程。

为什么会有协程的存在？

因为线程切换开销太大。

但是线程切换不是几条指令就可以完成嘛？

这里的开销还包括陷入内核进行线程调度的开销，简单的说，就是执行系统调用的开销。

所以结论就有了：协程的切换和创建不需要经过系统调用就可以实现，并且用户随时可以切出，下次切入时可以继续从切出的位置执行。

分析

首先看下协程数据结构的源码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
struct stCoRoutine_t
{
	stCoRoutineEnv_t *env;    ///协程执行环境
	pfn_co_routine_t pfn;     ///协程执行的函数
	void *arg;  			  ///协程执行函数传入的参数
	coctx_t ctx;			  ///协程切换时用于保存CPU上下文

	char cStart;			  ///是否第一次运行
	char cEnd;				  ///
	char cIsMain;			  ///是不是main协程
	char cEnableSysHook;	  ///
	char cIsShareStack;		  ///是否使用共享栈

	void *pvEnv;

	//char sRunStack[ 1024 * 128 ];
	stStackMem_t* stack_mem;    ///运行时栈


	//save satck buffer while confilct on same stack_buffer;
	char* stack_sp; 
	unsigned int save_size;
	char* save_buffer;

	stCoSpec_t aSpec[1024];

};

env是隶属于协程所在线程的一个数据结构，主要用来保存协程的调用链以及线程调度协程的调度器，每一个协程都指向这个协程环境结构。

1
2
3
4
5
6
7
8
9
10
struct stCoRoutineEnv_t
{
	stCoRoutine_t *pCallStack[ 128 ];     ///存放挂起的协程的栈
	int iCallStackSize;     ///栈中元素个数
	stCoEpoll_t *pEpoll;    ///调度器

	//for copy stack log lastco and nextco
	stCoRoutine_t* pending_co;
	stCoRoutine_t* occupy_co;
};

协程环境结构中有一个pCallStack栈，它就是用来保存协程的调用关系，可以看出数组大小固定上限128，即线程最多调度128个协程，换句话说，调用深度最多128层。每当运行一个新的协程，就压入栈中，挂起一个协程就弹出，需要维护使得栈顶的协程总是正在运行。

下面两个指针用于记录协程切换时占有共享栈和将要切换运行的协程，共享栈的内容后面会说。

继续看协程结构，里面有一个stack_mem指针，即协程的运行时栈，用于保存协程运行中发生的函数调用关系和函数中创建的临时变量。但并不是说每一个协程都给创建一个128kb的栈空间，这里又涉及到共享栈的应用，可以大大节省内存空间的使用。

libco中有一个主协程的存在，也就是执行main函数的协程，它是在第一次调用co_create创建一个协程的时候创建的，和env一起分配。当它创建并运行协程co1时，将co1压栈，co1创建并运行co2时，co2压栈。co2调用yield让出CPU时，只能让给co1，co2出栈，co1让出CPU时，只能让给main协程，co1出栈。

co_create 只负责创建一个协程结构并初始化，此时协程还没有运行，需要注意的是，在协程结构中的栈并不是进程中的栈空间，而是重新malloc出来的位于进程堆中的空间。

co_resume

co_resume 负责将CPU从调用者协程切换到被调用者协程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
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 );


}

首先获取到目前正在运行的协程（调用者协程），判断即将运行的协程（被调用者协程）是不是第一次运行，如果是第一次运行，那么需要初始化ctx结构，即用来保存CPU切换的上下文。然后将即将运行的协程压栈，调用co_swap切换CPU到co。

co_yield

co_yield负责将CPU从被调用者协程切换到调用者协程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
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);
}

void co_yield_ct()
{

	co_yield_env( co_get_curr_thread_env() );
}
void co_yield( stCoRoutine_t *co )
{
	co_yield_env( co->env );
}

逻辑很简单，不再赘述，需要说的是，一般应用中会调用co_yield_ct函数来进行CPU的让出，因为肯定会让给调用当前协程的调用协程，也就是env栈中的当前协程的前一个协程，所以即使调用co_yield让出CPU给指定协程co，也无济于事。

co_swap

co_swap把将要运行的协程的上下文从sava_buffer加载到其使用的共享栈/私有栈中，将CPU从当前运行的协程curr上下文切换到pending_co上下文中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
void co_swap(stCoRoutine_t* curr, stCoRoutine_t* pending_co)
{
 	stCoRoutineEnv_t* env = co_get_curr_thread_env();

	//get curr stack sp
	char 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 mem
		stCoRoutine_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 context
	coctx_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 buffer
		if (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);
		}
	}
}

参考资料：
libco协程库