云风coroutine协程库

前言

库介绍

协程是一种共享堆，不共享栈，由用户主动调度的执行体（一般需要提供 yield 和 resume 语义）。

这个coroutine协程库实现是基于多个协程共享栈的方式。但是每个 coroutine 都会从 heap 上分配内存来保存自己 stack 的内容，当前运行实只有一个 stack。

随着 Golang 的兴起，协程尤其是有栈协程 (stackful coroutine) 越来越受到程序员的关注。协程几乎成了程序员的一套必备技能。

云风实现了一套 C 语言的协程库，整体背景可以参考其 博客。

这个协程库非常轻量级，一共也才 200 多行代码，使用上更贴近于 lua 的写法（众所周知，云风是知名的 lua 粉)。整体基于 ucontext 和共享栈模型实现了有栈协程，代码质量毋庸置疑，本文将详细剖析该协程库的实现原理。

协程的几种实现方式及原理

协程又可以称为用户线程,微线程，可以将其理解为单个进程或线程中的多个用户态线程，这些微线程在用户态进程控制和调度.协程的实现方式有很多种，包括

1. 使用glibc中的ucontext库实现
2. 利用汇编代码切换上下文
3. 利用C语言语法中的switch-case的奇淫技巧实现(protothreads)
4. 利用C语言的setjmp和longjmp实现

实际上，无论是上述那种方式实现协程,其原理是相同的，都是通过保存和恢复寄存器的状态，来进行各协程上下文的保存和切换。

协程较于函数和线程的优点

• 相比于函数:协程避免了传统的函数调用栈，几乎可以无限地递归
• 相比与线程:协程没有内核态的上下文切换，近乎可以无限并发。协程在用户态进程显式的调度，可以把异步操作转换为同步操作，也意味着不需要加锁,避免了加锁过程中不必要的开销。

进程,线程以及协程的设计都是为了并发任务可以更好的利用CPU资源，他们之间最大的区别在于CPU资源的使用上:

• 进程和线程的任务调度是由内核控制的，是抢占式的；
• 协程的任务调度是在用户态完成,需要代码里显式地将CPU交给其他协程,是协作式的

由于我们可以在用户态调度协程任务，所以我们可以把一组相互依赖的任务设计为协程。这样,当一个协程任务完成之后,可以手动的进行任务切换，把当前任务挂起(yield),切换到另一个协程区工作.由于我们可以控制程序主动让出资源，很多情况下将不需要对资源进行加锁。

前置知识

ucontext 函数族

ucontext 函数有 4 个，如下所示：

#include <ucontext.h>

// 用户上下文的获取和设置
int getcontext(ucontext_t *ucp);
int setcontext(const ucontext_t *ucp);

// 操纵用户上下文
void makecontext(ucontext_t *ucp, void (*func)(void), int argc, ...);
int swapcontext(ucontext_t *oucp, const ucontext_t *ucp);

ucontext 函数用户进程内部的 context 控制，帮助用户更方便实现 coroutine，可视为更先进的 setjmp/longjmp。

4 个函数都依赖于 ucontext_t 类型，这个类型大致为：

typedef struct {
    ucontext_t *uc_link;
    sigset_t    uc_sigmask;
    stack_t     uc_stack;
    mcontext_t  uc_mcontext;
    ...
} ucontext_t;

其中：

• uc_link：当前上下文结束时要恢复到的上下文，其中上下文由 makecontext() 创建；
• uc_sigmask：上下文要阻塞的信号集合；
• uc_stack：上下文所使用的 stack；
• uc_mcontext：其中 mcontext_t

类型与机器相关的类型。这个字段是机器特定的保护上下文的表示，包括协程的机器寄存器；

这几个 API 的作用：

getcontext(ucontext_t *ucp)

将当前的 context 保存在 ucp 中。成功返回 0，错误时返回 -1 并设置 errno；

setcontext(const ucontext_t *ucp)

恢复用户上下文为 ucp 所指向的上下文，成功调用不用返回。错误时返回 -1 并设置 errno。 ucp 所指向的上下文应该是 getcontext() 或者 makecontext() 产生。 如果上下文是由 getcontext() 产生，则切换到该上下文后，程序的执行在 getcontext() 后继续执行。比如下面这个例子每隔 1 秒将打印 1 个字符串：

int main(void)
{
    ucontext_t context;

    getcontext(&context);
    printf("Hello world\n");
    sleep(1);
    setcontext(&context);
    return 0;
}

如果上下文是由 makecontext(ucontext_t *ucp, void (*func)(void), int argc, ...) 产生，切换到该上下文，程序的执行切换到 makecontext() 调用所指定的第二个参数的函数上。当函数返回后，如果 ucp.uc_link 为 NULL，则结束运行；反之跳转到对应的上下文。

void foo(void)
{
    printf("foo\n");
}
    
int main(void)
{
    ucontext_t context;
    char stack[1024];
       
    getcontext(&context);
    context.uc_stack.ss_sp = stack;
    context.uc_stack.ss_size = sizeof(stack);
    context.uc_link = NULL;
    makecontext(&context, foo, 0);
       
    printf("Hello world\n");
    sleep(1);
    setcontext(&context);
    return 0;
}

以上输出 Hello world 之后会执行 foo()，然后由于 uc_link 为 NULL，将结束运行。

下面这个例子：

void foo(void)
{
    printf("foo\n");
}
    
void bar(void)
{
    printf("bar\n");
}
    
int main(void)
{
    ucontext_t context1, context2;
    char stack1[1024];
    char stack2[1024];
       
    getcontext(&context1);
    context.uc_stack.ss_sp = stack1;
    context.uc_stack.ss_size = sizeof(stack1);
    context.uc_link = NULL;
    makecontext(&context1, foo, 0);
       
    getcontext(&context2);
    context.uc_stack.ss_sp = stack2;
    context.uc_stack.ss_size = sizeof(stack2);
    context.uc_link = &context1;
    makecontext(&context1, bar, 0);
        
    printf("Hello world\n");
    sleep(1);
    setcontext(&context2);
        
    return 0;
}

此时调用 makecontext() 后将切换到 context2 执行 bar()，然后再调用 context1 的 foo()。由于 context1 的 uc_link 为 NULL，程序停止。

makecontext()

修改 ucp 所指向的上下文；

swapcontext(ucontext_t *oucp, const ucontext_t *ucp)

保存当前的上下文到 ocup，并且设置到 ucp 所指向的上下文。成功返回 0，失败返回 -1 并设置 errno。

如下面这个例子所示：

#include <stdio.h>
#include <ucontext.h>
    
static ucontext_t ctx[3];
    
static void
f1(void)
{
    printf("start f1\n");
    // 将当前 context 保存到 ctx[1]，切换到 ctx[2]
    swapcontext(&ctx[1], &ctx[2]);
    printf("finish f1\n");
}
    
static void
f2(void)
{
    printf("start f2\n");
    // 将当前 context 保存到 ctx[2]，切换到 ctx[1]
    swapcontext(&ctx[2], &ctx[1]);
    printf("finish f2\n");
}
    
int main(void)
{
    char stack1[8192];
    char stack2[8192];
    
    getcontext(&ctx[1]);
    ctx[1].uc_stack.ss_sp = stack1;
    ctx[1].uc_stack.ss_size = sizeof(stack1);
    ctx[1].uc_link = &ctx[0]; // 将执行 return 0
    makecontext(&ctx[1], f1, 0);
    
    getcontext(&ctx[2]);
    ctx[2].uc_stack.ss_sp = stack2;
    ctx[2].uc_stack.ss_size = sizeof(stack2);
    ctx[2].uc_link = &ctx[1];
    makecontext(&ctx[2], f2, 0);
    
    // 将当前 context 保存到 ctx[0]，切换到 ctx[2]
    swapcontext(&ctx[0], &ctx[2]);
    return 0;
}      

此时将输出：

start f2
start f1
finish f2
finish f1

无栈协程

无栈协程，顾名思义，就是不需要使用栈的协程。在常规的协程或者线程中，我们通常会为每一个协程或线程分配一个栈。这个栈用于存储协程的执行状态，包括局部变量、函数调用关系等。当我们从一个协程切换到另一个协程时，我们会保存当前协程的栈信息，以便在切换回来时能够恢复到切换前的状态。

然而，在无栈协程中，我们并不为每个协程分配一个栈。相反，所有的协程共享全局的执行状态。这意味着，在无栈协程中，协程的执行状态不是保存在栈上，而是保存在堆上。换句话说，协程的栈帧内保存的不是状态，而是指向状态的指针。当我们需要切换协程时，我们只需要改变这些指针的指向，而不需要保存和恢复栈信息。

无栈协程的优点是节省了内存空间，因为我们不需要为每个协程分配一个栈。此外，因为不需要保存和恢复栈信息，所以无栈协程的切换速度更快。然而，无栈协程的缺点是编程模型更复杂，因为我们需要自己管理协程的执行状态。

总的来说，无栈协程是一种更轻量级的协程实现方式，它通过共享全局的执行状态，避免了栈的使用，从而实现了更快速的协程切换和更小的内存占用。

coroutine 的使用

我们首先基于 coroutine 的例子来讲下 coroutine 的基本使用，以方便后面原理的讲解

struct args {
	int n;
};

static void foo(struct schedule * S, void *ud) {
	struct args * arg = ud;
	int start = arg->n;
	int i;
	for (i=0;i<5;i++) {
		printf("coroutine %d : %d\n",coroutine_running(S) , start + i);
		// 切出当前协程
		coroutine_yield(S);
	}
}

static void test(struct schedule *S) {
	struct args arg1 = {0};
	struct args arg2 = {100};

	// 创建两个协程
	int co1 = coroutine_new(S, foo, &arg1);
	int co2 = coroutine_new(S, foo, &arg2);

	printf("main start\n");
	while (coroutine_status(S,co1) && coroutine_status(S,co2)) {
		// 使用协程 co1
		coroutine_resume(S,co1);
		// 使用协程 co2
		coroutine_resume(S,co2);
	}
	printf("main end\n");
}

int main() {
	// 创建一个协程调度器
	struct schedule * S = coroutine_open();

	test(S);

	// 关闭协程调度器
	coroutine_close(S);

	return 0;
}

运行结果

(base) sv@sv-NF5280M5:/home/sv/pengeHome/coroutine$ ./main
main start
coroutine 0 : 0
coroutine 1 : 100
coroutine 0 : 1
coroutine 1 : 101
coroutine 0 : 2
coroutine 1 : 102
coroutine 0 : 3
coroutine 1 : 103
coroutine 0 : 4
coroutine 1 : 104
main end

从代码看来，首先利用 coroutine_open 创建了协程调度器 S，用来统一管理全部的协程。
同时在 test 函数中，创建了两个协程 co1 和 co2，不断的反复 yield 和 resume 协程，直至两个协程执行完毕。

可以看出，最核心的几个对象和函数是:

1. struct schedule* S 协程调度器
2. coroutine_resume(S,co1); 切入该协程
3. coroutine_yield(S); 切出该协程

接下来，会从这几点出发，分析 coroutine 的原理。

struct schedule 协程调度器

/**
* 协程调度器
*/
struct schedule {
	char stack[STACK_SIZE];	// 运行时栈

	ucontext_t main; // 主协程的上下文
	int nco; // 当前存活的协程个数
	int cap; // 协程管理器的当前最大容量，即可以同时支持多少个协程。如果不够了，则进行扩容
	int running; // 正在运行的协程ID
	struct coroutine **co; // 一个一维数组，用于存放协程 
};

协程调度器 schedule 负责管理所有协程，有几个属性非常重要：

1. struct coroutine **co; 是一个一维数组，存放了目前所有的协程。
2. ucontext_t main; 主协程的上下文，方便后面协程执行完后切回到主协程。
3. char stack[STACK_SIZE]; 这个非常重要，是所有协程的运行时栈。具体共享栈的原理会在下文讲到。

此外，coroutine_open 负责创建并初始化一个协程调度器，coroutine_close 负责销毁协程调度器以及清理其管理的所有协程。

协程的创建: coroutine_new

/*
* 协程
*/
struct coroutine {
	coroutine_func func; // 协程所用的函数
	void *ud;  // 协程参数
	ucontext_t ctx; // 协程上下文
	struct schedule * sch; // 该协程所属的调度器
	ptrdiff_t cap; 	 // 已经分配的内存大小
	ptrdiff_t size; // 当前协程运行时栈，保存起来后的大小
	int status;	// 协程当前的状态
	char *stack; // 当前协程的保存起来的运行时栈
};

创建协程

int 
coroutine_new(struct schedule *S, coroutine_func func, void *ud) {
	struct coroutine *co = _co_new(S, func , ud);
	if (S->nco >= S->cap) {
		// 如果目前协程的数量已经大于调度器的容量，那么进行扩容
		int id = S->cap;	// 新的协程的id直接为当前容量的大小
		// 扩容的方式为，扩大为当前容量的2倍，这种方式和Hashmap的扩容略像
		S->co = realloc(S->co, S->cap * 2 * sizeof(struct coroutine *));
		// 初始化内存
		memset(S->co + S->cap , 0 , sizeof(struct coroutine *) * S->cap);
		//将协程放入调度器
		S->co[S->cap] = co;
		// 将容量扩大为两倍
		S->cap *= 2;
		// 尚未结束运行的协程的个数 
		++S->nco; 
		return id;
	} else {
		// 如果目前协程的数量小于调度器的容量，则取一个为NULL的位置，放入新的协程
		int i;
		for (i=0;i<S->cap;i++) {
			/* 
			 * 为什么不 i%S->cap,而是要从nco+i开始呢 
			 * 这其实也算是一种优化策略吧，因为前nco有很大概率都非NULL的，直接跳过去更好
			*/
			int id = (i+S->nco) % S->cap;
			if (S->co[id] == NULL) {
				S->co[id] = co;
				++S->nco;
				return id;
			}
		}
	}
	assert(0);
	return -1;
}

coroutine_new 负责创建并初始化一个新协程对象，同时将该协程对象放到协程调度器里面。

这里的实现有两个非常有意思的点：

1. 扩容：当目前尚存活的线程个数 nco 已经等于协程调度器的容量 cap 了，这个时候需要对协程调度器进行扩容，这里直接就是非常经典简单的 2 倍扩容。
2. 如果无需扩容，则需要找到一个空的位置，放置初始化好的协程。这里一般直接从数组第一位开始找，直到找到空的位置即可。但是云风把这里处理成从第 nco 位开始寻找（nco 代表当前存活的个数。因为一般来说，前面几位最开始都是存活的，从第 nco 位开始找，效率会更高。

这样，一个协程对象就被创建好，此时该协程的状态是 READY，但尚未正式执行。

coroutine_resume 函数会切入到指定协程中执行。当前正在执行的协程的上下文会被保存起来，同时上下文替换成新的协程，该协程的状态将被置为 RUNNING。

进入 coroutine_resume 函数的前置状态有两个 READY 和 SUSPEND，这两个状态下 coroutine_resume 的处理方法也是有很大不同。我们先看下协程在 READY 状态下进行 coroutine_resume 的流程。

coroutine_resume(READY -> RUNNING）

/**
* 切换到对应协程中执行
* 
* @param S 协程调度器
* @param id 协程ID
*/
void 
coroutine_resume(struct schedule * S, int id) {
	assert(S->running == -1);
	assert(id >=0 && id < S->cap);

    // 取出协程
	struct coroutine *C = S->co[id];
	if (C == NULL)
		return;

	int status = C->status;
	switch(status) {
	case COROUTINE_READY:
	    //初始化ucontext_t结构体,将当前的上下文放到C->ctx里面
		getcontext(&C->ctx);
		// 将当前协程的运行时栈的栈顶设置为S->stack
		// 每个协程都这么设置，这就是所谓的共享栈。（注意，这里是栈顶）
		C->ctx.uc_stack.ss_sp = S->stack; 
		C->ctx.uc_stack.ss_size = STACK_SIZE;
		C->ctx.uc_link = &S->main; // 如果协程执行完，将切换到主协程中执行
		S->running = id;
		C->status = COROUTINE_RUNNING;

		// 设置执行C->ctx函数, 并将S作为参数传进去
		uintptr_t ptr = (uintptr_t)S;
		makecontext(&C->ctx, (void (*)(void)) mainfunc, 2, (uint32_t)ptr, (uint32_t)(ptr>>32));

		// 将当前的上下文放入S->main中，并将C->ctx的上下文替换到当前上下文
		swapcontext(&S->main, &C->ctx);
		break;
	case COROUTINE_SUSPEND:
	    // 将协程所保存的栈的内容，拷贝到当前运行时栈中
		// 其中C->size在yield时有保存
        // 这块这里先不看
		memcpy(S->stack + STACK_SIZE - C->size, C->stack, C->size);
		S->running = id;
		C->status = COROUTINE_RUNNING;
		swapcontext(&S->main, &C->ctx);
		break;
	default:
		assert(0);
	}
}

这段函数非常的重要，有几个不可忽视的点：

1. getcontext(&C->ctx); 初始化 ucontext_t 结构体，将当前的上下文放到 C->ctx 里面
2. C->ctx.uc_stack.ss_sp = S->stack; 设置当前协程的运行时栈，也是共享栈。
3. C->ctx.uc_link = &S->main; 如果协程执行完，则切换到 S->main 主协程中进行执行。如果不设置, 则默认为 NULL，那么协程执行完，整个程序就结束了。

接下来是 makecontext，这个函数用来设置对应 ucontext 的执行函数。如上，将 C->ctx 的执行函数体设置为了 mainfunc。

makecontext 后面的两个参数也非常有意思，这个可以看出来是把一个指针掰成了两个 int 作为参数传给 mainfunc 了。而在 mainfunc 的实现可以看出来，又会把这两个 int 拼成了一个 struct schedule*。

那么，为什么不直接传 struct schedule* 呢，而要这么做，通过先拆两半，再在函数中拼起来？

这是因为 makecontext 的函数指针的参数是 uint32_t 类型，在 64 位系统下，一个 uint32_t 没法承载一个指针, 所以基于兼容性的考虑，才采用了这种做法。

接下来调用了 swapcontext 函数，这个函数比较简单，但也非常核心。作用是将当前的上下文内容放入 S->main 中，并将 C->ctx 的上下文替换到当前上下文。这样的话，将会执行新的上下文对应的程序了。在 coroutine 中, 也就是开始执行 mainfunc 这个函数。(mainfunc 是对用户提供的协程函数的封装)。

mainfunc

/*
 * 通过low32和hi32 拼出了struct schedule的指针，这里为什么要用这种方式，而不是直接传struct schedule*呢？
 * 因为makecontext的函数指针的参数是int可变列表，在64位下，一个int没法承载一个指针
*/
static void
mainfunc(uint32_t low32, uint32_t hi32) {
	uintptr_t ptr = (uintptr_t)low32 | ((uintptr_t)hi32 << 32);
	struct schedule *S = (struct schedule *)ptr;

	int id = S->running;
	struct coroutine *C = S->co[id];
	C->func(S,C->ud);	// 中间有可能会有不断的yield
	_co_delete(C);
	S->co[id] = NULL;
	--S->nco;
	S->running = -1;
}

协程的切出：coroutine_yield

用 coroutine_yield 可以使当前正在运行的协程切换到主协程中运行。此时，该协程会进入 SUSPEND 状态

coroutine_yield 的具体实现依赖于两个行为：

1. 调用 _save_stack 将当前协程的栈保存起来。因为 coroutine 是基于共享栈的，所以协程的栈内容需要单独保存起来。
2. swapcontext 将当前上下文保存到当前协程的 ucontext 里面，同时替换当前上下文为主协程的上下文。 这样的话，当前协程会被挂起，主协程会被继续执行。

/**
* 将当前正在运行的协程让出，切换到主协程上
* @param S 协程调度器
*/
void
coroutine_yield(struct schedule * S) {
	// 取出当前正在运行的协程
	int id = S->running;
	assert(id >= 0);

	struct coroutine * C = S->co[id];
	assert((char *)&C > S->stack);

	// 将当前运行的协程的栈内容保存起来
	_save_stack(C,S->stack + STACK_SIZE);
	
	// 将当前栈的状态改为 挂起
	C->status = COROUTINE_SUSPEND;
	S->running = -1;

	// 所以这里可以看到，只能从协程切换到主协程中
	swapcontext(&C->ctx , &S->main);
}

这里也有个点极其关键, 就是如何保存当前协程的运行时栈, 也就是如何获取整个栈的内存空间。

这里我们需要了解下栈内存空间的布局，即栈的生长方向是从高地址往低地址。我们只要找到栈的栈顶和栈底的地址，就可以找到整个栈内存空间了。

在 coroutine 中，因为协程的运行时栈的内存空间是自己分配的。在 coroutine_resume 阶段设置了 C->ctx.uc_stack.ss_sp = S.S->stack。根据以上理论，栈的生长方向是高地址到低地址，因此栈底的就是内存地址最大的位置，即 S->stack + STACK_SIZE 就是栈底位置。

那么，如何找到栈顶的位置呢？coroutine 是基于以下方法做的：

/*
* 将本协程的栈内容保存起来
* @top 栈顶 
* 
*/
static void
_save_stack(struct coroutine *C, char *top) {
    // 这个dummy很关键，是求取整个栈的关键
    // dummy变量在栈上分配，其地址即为当前栈顶地址
    char dummy = 0;

    // 判断当前协程的栈大小是否超过了最大栈大小
    // 如果超过，程序将终止运行
    assert(top - &dummy <= STACK_SIZE);

    // 如果当前协程的栈容量不足以保存所有的栈内容
    // 那么就需要重新分配一个更大的栈空间
    if (C->cap < top - &dummy) {
        // 释放原来的栈空间
        free(C->stack);

        // 计算新的栈容量
        C->cap = top - &dummy;

        // 分配新的栈空间
        C->stack = malloc(C->cap);
    }

    // 计算当前协程的栈大小
    C->size = top - &dummy;

    // 将当前协程的栈内容复制到新的栈空间中
    memcpy(C->stack, &dummy, C->size);
}

这里特意用到了一个 dummy 变量，这个 dummy 的作用非常关键也非常巧妙，大家可以细细体会下。因为 dummy 变量是刚刚分配到栈上的，此时就位于 栈的最顶部位置。整个内存布局如下图所示：

因此整个栈的大小就是从栈底到栈顶，S->stack + STACK_SIZE - &dummy。

最后又调用了 memcpy 将当前运行时栈的内容，拷贝到了 C->stack 中保存了起来。

coroutine_resume(SUSPEND -> RUNNING）

当协程被 yield 之后会进入 SUSPEND 阶段，对该协程调用 coroutine_resume 会再次切入该协程。

这里的实现有两个重要的点：

1. memcpy(S->stack + STACK_SIZE - C->size, C->stack, C->size);
   我们知道，在 yield 的时候，协程的栈内容保存到了 C->stack 数组中。
   这个时候，就是用 memcpy 把协程的之前保存的栈内容，重新拷贝到运行时栈里面。这里有个点，拷贝的开始位置，需要简单计算下
   S->stack + STACK_SIZE - C->size 这个位置就是之前协程的栈顶位置。
2. swapcontext(&S->main, &C->ctx); 交换上下文。

/**
* 切换到对应协程中执行
* 
* @param S 协程调度器
* @param id 协程ID
*/
void 
coroutine_resume(struct schedule * S, int id) {
	assert(S->running == -1);
	assert(id >=0 && id < S->cap);

    // 取出协程
	struct coroutine *C = S->co[id];
	if (C == NULL)
		return;

	int status = C->status;
	switch(status) {
	case COROUTINE_READY:
			...
	case COROUTINE_SUSPEND:
	    // 将协程所保存的栈的内容，拷贝到当前运行时栈中
		// 其中C->size在yield时有保存
		memcpy(S->stack + STACK_SIZE - C->size, C->stack, C->size);
		S->running = id;
		C->status = COROUTINE_RUNNING;
		swapcontext(&S->main, &C->ctx);
		break;
	default:
		assert(0);
	}
}

状态机转换

在 coroutine 中协程定义了四种状态，整个运行期间，也是根据这四种状态进行轮转。

简单点说，就是每个协程的调用信息保存在堆内，然后共享一个栈进行运行。

有个疑问： 为什么要有这个S->stack结构？

context.uc_stack.ss_sp = stack; 这行代码是设置协程的栈的起始位置为 stack，也就是 stack 数组的起始地址。这个 stack 数组就是协程的运行时栈，协程可以在这个栈上存储自己的局部变量和函数调用关系。

我的理解是专门指定一个空间，让其将上下文的信息放进去。

强烈推荐这篇：微信 libco 协程库源码分析

本文学习自：云风 coroutine 协程库源码分析