云风轻量级协程coroutine源码分析（linux系统下基于ucontext）

89 阅读 0 评论 59 点赞

我是靠谱客的博主生动芒果，最近开发中收集的这篇文章主要介绍云风轻量级协程coroutine源码分析（linux系统下基于ucontext），觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

云风coroutine简介

github源码地址：https://github.com/cloudwu/coroutine.git

源码下载后可以直接make，执行。示例程序清晰易懂，基本看完后大多数人都会使用了。

coroutine属于非常轻量级的协程实现，核心接口coroutine_open、coroutine_close、coroutine_new、coroutine_resume、coroutine_yield。
coroutine_open、coroutine_close：创建和关闭调度器。
coroutine_new：创建一个新的协程。
coroutine_resume：恢复指定协程运行。
coroutine_yield：阻塞当前协程，切换回调度器主流程。

看概念可能有些空，建议直接看 main.c 文件中的示例代码，5min即可读懂。

以下是核心源码介绍。

ucontext（在glibc库中实现，posix标准）

ucontext为用户层程序提供了一组上下文(context)切换的接口。所谓上下文切换，我们可以类比线程的切换，通过ucontext提供的接口，可以直接修改cpu中寄存器值，从而完成程序的跳转，类似线程的切换。

github源码地址：
https://github.com/lattera/glibc/blob/master/sysdeps/unix/sysv/linux/x86_64/getcontext.S
https://github.com/lattera/glibc/blob/master/sysdeps/unix/sysv/linux/x86_64/setcontext.S
https://github.com/lattera/glibc/blob/master/sysdeps/unix/sysv/linux/x86_64/swapcontext.S
https://github.com/lattera/glibc/blob/master/sysdeps/unix/sysv/linux/x86_64/makecontext.c
…

使用介绍见：
https://blog.csdn.net/m0_37329910/article/details/98471368
https://segmentfault.com/p/1210000009166339/read#2-_ucontext_u5206_u6790

getcontext：获取当前context
setcontext：切换到指定context
makecontext: 用于将一个新函数和堆栈，绑定到指定context中
swapcontext：保存当前context，并且切换到指定context

makecontext

#include <ucontext.h>
void makecontext(ucontext_t *ucp, void (*func)(), int argc, ...);

The makecontext() function modifies the context pointed to by ucp (which was obtained from a call to getcontext(3)). Before invoking makecontext(), the caller must allocate a new stack for this context and assign its address to ucp->uc_stack, and define a successor context and assign its address to ucp->uc_link.
When this context is later activated (using setcontext(3) or swapcontext()) the function func is called, and passed the series of integer (int) arguments that follow argc; the caller must specify the number of these arguments in argc. When this function returns, the successor context is activated. If the successor context pointer is NULL, the thread exits.

ucontext注意：

上下文切换，这部分其实也不难，调用接口即可。setcontext或者swapcontext就帮我们完成了所有寄存器值的切换。
如果不同上下文共用一个栈，则需要注意栈的恢复，如云风的coroutine中，调度器栈的恢复。
如果不同上下文使用不同的栈，栈内容不会被破坏，这种情况也就不需要关注第2步了。

核心数据结构

struct schedule {
	char stack[STACK_SIZE];  // 每个协程分配栈大小
	ucontext_t main;  // Posix接口标准中关于创建、保存、切换用户态上下文的API
	int nco;  // 当前协程个数
	int cap;  // 协程容量（上限）
	int running; // 调用器正在执行的协程ID，没有协程执行时等于-1
	struct coroutine **co; // 协程位置指针
};

struct coroutine {
	coroutine_func func; // 协程回调
	void *ud; // 回调函数参数
	ucontext_t ctx;  // 协程上下文
	struct schedule * sch; // 该协程位于的调度器
	ptrdiff_t cap; 
	ptrdiff_t size;
	int status;
	char *stack;
};

核心函数

coroutine_open

创建调度器instance并初始化。

struct schedule * coroutine_open(void) {
	struct schedule *S = malloc(sizeof(*S));
	S->nco = 0;
	S->cap = DEFAULT_COROUTINE; // DEFAULT_COROUTINE=16
	S->running = -1;
	S->co = malloc(sizeof(struct coroutine *) * S->cap);
	memset(S->co, 0, sizeof(struct coroutine *) * S->cap);
	return S;
}

coroutine_new

创建新的协程关联回调，并将该协程指针存入调度器中。

int coroutine_new(struct schedule *S, coroutine_func func, void *ud) {
	struct coroutine *co = _co_new(S, func , ud);
	 // 1.当前协程数量大于等于调度器容许的最大值，扩容为两倍，初始化新增的协程指针存储空间
     // 2.新创建的协程添加到调度器的协程指针存储空间中
     if (S->nco >= S->cap) {
		int id = S->cap;
        // realloc会保留原指针指向的内容，但是指针值可能会变化。
        // 参考 https://www.cnblogs.com/droidxin/p/3617854.html
		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 {
		int i;
		for (i=0;i<S->cap;i++) {
			int id = (i+S->nco) % S->cap;
            // 找到第一个为NULL的位置将新的协程指针存下来
			if (S->co[id] == NULL) {
				S->co[id] = co;
				++S->nco;
				return id;
			}
		}
	}
	assert(0);
	return -1;
}

// 创建一个协程instance并初始化
struct coroutine * _co_new(struct schedule *S , coroutine_func func, void *ud) {
	struct coroutine * co = malloc(sizeof(*co));
	co->func = func;
	co->ud = ud;
	co->sch = S;
	co->cap = 0;
	co->size = 0;
	co->status = COROUTINE_READY;
	co->stack = NULL;
	return co;
}
// 协程回调typdef为coroutine_func
typedef void (*coroutine_func)(struct schedule *, void *ud);

coroutine_resume

协程状态 COROUTINE_READY：创建新协程的上下环境，关联回调和栈信息。
协程状态 COROUTINE_SUSPEND：被阻塞的协程恢复执行。

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:
        // 获取一个上下文，只是为了给C->ctx赋初值
        // 后面会根据切换目标修改其中部分内容
		getcontext(&C->ctx);
		C->ctx.uc_stack.ss_sp = S->stack;
		C->ctx.uc_stack.ss_size = STACK_SIZE;
		
        // 调度器中上下文存入uc_link，配合后面的swapcontext将
        // 当前函数上下文存入main中，也就同时存入了uc_link中
		C->ctx.uc_link = &S->main; 
		S->running = id;
		C->status = COROUTINE_RUNNING;
		uintptr_t ptr = (uintptr_t)S;
		
        // 新协程上下文关联回调和参数
		makecontext(&C->ctx, (void (*)(void)) mainfunc, 2, 
                  (uint32_t)ptr, (uint32_t)(ptr>>32));
                  
        // 切换协程，当前执行上下文存入S->main中，将C->ctx载入
        // 寄存器，从而去执行C->ctx关联的回调      
		swapcontext(&S->main, &C->ctx);
		break;
	case COROUTINE_SUSPEND:
        // 协程恢复执行，首先将协程栈上保存的内容恢复到调度器栈中，然后置标志，协程切换
		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);
	}
}

// mainfunc为协程上下文关联的回调函数，其中会解析出调度器地址和将要执行的协程ID，并执行该协程中由用户
// 指定的回调函数。
// 注意这个函数不一定能一次性执行完，如果用户传入的回调函数（即C->func）中调用了coroutine_yield，则
// 会阻塞当前的函数，而直接跳转回 coroutine_resume的swapcontext(&S->main, &C->ctx);位置处继续执行。
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);
	_co_delete(C);
	S->co[id] = NULL;
	--S->nco;
	S->running = -1;
}

linux2.6.13内核中ucontext相关结构体的定义：

struct ucontext {
	unsigned long	  uc_flags;    
	struct ucontext  *uc_link;  //后继上下文
	stack_t		  uc_stack; //用户自定义栈
	struct sigcontext uc_mcontext; //保存当前上下文，即各个寄存器的状态
	sigset_t	  uc_sigmask; //保存当前线程的信号屏蔽掩码
};

typedef struct sigaltstack {
	void __user *ss_sp;
	int ss_flags;
	size_t ss_size;
} stack_t;

coroutine_yield

在协程回调函数中调用，从而调度出当前协程。

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);
}

static void _save_stack(struct coroutine *C, char *top) {
   // 这个变量的目的就是为了定位当前协程的栈顶指针
   // dummy这个变量现在本质上是存储在 coroutine_resume函数 COROUTINE_READY分支
   // 处理时赋值的栈上 C->ctx.uc_stack.ss_sp = S->stack，所以实际是存储在调度器
   // 的栈上
	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);
}