zbh的协程学习总结 c++

IO发展历程1 从同步IO谈起
? 长期以来，使用C/C++编写服务器程序的时候，使用的往往是多进程模式：一个父进程负责 accept 传入连接，然后 fork 一个子进程处理；或者是一个父进程创建了一个 socket 之后，fork 出多个子进程同时执行 accept 和处理，（例如Apache Http Server 2.4版本前，采用per request per thread模型2)。
? 这时候一般使用同步IO来处理，也就是使用阻塞的read、write，当进程被IO阻塞的时候挂起进程。好处是这种同步开发方式和人类思维相近，编写简单。坏处之一是效率低：随着C10K问题3的提出，在高并发场景下，cpu忙于进程或线程的切换，大量cpu时间被浪费。
异步IO来临
? 所谓异步IO，就是发起IO请求的线程不等IO操作完成，就继续执行随后的代码，IO结果用其他方式通知发起IO请求的程序。
? 最简单的情形就是通过设置O_NONBLOCK标志使用非阻塞的read、write，定时轮询资源是否可用。如果资源不可用，会返回相应的错误如EAGAIN。
? 但是在并发量很大时，cpu同样沦陷于轮询中，效率很低（主要是每次都要进行系统调用，开销大）。有没有什么方法让我们被动接收资源就绪的通知，而不是逐个主动轮询呢？在Linux下，我们可以使用select/poll/epoll来同时监听多个文件描述符，看似将非阻塞IO阻塞在了select/poll/epoll中，但是这种方式监听了多个描述符，而且避免了大部分无用的系统调用开销。
? 好处是高效：没有切换上下文的开销、没有进程间同步、通信问题，但是坏处是开发不便。异步IO需要配合基于事件驱动的异步开发模式。在EventLoop中，什么事件到来，就调用相应的回调函数进行处理，和传统同步思维有很大不同。(std::function满天飞～，容易破坏逻辑的连续性)。
协程登场
? 有种说法是协程是一种轻量级的线程，这是针对它不需要操作系统调度介入、开销小而言，但是我觉得不够准确。本质上，协程就是一个可以暂停的函数，遇到阻塞的时候，就让出cpu，当条件满足了就继续恢复运行。
? 从另一个角度来说，协程和进程、线程的区别就是两点 – 是否隔离变量，是否自动切换运行上下文。满足这两点区别的，就是协程。

进程：变量隔离，自动切换运行上下文
线程：不变量隔离，自动切换运行上下文切换
协程：不变量隔离，不自动切换运行上下文切换

在不隔离变量的情况下，自动切换运行上下文，简直就是 bug 之源，这两件事根本就不该同时支持。以后线程这个东西，应该被扫进历史的垃圾堆，线程带来的便利，比起它挖出的坑相比，不足一提。4

? 之所以在协程的总结小文中上来就谈到这么多IO相关内容，是因为C++中的协程和IO有很大关系。协程适用于IO密集型应用，而非CPU密集型应用。协程的暂停等待条件满足后恢复执行的特点天然适合IO。当文件描述符不可读/不可写时，让出cpu；可读/可写后被恢复执行，继续处理相应逻辑。让出cpu的动作与同步IO中进程/线程被阻塞而挂起类似，但是避免了进程/线程调度开销，而保留了同步IO的同步开发方式。这正是协程的强大之处: 并发 + 同步5。
协程分类6 协程可以按照以下 3 个角度进行分类

对称/不对称：前者提供让出执行权的原语，后者提供唤起协程和把执行权转交给调用者的原语
第一类/受限：协程是不是第一类对象
有栈/无栈：协程能在调用栈的任意处挂起，还是只能在协程的函数体内挂起

这里看第三个分类的区别：

文章图片

此处「有栈」和「无栈」的含义不是指协程在运行时是否需要栈，对于大多数语言来说，一个函数调用另一个函数，总是存在调用栈的；
而是指协程是否可以在其任意嵌套函数中被挂起，此处的嵌套函数可以理解为子函数、匿名函数等。显然有栈协程是可以的，而无栈协程则不可以。
它们的主要区别是：有栈协程能在调用栈的任意处挂起，无栈协程只能在协程的函数体内挂起。
有栈协程
很多地方又把协程称为 subroutine，subroutine 是什么，就是函数。上古时期的计算机科学家们早就给出了概念，coroutine 就是可以中断并恢复执行的 subroutine，从这个角度来看协程拥有调用栈并不是一个奇怪的事情。
【zbh的协程学习总结】我们再来思考 coroutine 与 subroutinue 相比有什么区别，你会发现区别仅有一个，就是 coroutinue 可以中断并恢复，对应的操作就是 yield/resume，这样看来 subroutinue 不过是 coroutinue 的一个子集罢了。
也就是说把协程当做一个特殊的函数调用，有栈协程就是我们理想中协程该有的模样。
既然把其当做一个特殊的函数调用，对我们来说最严峻的挑战就是如何像切换函数一样去切换协程，难点在于除了像函数一样切换出去，还要在某种条件满足的时候切换回来，我们的做法可以是在协程内部存储自身的上下文，并在需要切换的时候把上下文切换就可以了。
Go 有栈协程 Go 语言的出现提供了一种新的思路。
Go 语言的协程则相当于提供了一种很低成本的类似于多线程的执行体。
在Go 语言中，协程的实现与操作系统多线程非常相似。操作系统一般使用抢占的方式来调度系统中的多线程（即自动分配调度），而 Go 语言中，依托于操作系统的多线程，在运行时刻库中实现了一个协作式的调度器。
这里的调度真正实现了上下文的切换，简单地说，Go 系统调用执行时，调度器可能会保存当前执行协程的上下文到堆栈中。然后将当前协程设置为睡眠，转而执行其他的协程。
这里需要注意，所谓的 Go 系统调用并不是真正的操作系统的系统调用，而是 Go 运行时刻库提供的对底层操作系统调用的一个封装。
举例说明：Socket recv。我们知道这是一个系统调用，Go 的运行时刻库也提供了几乎一模一样的调用方式，但这只是建立在 epoll 之上的模拟层，底层的 socket 是工作在非阻塞的方式，而模拟层提供给我们了看上去是阻塞模式的 socket。读写这个模拟的 socket 会进入调度器，最终导致协程切换。
目前 Go 调度器实现在用户空间，本质上是一种协作式的调度器。这也是为什么如果写了一个死循环在协程里，则协程永远没有机会被换出，一个 Processor 相当于就被浪费掉了。
有栈的协程和操作系统多线程是很相似的。考虑以下伪代码：

Go func routine() int { var a = 5 sleep(1000) a += 1 return a }

sleep 调用时，会发生上下文的切换，当前的执行体被挂起，直到约定的时间再被唤醒。
局部变量 a 在切换时会被保存在栈中，切换回来后从栈中恢复，从而得以继续运行。
所谓有栈就是指执行体本身的栈。每次创建一个协程，需要为它分配栈空间。究竟分配多大的栈的空间是一个技术活。分的多了，浪费，分的少了，可能会溢出。Go在这里实现了一个协程栈扩容的机制，相对比较优雅的解决了这个问题。
无栈协程
相比于有栈协程直接切换栈帧的思路，无栈协程在不改变函数调用栈的情况下，采用类似生成器（generator）的思路实现了上下文切换。
那么所谓的无栈协程是什么呢？其实无栈协程的本质就是一个状态机（state machine），可以看做是有状态的函数（generator），每次执行时会根据当前的状态和输入参数，得到（generate）输出，但不一定为最终结果。
即：将由 yield 切割的代码组成一个状态机（其实就是迭代器），切换状态 next() 就是执行下一块，如下：

Js function* anotherGenerator(i) { yield i + 1; yield i + 2; yield i + 3; }

Rust 的无栈协程7 无栈协程顾名思义就是不使用栈和上下文切换来执行异步代码逻辑的机制。这里异步代码虽然是异步的，但执行起来看起来是一个同步的过程。
从这一点上来看 Rust 协程与 Go 协程也没什么两样。举例说明：

Rust async fn routine() { let mut a = 5; sleep(1000).await; a = a + 1; a }

几乎是一样的流程。Sleep 会导致睡眠，当时间已到，重新返回执行，局部变量 a 内容应该还是5。
Go协程是有栈的，所以这个局部变量保存在栈中，而 Rust 是怎么实现的呢？答案就是 Generator 生成的状态机。
Generator 和闭包类似，能够捕获变量 a，放入一个匿名的结构中，在代码中看起来是局部变量的数据 a，会被放入结构，保存在全局（线程）栈中。
另外值得一提的是，Generator 生成了一个状态机以保证代码正确的流程。从 sleep.await 返回之后会执行 a= a + 1 这行代码。async routine() 会根据内部的 .await 调用生成这样的状态机，驱动代码按照既定的流程去执行。
按照一般的说法。无栈协程有很多好处。首先不用分配栈。因为究竟给协程分配多大的栈是个大问题。特别是在32位的系统下，地址空间是有限的。每个协程都需要专门的栈，很明显会影响到可以创建的协程总数。其次，没有上下文切换，性能也更好一些。
总结8

有栈协程：用rsp栈寄存器来索引局部变量，上下文是协程私有的栈。访问上下文数据也就是局部变量的时候，我们无需显式的使用栈寄存器+偏移量来访问，而是直接访问变量名。
无栈协程：把栈指针寄存器换成this指针，将局部变量捕获作为对象成员，整个上下文打包成一个generator结构体对象，可以称之为面向对象的上下文。
用this来索引对象的成员变量，上下文就是对象自己。访问上下文数据也就是成员变量的时候，我们无需显式的使用this+成员偏移量（或者变量名）来访问（通过语法糖），而是直接访问变量名。

C++协程设计9 ? 由于目前C++标准对协程支持还比较简陋（C++20中引入了协程，但是较为底层+没有好用的库，距离可用还有较远距离，乐观来看在C++23中 Asio进入标准库后有很大发挥空间），暂时没有一个标准协程库，因此市面上诞生了很多不同类型协程库，设计目标不同，实现方式也有差异。下面就进行一个简短的总结。
协程设计的不同层次
首先，根据协程库的完善程度，总体上可以分为四个类型：
1.API 级：这一级别实现协程上下文切换 api，或添加一些便于使用的封装；特点：没有协程调度。
代表作：boost.context, boost.coroutine, ucontext(unix), fiber(windows)
这一层次的协程库，仅仅提供了一个底层 api，要想拿来做项目，还有非常非常遥远的距离；不过这些协程 api 可以为我们实现自己的协程库提供一个良好的基础。
2.玩具级：这一级别实现了协程调度，无需用户手动处理协程上下文切换；特点：没有 HOOK
代表作：libmill
这一层次的协程库，实现了协程调度(类似于操作系统有了进程调度机制); 稍好一些的意识到了阻塞网络 io 与协程的不协调之处，自己实现了一套网络 io 相关函数；
但是这也意味着涉及网络的第三方库全部不可用了。hiredis 不能用了，mysqlclient 不能用了。放弃整个 C/C++生态全部自己轮，这个玩笑开的有点大，所以只能称之为“玩具级”。
3.工业框架级：这一级别以正确的方式 HOOK 了网络 io 相关的 syscall，可以不改代码的兼容大多数第三方库
代表作：libco、libgo
这一层次的协程库，能够模拟被 hook 的 syscall 的行为，能够兼容任何网络 io 行为的同步模型的第三方库。但由于 C++的灵活性，用户行为是不受限的，所以依然存在几个边边角角的难点需要开发者注意：没有 gc（开发者要了解协程的调度时机和生命期），TLS 的问题，用户不按套路出牌、把逻辑代码 run 在协程之外，粗粒度的线程锁等等。
（ps : libco的开源社区很不活跃，我还提了个pr 10）
4.语言级代表作：golang
这一层次的协程库，开发者的一切行为都是受限行为，可以实现无死角的完善的协程。
C++协程实现方案
回想协程的暂停然后恢复执行的本质，我们可以总结下协程函数的执行特点（协程函数指协程上运行的函数，类似于线程执行函数，即pthread_create中传入的函数）11：

首次调用协程函数，会从堆中分配一个协程上下文，调用方的返回地址、入口函数、交出控制权等信息保存在协程上下文中
? 实际上，这个协程上下文很值得一提。对于一个控制流程而言，所需要的环境不外乎 CPU上下文（寄存器信息）、栈、堆以及所要执行的代码四个要素，代码执行位置信息由RIP寄存器决定(以x86-64为例)，而且从堆中分配出的内存地址信息一般也会直接或间接的保存在运行栈中，所以，这四个要素可以简化为寄存器和栈两个要素（这也正是垃圾回收算法之一的可达性分析算法的原理，以栈中和寄存器中对象作为垃圾回收的起点，遍历所有引用链，不可达的对象就视作垃圾予以回收。而另一种垃圾回收算法是我们熟悉的引用计数算法12）。
下面是libco中(x86-64架构)协程上下文的设计，也正是我们刚刚所讨论的情形。

struct coctx_t { void *regs[ 14 ]; // 寄存器信息 size_t ss_size; // 协程栈的大小 char *ss_sp; // 协程栈低地址};

当协程中途交出控制权后，协程上下文不会删除（类似于线程切换时候保存上下文环境）。
当协程再次获得控制权后，会自动从协程上下文恢复调用环境，然后从上一次交出控制权的下一条语句继续执行。
协程函数运行完毕后，协程上下文将被删除。

结合协程函数的运行特点，根据C++中协程的不同实现方式，可以分为有栈协程和无栈协程。
有栈协程： ? 技术路线：一个线程可以创建大量协程，每个协程都会保留一个私有栈（实际上，不一定每个协程都有一个独立栈，也可能是若干个协程共用一个栈以节省内存），协程一旦执行到异步IO处，就会主动交出控制权。同一线程的所有协程以串行方式协作执行，没有数据争用的问题。
? 一般来说，一个线程可以包含多个有栈协程，而且线程本身也不外乎一个控制流，也可以视作一个协程，线程栈即为协程栈，一般称作主协程。
? 相比无栈协程，有栈协程无需修改语言标准和编译器，利用操作系统提供的API即可完成（例如Linux下的ucontext一族的{get,set,make,swap}context系统调用，但是这个实现方案性能堪忧，主要是ucontext需要需要多一次系统调用来保存信号掩码13，需要几百～上千个cycle，或者也可以采用手写汇编完成，例如libco，也可以利用强大的boost库中的coroutine21415)。另外，有栈协程支持协程嵌套，也就是在协程函数中可以继续调用其他协程函数，这是因为有了协程栈，调用信息可以入栈得到保存。正是因为有栈协程的这些有点，目前主流的协程库都是以有栈协程为主。
协程栈的实现方案协程栈的实现也有很多方案可供选择。

静态栈(Static Stack)静态栈就是固定大小的栈，一般每个协程有一个固定大小的独立栈。这种方式实现起来最为简便，但是实际使用中，栈的大小往往难以事先确定。太大了内存利用率低，造成内存浪费，且由于内存瓶颈，无法创建更多的协程，并发不够；太小了则容易造成栈溢出。
分段栈(Segmented Stack)gcc 提供的“黄金链接器”支持一种允许栈内存不连续的编译参数16，实现原理是在每个函数调用开头都插入一段栈内存检测的代码，如果栈内存不够用了就申请一块新的内存，作为栈内存的延续。
这种方案本应是最佳的实现，但如果遇到的第三方库没有使用这种方式来编译(注意:glibc 也是这里提到的”第三方库")，那就无法在其中检测栈内存是否需要扩展，栈溢出的风险很大。
拷贝栈(Copy Stack)每次检测到栈内存不够用时，申请一块更大的新内存，将现有的栈内存 copy 过去，就像 std::vector 那样扩展内存。
在某些语言上是可以实现这样的机制，但 C++ 是有指针的，栈内存的 Copy 会导致指向其内存地址的指针失效；又因为其指针的灵活性(可以加减运算)，修改对应的指针成为了一种几乎不可能实现的事情，这也正是C++难以实现追踪式垃圾回收的原因。
如下面例子所示17，对指针p做了自加和自减操作。这在C/C++中是被认为合理的。因为指针有所指向的对象，所以自加或自减可以找到"下一个"同样的对象。

int main(){ int *p = new int; p += 10; //移动指针 p -= 10; *p = 10; }

共享栈(Shared Stack)申请一块大内存作为共享栈(比如：8MB)，每次开始运行协程之前，先把协程栈的内存 copy 到共享栈中，运行结束后再计算协程栈真正使用的内存，copy 出来保存起来，这样每次只需保存真正使用到的栈内存量即可。
这种方案极大程度上避免了内存的浪费，做到了用多少占多少，同等内存条件下，可以启动的协程数量更多，libco中采用的就是这种方法。不过缺点是memcpy保存恢复协程栈需要额外的时间。

协程调度方案根据调度方式可以将调度方案分为对称调度和非对称调度。
非对称调度非对称调度是指所有创建的协程都必须和调用者进行切换，有着明确的调用和返回关系，路径如下：

主协程->协程A->协程B->协程A->主协程->协程C.....

如果不进行协程嵌套的话，所有协程都必须和主协程进行切换，因此，主协程也被称作调度协程。
对称调度对称调度是指所有协程之间可以自由切换，一般需要调度器的参与，让出CPU后由调度器决定下一个运行哪个协程，如golang中的goroutine调度。
这两种方式仅仅是单个线程上进行的最基本的协程调度方式，如果追求多核扩展性还可以配合任务窃取(Work Stealing18)设计更复杂的调度方式。
无栈协程 ? 技术路线：将异步IO封装到协程函数中，协程函数发起异步IO后，立即保存执行环境，然后吧控制权交给调用方（Caller），调用方继续往下执行；异步IO完成后，负责处理IO完成事件的回调函数获得控制权，回调函数再把控制权转交给发起IO的协程，发起IO的协程首先恢复执行环境，然后从上一次交出控制权的下一条语句继续往下执行
? 无栈协程中每个协程的执行环境，仅需要包含调用栈(Call Stack)的顶层栈帧(Top Call Stack Frame)，而非整个调用栈，空间开销极小，但是因此不能嵌套调用。另外，无栈协程需要语言标准和编译器的支持(编译器将协程翻译成状态机)，C++20中引入的就是无栈协程。
HOOK模块
基本原理与实现方式 ? hook意为钩子，在Linux下，可以通过hack掉glibc的read、write等IO操作函数，将其替换自己的实现，达到额外进行某些操作的目的。例如，我们实现自己的协程框架的时候，一般希望当文件描述符不可读时自动让出CPU，而不是将协程阻塞（协程阻塞会阻塞运行协程的线程！），我们就可以通过劫持read函数，替换为自己的实现达到该目的。
? 基本原理是使用LD_PRELOAD环境变量，影响程序运行时的链接。因为linux下的动态连接器存在规则：当一个符号加入全局符号表的时候，如果相同的符号名已经存在，则后面加入的符号被忽略。因为glic是最后链接的，那么通过定义在程序运行前优先加载的动态链接库，就可以达到用自己的.so函数替换原有的函数。
? 因为hook的目的都是在原有的函数行为上增加一些额外操作，并不需要改变函数的行为（例如通过hook malloc 和 free函数，我们可以检查是否所有分配的内存都得到了释放)，所以我们一般需要获取被hook的函数。这一操作是通过dlsym函数实现的(Linux 下的dlfcn.h 给予了我们运行时加载和处理动态链接库的能力）。
? 如下所示，使用 RTLD_NEXT 来获取系统glibc的 malloc 函数地址，由于待会使用 LD_PRELOAD 来优先加载我们创建的 so 文件，因而系统的 libc.so.6 排在第二位，也就是 next 19。

#include void* (*g_sys_malloc)(size_t) = dlsym(RTLD_NEXT, "malloc"); void* malloc(size_t size) { void *p = g_sys_malloc(size); printf("in malloc hook function ...\n"); return p; }

基本守则与具体实践 ? HOOK中有一条重要的基本守则：HOOK接口表现出来的行为与被HOOK的接口保持100%一致。因为在协程中还是需要兼容使用第三方的同步网络库，接口行为不一致会导致第三方库莫名其妙的错误。
? HOOK的基本思想是凡是有可能引起阻塞的系统调用或库函数都要被hook住，因为协程本身是不能被阻塞的，一旦阻塞运行协程的线程也会跟着阻塞，其他协程也就无法得到运行。
? 以libco为例，常用的read、recv、write、send、connect等都得到了hook，与epoll配合使用。以read函数为例，我们自己的read函数的行为应该如下（将glibc中的read函数记为g_sys_read）

ssize_t read(int fd, void *buf, size_t count){ 将当前协程注册到定时器上，注册回调函数用于将来处理 g_sys_read() 函数的读超时调用 epoll_ctl() 将自己(op = EPOLLIN, fd = current fd)注册到当前执行环境的 epoll 实例上，注册回调函数用于处理可读事件调用 co_yield_env 函数让出 CPU等到该协程被主协程重新“唤醒”后继续执行 }

此外，为了满足用户的不同需求，我们还需要进行判断

如果用户没有enable hook，那么不使用协程，那么直接返回g_sys_read。
如果用户enable hook，但是将文件描述符设置为NonBlock，这代表用户自己处理IO，返回非阻塞的g_sys_read。
其他情况，使用自己的read函数hookg_sys_read函数。

补充5 纤程
? 事实上，仅仅有了协程还不够，如果想要真正使用，还需要额外的调度器。否则需要使用者自己yield、resume非常不友好。前面我们无意中将协程和调度机制绑定了在一起，但是实际上并不是这样的。除了我们目前一直讨论的协程(coroutine)外，还有另外一种叫做fiber的纤程，fiber可以视作是实现了调度机制的协程。在boost库中，boost::fiber = boost::coroutine2+调度机制。
正确使用姿势
? 协程适用于IO密集的场景，主要是不断的创建协程，让调度器调度运行，在协程运行过程对于一些会阻塞的条件，做一个非阻塞的检测中,发现条件不满足就让出cpu，这就是常见轮转+非阻塞同步。
无栈协程的实现参考
? 无栈协程由于每个中断点都是确定，那其实只需要将函数的代码再进行细分，保存好局部变量，做好调用过程的状态变化, 下面就将一个协程函数fn进行切分后变成一个Struct，这样的实现相对于有栈协程而言使用的内存更少。
? 从下面代码就不难理解，为什么我们说无栈协程本质上是一个状态机了，也能够理解，为什么无栈协程需要语言和编译器的支持（需要编译器将代码翻译成状态机）。

void fn(){ int a, b, c; a = b + c; yield(); b = c + a; yield(); c = a + b; }----------------------------分割线--------------------------------- Struct fn{ int a, b, c; int __state = 0; void resume(){ switch(__state) { case 0: return fn1(); case 1: return fn2(); case 2: return fn3(); } }void fn1(){ a = b + c; }void fn2(){ b = c + a; }void fn3(){ c = a + b; } };

libco实现分析以libco源码为例，作一个简短的分析
libco采用共享栈作为协程栈，栈的大小128KB，每次运行协程前将保存的协程栈拷贝到共享栈上面运行。
如下是共享栈的结构，可以看到，一个共享栈中实际上并非只有一个栈，而是可以指定数目。如果有多个共享栈，在分配的时候采用round-robin算法选择一个。

struct stShareStack_t { unsigned int alloc_idx; //当前下标，用于round-robin算法 int stack_size; //共享栈大小 int count; // 共享栈数量 stStackMem_t** stack_array; //若干个共享栈 }; static stStackMem_t* co_get_stackmem(stShareStack_t* share_stack) { if (!share_stack) { return NULL; } int idx = share_stack->alloc_idx % share_stack->count; //采用round-robin算法分配共享栈 share_stack->alloc_idx++; return share_stack->stack_array[idx]; }

接下来去看共享栈具体内存的结构，也没有什么特殊之处。

struct stStackMem_t { stCoRoutine_t* occupy_co; // 当前运行的协程stack_bp ---高地址 int stack_size; //栈大小，默认128KB| char* stack_bp; //stack_buffer + stack_size| char* stack_buffer; // 栈顶（低地址）stack_buffer --- 低地址 };

下面就是协程的主体结构，基本上都是最基本的信息，没有什么额外功能（例如信号掩码之类）。将必要信息加在了注释中方便理解，不再一一赘述。

struct stCoRoutine_t { stCoRoutineEnv_t *env; // 当前线程的协程运行环境 pfn_co_routine_t pfn; // 协程函数==typedef void* (*pfn_co_routine_t)(void*); void *arg; // 函数参数 coctx_t ctx; // 协程上下文char cStart; // 标志位，1为协程已经启动，否则为0 char cEnd; // 标志为，1为协程已经结束，否则为0 char cIsMain; // 是否是主协程 char cEnableSysHook; // 是否启用hook 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]; };

struct stCoRoutineEnv_t { stCoRoutine_t *pCallStack[ 128 ]; // 作为协程调用栈使用，支持协程嵌套调用。可以看到，最多支持128个协程嵌套 int iCallStackSize; // 记录入栈的协程数量,便于resume()（总是resume()栈顶的协程） stCoEpoll_t *pEpoll; // 封装的epoll结构体，每个线程一个，阻塞时将当前协程监听的事件注册到该epoll上面，同时让出CPU//for copy stack log lastco and nextco stCoRoutine_t* pending_co; // 要执行的下个协程 stCoRoutine_t* occupy_co; // 要执行的上个协程 };

struct coctx_t {void *regs[ 14 ]; // 保存寄存器组 size_t ss_size; // 协程实际用到的栈的大小 char *ss_sp; // 保存协程栈的首地址};

下面以一个最简单的、什么都不做、只输出echo信息的demo来看libco的主要执行流程

#include #include #include #include #include #include #include #include "../coctx.h" #include "../co_routine.h" #include "../co_routine_inner.h"void* RoutineFunc(void* args){ co_enable_hook_sys(); int* routineid = (int*)args; while (true) { printf("echo from routine %d\n", *routineid + 1); co_yield_ct(); // 协程暂停执行 } return NULL; }int main() { stShareStack_t* share_stack= co_alloc_sharestack(1, 1024 * 128); // 1. 创建共享栈 stCoRoutineAttr_t attr; attr.stack_size = 0; attr.share_stack = share_stack; stCoRoutine_t* co[2]; int routineid[2]; for (int i = 0; i < 2; i++) { routineid[i] = i; co_create(&co[i], &attr, RoutineFunc, routineid + i); // 2. 创建协程 }while(true){ std::cout << "before resume1" << std::endl; co_resume(co[0]); // 3. 执行协程0 std::cout << "after resume1 and before resume2" << std::endl; co_resume(co[1]); // 4. 执行协程1 std::cout << "after resume2" << std::endl; } return 0; }

输出为

before resume1 echo from routine 1 after resume1 and before resume2 echo from routine 2 after resume2 ...

第一步是创建共享栈

stShareStack_t* co_alloc_sharestack(int count, int stack_size) { stShareStack_t* share_stack = (stShareStack_t*)malloc(sizeof(stShareStack_t)); // 1. 创建共享栈结构体 share_stack->alloc_idx = 0; share_stack->stack_size = stack_size; //alloc stack array share_stack->count = count; stStackMem_t** stack_array = (stStackMem_t**)calloc(count, sizeof(stStackMem_t*)); // 2. 创建共享栈数组 for (int i = 0; i < count; i++) { stack_array[i] = co_alloc_stackmem(stack_size); // 3. 创建共享栈实际空间 } share_stack->stack_array = stack_array; return share_stack; }

第二步就是创建协程了, 进入co_create函数

int co_create( stCoRoutine_t **ppco,const stCoRoutineAttr_t *attr,pfn_co_routine_t pfn,void *arg ) { if( !co_get_curr_thread_env() ) { co_init_curr_thread_env(); // 如果是第一次创建协程，尚未初始化当前线程的协程运行环境， }// 那么首先初始化协程运行环境 stCoRoutine_t *co = co_create_env( co_get_curr_thread_env(), attr, pfn,arg ); // 创建协程 *ppco = co; return 0; }

首先判断是不是已经准备好了当前线程环境（第一个协程负责初始化线程环境）
线程环境是一个静态TLS变量stCoRoutineEnv_t，再次回顾一下这个结构

struct stCoRoutineEnv_t { stCoRoutine_t *pCallStack[ 128 ]; // 作为协程调用栈使用，支持协程嵌套调用。可以看到，最多支持128个协程嵌套 int iCallStackSize; // 记录入栈的协程数量,便于resume()（总是resume()栈顶的协程） stCoEpoll_t *pEpoll; // 封装的epoll结构体，每个线程一个，阻塞时将当前协程监听的事件注册到该epoll上面，同时让出CPU//for copy stack log lastco and nextco stCoRoutine_t* pending_co; // 要执行的下个协程 stCoRoutine_t* occupy_co; // 要执行的上个协程 };

如果是创建第一个协程，那么首先初始化协程运行环境

void co_init_curr_thread_env() { gCoEnvPerThread = (stCoRoutineEnv_t*)calloc( 1, sizeof(stCoRoutineEnv_t) ); stCoRoutineEnv_t *env = gCoEnvPerThread; env->iCallStackSize = 0; struct stCoRoutine_t *self = co_create_env( env, NULL, NULL,NULL ); // 创建主协程（即调度协程），也就是将线程本身视作协程 self->cIsMain = 1; env->pending_co = NULL; env->occupy_co = NULL; coctx_init( &self->ctx ); // 初始化协程结构体==>memset(ctx, 0, sizeof(*ctx)); env->pCallStack[ env->iCallStackSize++ ] = self; // 将主协程入栈，因此任何时候，无论如何出栈，最终都会返回主协程，而主协程永不退出(永不yield)stCoEpoll_t *ev = AllocEpoll(); // 注册线程的epoll实例 SetEpoll( env,ev ); }

接下来真正开始创建协程

struct stCoRoutine_t *co_create_env( stCoRoutineEnv_t * env, const stCoRoutineAttr_t* attr, pfn_co_routine_t pfn,void *arg ) {// 设置协程属性，主要是设置协程栈的大小 stCoRoutineAttr_t at; if( attr ) { memcpy( &at,attr,sizeof(at) ); } if( at.stack_size <= 0 ) { at.stack_size = 128 * 1024; // 栈最小为128KB } else if( at.stack_size > 1024 * 1024 * 8 ) { at.stack_size = 1024 * 1024 * 8; // 栈最大为8MB }if( at.stack_size & 0xFFF )// 栈大小4K对齐 { at.stack_size &= ~0xFFF; at.stack_size += 0x1000; }stCoRoutine_t *lp = (stCoRoutine_t*)malloc( sizeof(stCoRoutine_t) ); memset( lp,0,(long)(sizeof(stCoRoutine_t))); lp->env = env; // 设置协程环境 lp->pfn = pfn; // 设置协程函数 lp->arg = arg; // 设置协程参数stStackMem_t* stack_mem = NULL; if( at.share_stack ) { stack_mem = co_get_stackmem( at.share_stack); // 使用round-robin算法分配一个共享栈 at.stack_size = at.share_stack->stack_size; } else { stack_mem = co_alloc_stackmem(at.stack_size); // 或者使用独立栈 } lp->stack_mem = stack_mem; lp->ctx.ss_sp = stack_mem->stack_buffer; // 给协程上下文注册栈地址以及栈大小 lp->ctx.ss_size = at.stack_size; lp->cStart = 0; lp->cEnd = 0; lp->cIsMain = 0; lp->cEnableSysHook = 0; lp->cIsShareStack = at.share_stack != NULL; lp->save_size = 0; lp->save_buffer = NULL; return lp; }

创建协程至此就暂时告一段落，不过创建后协程还并没有跑起来，需要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 ); // 将要运行的协程和当前协程进行切换 }

在我们继续探索之前，需要回忆一下函数调用和汇编的知识。同时看一下14个寄存器的规定20

push param // 压入函数参数 call // 压入调用函数的下一条指令的地址 push ebp// 栈帧开始 mov ebp, esp // esp 指向栈顶元素的靠近栈的的地址 push ... // 保存寄存器 (分为调用者保存和被调用者保存两部分，根据指令集的不同) //创建局部变量 pop ... //弹出寄存器 mov esp, ebp pop ebp ret // low | regs[0]: r15 | //| regs[1]: r14 | //| regs[2]: r13 | //| regs[3]: r12 | //| regs[4]: r9| //| regs[5]: r8| //| regs[6]: rbp | //| regs[7]: rdi | //| regs[8]: rsi | //| regs[9]: ret |//ret func addr //| regs[10]: rdx | //| regs[11]: rcx | //| regs[12]: rbx | // hig | regs[13]: rsp | 在64位下保存14个寄存器。我们知道X86架构下有8个通用寄存器，X64则有16个寄存器，那么为什么64位只使用保存14个寄存器呢？我们可以在coctx_swap.S中看到64位下缺少了对%r10, %r11寄存器的备份， x86-64的16个64位寄存器分别是：%rax, %rbx, %rcx, %rdx, %esi, %edi, %rbp, %rsp, %r8-%r15。其中：%rax 作为函数返回值使用 %rsp栈指针寄存器，指向栈顶 %rdi，%rsi，%rdx，%rcx，%r8，%r9 用作函数参数，依次对应第1参数，第2参数 %rbx，%rbp，%r12，%r13，%14，%15 用作数据存储，遵循被调用者保护规则，简单说就是随便用，调用子函数之前要备份它，以防被修改 %r10，%r11 用作数据存储，遵循调用者保护规则，简单说就是使用之前要先保存原值我们来看看两个陌生的名词调用者保护&被调用者保护：调用者保护：表示这些寄存器上存储的值，需要调用者(父函数)自己想办法先备份好，否则过会子函数直接使用这些寄存器将无情的覆盖。如何备份？当然是实现压栈(pushl),等子函数调用完成，再通过栈恢复(popl) 被调用者保护：即表示需要由被调用者(子函数)想办法帮调用者(父函数)进行备份

?
这些就足够了！

int coctx_make(coctx_t* ctx, coctx_pfn_t pfn, const void* s, const void* s1) { char* sp = ctx->ss_sp + ctx->ss_size - sizeof(void*); sp = (char*)((unsigned long)sp & -16LL); // 取得栈底地址（高地址，栈是从高向低生长）memset(ctx->regs, 0, sizeof(ctx->regs)); void** ret_addr = (void**)(sp); *ret_addr = (void*)pfn; // 为什么coctx_make（）中需要将返回函数的地址赋值给rsp？ // 后面会进行解答 ctx->regs[kRSP] = sp; // 设置寄存器ctx->regs[kRETAddr] = (char*)pfn; // 设置返回地址，即协程函数地址ctx->regs[kRDI] = (char*)s; // 设置函数参数 ctx->regs[kRSI] = (char*)s1; return 0; }

接下来就真正开始进行切换！

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; // 这里用了一个小trick，在栈上声明了一个char变量，其地址对齐后就是栈顶地址。 // 如果不理解，请看一下上面的函数调用栈帧图// get curr stack sp 以及对应的汇编 // char c; // lea 0x7(%rsp), %rdx这里rsp + 7 是因为char变量无需对齐，而rsp需要8字节对齐 // curr->stack_sp= &c; // mov %rdx, 0xb0(%rbp)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); // 恢复要运行的协程的栈 } } }

void save_stack_buffer(stCoRoutine_t* occupy_co) { ///copy out stStackMem_t* stack_mem = occupy_co->stack_mem; int len = stack_mem->stack_bp - occupy_co->stack_sp; // 栈底（高地址） - 栈顶（刚才通过定义char变量获得的低地址）即为实际使用的大小if (occupy_co->save_buffer) { free(occupy_co->save_buffer), occupy_co->save_buffer = NULL; }occupy_co->save_buffer = (char*)malloc(len); //malloc buf; occupy_co->save_size = len; memcpy(occupy_co->save_buffer, occupy_co->stack_sp, len); }

.globl coctx_swap #if !defined( __APPLE__ ) .typecoctx_swap, @function #endif coctx_swap: leaq (%rsp),%rax// 首先保存当前寄存器上下文到current->ctx中，current->ctx也就是第一个参数，根据传参规定第一个参数通过rdi传入 movq %rax, 104(%rdi) movq %rbx, 96(%rdi) movq %rcx, 88(%rdi) movq %rdx, 80(%rdi) movq 0(%rax), %rax movq %rax, 72(%rdi) movq %rsi, 64(%rdi) movq %rdi, 56(%rdi) movq %rbp, 48(%rdi) movq %r8, 40(%rdi) movq %r9, 32(%rdi) movq %r12, 24(%rdi) movq %r13, 16(%rdi) movq %r14, 8(%rdi) movq %r15, (%rdi) xorq %rax, %raxmovq 48(%rsi), %rbp// 从要运行的pending->ctx中恢复寄存器的值，pending->ctx也就是第二个参数，根据传参规定通过rsi传入 movq 104(%rsi), %rsp movq (%rsi), %r15 movq 8(%rsi), %r14 movq 16(%rsi), %r13 movq 24(%rsi), %r12 movq 32(%rsi), %r9 movq 40(%rsi), %r8 movq 56(%rsi), %rdi movq 80(%rsi), %rdx movq 88(%rsi), %rcx movq 96(%rsi), %rbx leaq 8(%rsp), %rsp pushq 72(%rsi)// regs[9] 中保存了协程函数运行地址，即压入栈，ret时即可成功跳转movq 64(%rsi), %rsi ret

读到这里，你可能会有一个疑问，为什么保存寄存器上下文的时候将rsp保存到了regs[kRETAddr]?换言之，为什么rsp保存了返回地址？同时，我在coctx_make()的注释中也留下了问题 “为什么coctx_make（）中需要将返回函数的地址赋值给rsp？“ 这需要函数调用的知识来解答。

--------- ret addr --------- old rbp<--- RBP --------- local var1 --------- local var2 --------- fcn param #2 --------- fcn param #1 --------- ret addr<--- RSP --------- old rbp ---------

如图是coctx_swap执行前的栈帧，可以看到，RBP仍然指向上个栈帧的基地址，RSP指向函数返回地址。
如果作为正常的函数调用，那么第一件事情就是要开辟自己的栈帧，也就是

push rbp mov rbp, rsp

这两个动作由编译器自动插入，然后才将rsp增长到栈底。但是如果没有仔细思考的话，很容易误以为rsp默认就指向栈顶，从而产生为什么rsp指向的是函数地址的疑惑。
显然在coctx_swap中，我们并没有开辟自己的栈帧，因此rsp指向的仍然是函数的返回地址，一切都得到了解释！
在协程中让出cpu，操作就简单的多啦

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

参考资料

[基于汇编的 C/C++ 协程]: https://cloud.tencent.com/dev... ?
[IO模型和基于事件驱动的IO多路复用模式]: https://zhuanlan.zhihu.com/p/... ?
[C10K problem]: https://en.wikipedia.org/wiki... ?
[Node.js 真的有协程吗？]https://www.zhihu.com/questio... ?
[深入浅出c++协程 ]:https://www.cnblogs.com/ishen... ?
[编程中的 “协程” 思想]https://alsritter.icu/posts/4... ?
[浅谈Rust和Golang协程设计]https://computeinit.com/archi... ?
[Node.js 真的有协程吗？]https://www.zhihu.com/questio... ?
[C++协程的近况、设计与实现]: https://www.jianshu.com/p/837... ?
[libco中内存浪费的pr]:https://github.com/Tencent/li... ?
[C++协程综述]:http://qiusuoge.com/16296.html ?
[GC垃圾回收的原理和涉及的几种算法]:https://segmentfault.com/a/11... ?
[Why ucontext is so slow]https://news.ycombinator.com/... ?
[What is the difference between Coroutine, Coroutine2 and Fiber?]https://stackoverflow.com/que... ?
[boost.coroutine2]https://www.boost.org/doc/lib... ?
[GCC split stacks]https://gcc.gnu.org/wiki/Spli... ?
[深入理解C++11:C++11新特性解析与应用 5.2.4节]https://book.douban.com/subje... ?
[Brpc bthread 简介]https://zhuanlan.zhihu.com/p/... ?
[linux hook机制研究]https://zhuanlan.zhihu.com/p/... ?
[libco源码解析1]https://blog.csdn.net/weixin_... ?