【网络通信 -- 直播】SRS -- 协程基础知识点总结(一)
本文共 14391 字，大约阅读时间需要 47 分钟。
【网络通信 -- 直播】SRS -- 协程基础知识点总结(一)
【1】协程简介
IO 同步操作的逻辑代码
IO 异步操作的逻辑代码
IO 异步操作与 IO 同步操作对比
协程主要解决的问题，提供同时具有异步性能与同步代码逻辑的解决方案；C/C++ 典型的协程库 NtyCo，；
【2】NtyCo 简介
【2.1】NtyCo 相关 API 简介
NtyCo 封装了若干接口，一类是协程本身的，一类是 posix 异步封装协程 API；
协程相关 API 函数接口
协程创建int nty_coroutine_create(nty_coroutine **new_co, proc_coroutine func, void *arg);协程调度器运行void nty_schedule_run(void);
POSIX 异步封装 API 函数接口
创建 socke 实例int nty_socket(int domain, int type, int protocol);接收请求int nty_accept(int fd, struct sockaddr *addr, socklen_t *len);接收数据ssize_t nty_recv(int fd, void *buf, size_t len, int flags);发送数据ssize_t nty_send(int fd, const void *buf, size_t len, int flags);关闭 socket 实例int nty_close(int fd);
【2.2】协程的实现
1. 协程的实现之创建协程
int nty_coroutine_create(nty_coroutine **new_co, proc_coroutine func, void *arg);
- 参数 1 : nty_coroutine **new_co，需要传入空的协程的对象，这个对象是由函数内部创建并返回的；
- 参数 2 : proc_coroutine func 协程的子过程；当协程被调度的时候，就会执行该函数；
- 参数 3 : void *arg 需要传入到新协程中的参数；
协程不存在亲属关系，都是一致的调度关系，接受调度器的调度；调用 create API 就会创建一个新协程，新协程就会加入到调度器的就绪队列中；
2. 协程的实现之实现 IO 异步操作
调度器与协程的上下文切换图示
在协程的上下文 IO 异步操作函数，步骤如下
- 1. 将 sockfd 添加到 epoll 管理中
- 2. 进行上下文环境切换，由协程上下文 yield 到调度器的上下文
- 3. 调度器获取下一个协程上下文，resume 新的协程
IO 异步操作的上下文切换的时序图
3. 协程的实现之回调协程的子过程
关键示例代码
static void nty_coroutine_init(nty_coroutine *co) { void **stack = (void **)(co->stack + co->stack_size); stack[-3] = NULL; stack[-2] = (void *)co; // ctx 协程的上下文 // 可以将回调函数地址存入 EIP 中，将相应参数存储到相应的参数寄存器中 co->ctx.esp = (void*)stack - (4 * sizeof(void*)); co->ctx.ebp = (void*)stack - (3 * sizeof(void*)); // CPU 寄存器 EIP 用于存储 CPU 运行下一条指令的地址 co->ctx.eip = (void*)_exec; // 设置回调函数入口 co->status = BIT(NTY_COROUTINE_STATUS_READY); } static void _exec(void *lt) {#if defined(__lvm__) && defined(__x86_64__) __asm__("movq 16(%%rbp), %[lt]" : [lt] "=r" (lt));#endif nty_coroutine *co = (nty_coroutine*)lt; co->func(co->arg); // co->status |= (BIT(NTY_COROUTINE_STATUS_EXITED) | BIT(NTY_COROUTINE_STATUS_FDEOF) | BIT(NTY_COROUTINE_STATUS_DETACH)); #if 1 nty_coroutine_yield(co);#else co->ops = 0; _switch(&co->sched->ctx, &co->ctx);#endif} 4. 协程实现之原语操作
协程的核心原语操作 : create, resume, yield；
协程的原语操作有 create 但却没有 exit 的分析
- 以 NtyCo 为例，协程一旦创建就不能由用户自己销毁，必须在子过程执行结束时便自动销毁协程的上下文数据；
- 以 _exec 执行入口函数返回而销毁协程的上下文与相关信息；
- co->func(co->arg) 是子过程，若用户需要长久运行协程，就必须要在 func 函数里面写入循环等操作；
- 因此 NtyCo 里面没有实现 exit 的原语操作
原语 create : 创建一个协程
// 1. 调度器是否存在，不存在也创建；调度器作为全局的单例；将调度器的实例存储在线程的私有空间 pthread_setspecific// 2. 分配一个 coroutine 的内存空间，分别设置 // coroutine 的数据项，栈空间，栈大小，初始状态，创建时间，子过程回调函数，子过程的调用参数// 3. 将新分配协程添加到就绪队列 ready_queue 中int nty_coroutine_create(nty_coroutine **new_co, proc_coroutine func, void *arg)
原语 yield : 让出 CPU
// 参数：需要恢复运行的协程实例// 调用后该函数也不会立即返回，而是切换到运行协程实例的 yield 的位置；// 返回是在等协程相应事务处理完成后，主动 yield 会返回到 resume 的地方；void nty_coroutine_yield(nty_coroutine *co)
原语 resume : 恢复协程的运行权
// 参数：需要恢复运行的协程实例// 调用后该函数也不会立即返回，而是切换到运行协程实例的 yield 的位置；// 返回是在等协程相应事务处理完成后，主动 yield 会返回到 resume 的地方；int nty_coroutine_resume(nty_coroutine *co)
5. 协程实现之切换
上下文切换，就是将 CPU 的寄存器暂时保存，再将即将运行的协程的上下文寄存器，分别 mov 到相对应的寄存器上，从而完成上下文的切换；
nty_cpu_ctx 结构体
// 关联相关寄存器的结构体typedef struct _nty_cpu_ctx { void *esp; void *ebp; void *eip; void *edi; void *esi; void *ebx; void *r1; void *r2; void *r3; void *r4; void *r5;} nty_cpu_ctx; _switch 函数 API
// 参数 1 : 即将运行协程的上下文，寄存器列表// 参数 2 : 正在运行协程的上下文，寄存器列表int _switch(nty_cpu_ctx *new_ctx, nty_cpu_ctx *cur_ctx);// _switch 返回后，执行即将运行协程的上下文，从而实现上下文的切换
_switch 函数实现代码分析
// %rdi 函数第一个参数， %rsi 函数第二个参数__asm__ (" .text \n"" .p2align 4,,15 \n"".globl _switch \n"".globl __switch \n""_switch: \n""__switch: \n"" movq %rsp, 0(%rsi) # save stack_pointer \n" // 保存栈指针到 cur_ctx 实例的 rsp 项" movq %rbp, 8(%rsi) # save frame_pointer \n" // 保存帧指针到 cur_ctx 实例的 rbp 项" movq (%rsp), %rax # save insn_pointer \n" // 将栈顶地址里面的值存储到 rax 寄存器中" movq %rax, 16(%rsi) \n" // 保存 rax，rbx，r12 - r15" movq %rbx, 24(%rsi) # save rbx,r12-r15 \n"" movq %r12, 32(%rsi) \n"" movq %r13, 40(%rsi) \n"" movq %r14, 48(%rsi) \n"" movq %r15, 56(%rsi) \n"" movq 56(%rdi), %r15 \n" // 将 rdi 对应偏移量中的值保存入相应的寄存器中" movq 48(%rdi), %r14 \n" // 即将 rdi 对应偏移量中的值保存入 new_ctx 的每一个项中" movq 40(%rdi), %r13 # restore rbx,r12-r15 \n"" movq 32(%rdi), %r12 \n"" movq 24(%rdi), %rbx \n"" movq 8(%rdi), %rbp # restore frame_pointer \n"" movq 0(%rdi), %rsp # restore stack_pointer \n"" movq 16(%rdi), %rax # restore insn_pointer \n" // 将指令指针 rip 的值存储到 rax 中" movq %rax, (%rsp) \n" // 将存储 rip 值的 rax 寄存器赋值给栈指针指向的地址处" ret \n" // 出栈，回到栈指针，执行 rip 指向的指令); 6. 协程实现之定义
协程的运行体 R 与运行体调度器 S
- 1. 运行体 R：包含运行状态 { 就绪，睡眠，等待 }，运行体回调函数，回调参数，栈指针，栈大小，当前运行体；
- 2. 调度器 S：包含执行集合 { 就绪，睡眠，等待 }；
6.1 运行体在多种状态集合的切换
协程的状态
- 创建新的协程，创建完成后，便加入到就绪集合，等待调度器的调度；
- 协程在运行完成后，进行 IO 操作，此时 IO 并未准备好，便进入等待状态集合；
- IO 准备就绪，协程开始运行，后续进行 sleep 操作，便进入到睡眠状态集合；
就绪 (ready)，睡眠 (sleep)，等待 (wait) 集合对应的数据结构
- 就绪 (ready) 集合并没有设置优先级的选型，所有的协程优先级一致，可以使用队列来存储就绪的协程，简称为就绪队列 (ready_queue)
- 睡眠 (sleep) 集合需要按照睡眠时长进行排序，因此采用红黑树来存储，简称睡眠树 (sleep_tree)；红黑树在工程中为 <key, value>, key 为睡眠时长，value 为对应的协程结点
- 等待 (wait) 集合，其功能是在等待 IO 准备就绪，等待 IO 是有时长的，因此等待 (wait) 集合采用红黑树存储，简称等待树 (wait_tree)
图示中，Coroutine 是协程的相应属性，status 表示协程的运行状态，并包含 sleep 与 wait 两棵红黑树以及 ready 队列；前提条件是不管何种运行状态的协程，都在就绪队列中，只是同时包含有其他的运行状态；
6.2 调度器与协程的功能界限
协程属性 : 每一协程都需要使用的而且可能会不同的属性；调度器属性 : 每一协程都需要的而且一致的数据；
协程的核心结构体
typedef struct _nty_coroutine { //private // 协程自身的上下文，需要保存 CPU 的寄存器 ctx nty_cpu_ctx ctx; // 子过程的回调函数 func proc_coroutine func; // 子过程回调函数的参数 arg 以及数据 data void *arg; void *data; // 协程自身的栈空间的大小 stack_size size_t stack_size; size_t last_stack_size; // 协程当前的运行状态 nty_coroutine_status status; // 调度器的全局对象 nty_schedule *sched; // 协程的创建时间 uint64_t birth; // 协程 id uint64_t id;#if CANCEL_FD_WAIT_UINT64 int fd; unsigned short events; //POLL_EVENT#else int64_t fd_wait;#endif char funcname[64]; struct _nty_coroutine *co_join; void **co_exit_ptr; // 协程自身的栈空间 void *stack; void *ebp; uint32_t ops; uint64_t sleep_usecs; // 当前运行状态的结点 RB_ENTRY(_nty_coroutine) sleep_node; RB_ENTRY(_nty_coroutine) wait_node; LIST_ENTRY(_nty_coroutine) busy_next; // TAILQ_ENTRY(_nty_coroutine) ready_next; TAILQ_ENTRY(_nty_coroutine) defer_next; TAILQ_ENTRY(_nty_coroutine) cond_next; TAILQ_ENTRY(_nty_coroutine) io_next; TAILQ_ENTRY(_nty_coroutine) compute_next; struct { void *buf; size_t nbytes; int fd; int ret; int err; } io; struct _nty_coroutine_compute_sched *compute_sched; int ready_fds; struct pollfd *pfds; nfds_t nfds;} nty_coroutine; 调度器的属性，需要有保存 CPU 的寄存器上下文 ctx，可以从协程运行状态 yield 到调度器运行；从协程到调度器用 yield，从调度器到协程用 resume；
调度器结构体
typedef struct _nty_coroutine_link nty_coroutine_link;typedef struct _nty_coroutine_queue nty_coroutine_queue;typedef struct _nty_coroutine_rbtree_sleep nty_coroutine_rbtree_sleep;typedef struct _nty_coroutine_rbtree_wait nty_coroutine_rbtree_wait;typedef struct _nty_schedule { uint64_t birth; nty_cpu_ctx ctx; void *stack; size_t stack_size; int spawned_coroutines; uint64_t default_timeout; struct _nty_coroutine *curr_thread; int page_size; int poller_fd; int eventfd; struct epoll_event eventlist[NTY_CO_MAX_EVENTS]; int nevents; int num_new_events; pthread_mutex_t defer_mutex; nty_coroutine_queue ready; nty_coroutine_queue defer; nty_coroutine_link busy; nty_coroutine_rbtree_sleep sleeping; nty_coroutine_rbtree_wait waiting; //private } nty_schedule; 7. 协程实现之调度器
调度器的实现，有两种方案，一种是生产者消费者模式，另一种是多状态运行；
7.1 生产者消费者模式
逻辑代码示例
while (1) { // 遍历睡眠集合，将满足条件的协程加入到 ready 队列 nty_coroutine *expired = NULL; while ((expired = sleep_tree_expired(sched)) != NULL) { TAILQ_ADD(&sched->ready, expired); } // 遍历等待集合，将满足条件的协程加入到 ready 队列 nty_coroutine *wait = NULL; int nready = epoll_wait(sched->epfd, events, EVENT_MAX, 1); for (i = 0; i < nready; i ++) { wait = wait_tree_search(events[i].data.fd); TAILQ_ADD(&sched->ready, wait); } // 使用 resume 恢复 ready 的协程运行权 while (!TAILQ_EMPTY(&sched->ready)) { nty_coroutine *ready = TAILQ_POP(sched->ready); resume(ready); }} 7.2 多状态运行
逻辑代码示例
while (1) { // 遍历睡眠集合，使用 resume 恢复 expired 的协程运行权 nty_coroutine *expired = NULL; while ((expired = sleep_tree_expired(sched)) != NULL) { resume(expired); } // 遍历等待集合，使用 resume 恢复 wait 的协程运行权 nty_coroutine *wait = NULL; int nready = epoll_wait(sched->epfd, events, EVENT_MAX, 1); for (i = 0; i < nready; i ++) { wait = wait_tree_search(events[i].data.fd); resume(wait); } // 使用 resume 恢复 ready 的协程运行权 while (!TAILQ_EMPTY(sched->ready)) { nty_coroutine *ready = TAILQ_POP(sched->ready); resume(ready); }} 补充知识点
1. X86-64寄存器和栈帧
1.1 寄存器
通用寄存器
X86_64 有 16 个 64 位寄存器，分别是：%rax，%rbx，%rcx，%rdx，%rsi，%rdi，%rbp，%rsp，%r8，%r9，%r10，%r11，%r12，%r13，%r14，%r15；
- %rax 作为函数返回值使用
- %rsp 栈指针寄存器，指向栈顶
- %rdi，%rsi，%rdx，%rcx，%r8，%r9 用作函数参数，依次对应第 1 参数，第 2 参数...
- %rbx，%rbp，%r12，%r13，%r14，%r15 用作数据存储，遵循被调用者使用规则，调用子函数之前要备份上述寄存器，以防被修改
- %r10，%r11 用作数据存储，遵循调用者使用规则，即使用之前要先保存原值
段寄存器
- cs 代码段寄存器
- ds, es, fs, gs 数据段寄存器
- ss 堆栈段寄存器
状态和控制寄存器 eflags
该寄存器表示的意义非常丰富，程序中并不直接操作此寄存器，并由此衍生出很多操作指令，除去一些保留位，其他每位都代表一个具体的含义，
其中 bits 0, 2, 4, 6, 7, 11 是状态位，标识了此操作后的状态
- CF (bit 0) —— 进位标识，算术操作进行了进位和借位，则此位被设置
- PF (bit 2) —— 奇偶标识，结果包含奇数个1，则设置此位
- AF (bit 4) —— 辅助进位标识，结果的第3位向第4位借位，则此位被设置
- ZF (bit 6) —— 零标识，结果为零，此位设置
- SF (bit 7) —— 符号标识，若为负数则设置此位
- OF (bit 11) —— 溢出标识，结果向最高位符号位进行借位或者进位，此标志被设置
8, 9, 10 位为控制标识
- TF (bit 8) —— 陷阱标识，设置进程可以被单步调试
- IF (bit 9) —— 中断标识，设置能够响应中断请求
- DF (bit 10) —— 方向标识，用于标示字符处理过程中指针移动方向
指令寄存器 EIP
EIP —— 标志当前进程将要执行指令位置，在64位模式下扩展为 RIP 64位指令寄存器
1.2 栈帧
1.2.1 栈帧结构
C 语言属于面向过程语言，其最大特点就是把一个程序分解成若干过程 (函数)，比如：入口函数是 main，然后调用各个子函数；在对应机器语言中，GCC 把过程转化成栈帧 (frame)，每个栈帧对应一个过程；X86-32 典型栈帧结构中，由 %ebp 指向栈帧开始，%esp 指向栈顶；
1.2.2 C 函数与汇编分析
int foo ( int x ){ int array[] = {1,3,5}; return array[x];} /* ----- end of function foo ----- */int main ( int argc, char *argv[] ){ int i = 1; int j = foo(i); fprintf(stdout, "i=%d,j=%d\n", i, j); return EXIT_SUCCESS;} /* ----- end of function main ----- */ 未经过优化的结果
优化后的结果
从优化可见，此处直接引用了栈顶之外的空间；
访问栈顶之外
通过 readelf 查看可执行程序的 header 信息
根据红色区域可见 x86-64 遵循 ABI 规则的版本，其定义了一些规范，遵循 ABI 的具体实现应该满足这些规范，其中，便规定了程序可以使用栈顶之外 128 字节的地址；
汇编相关分析
一、movl value, %eax
- movl value, %eax 这条指令是把 value 的值 (32位)，送入 %eax 寄存器；
- value 表达式格式 : base_address (offset_address, index, size)；计算公式 : base_address + offset_address + index * size；
二、call foo
- Pushl %rip // 保存下一条指令的地址，用于函数返回继续执行
- Jmp foo // 跳转到函数 foo
三、ret
- popl %rip // 恢复指令指针寄存器
四、cltq
- cltq 指令，特指 %eax->%rax 的符号拓展转换，等价于 movslq %eax, %rax
五、.LC0
.LC0 指一个字符串，此处代表 fprintf 的格式化字符串；
六、leave
leave 指令等价于
Movq %rbp %rsp // 撤销栈空间，回滚 %rspPopq %rbp // 恢复上一个栈帧的 %rbp
1.3 寄存器保存惯例
C 语言示例代码
#include#include void sfact_helper ( long int x, long int * resultp){ if (x <= 1) *resultp = 1; else { long int nresult; sfact_helper(x-1, &nresult); *resultp = x * nresult; }} /* ----- end of function foo ----- */long int sfact ( long int x ){ long int result; sfact_helper(x, &result); return result;} /* ----- end of function sfact ----- */intmain ( int argc, char *argv[] ){ int sum = sfact(10); fprintf(stdout, "sum=%d\n", sum); return EXIT_SUCCESS;} /* ---------- end of function main ---------- */
sfact_helper 函数生成的汇编代码
在函数 sfact_helper 中，用到了寄存器 %rbx 和 %rbp，在覆盖之前，GCC 选择了先保存他们的值，如代码 6~9 所示；在函数返回之前，GCC 依次恢复了他们，如代码 27-28 所示；
%rbx 在函数进入的时候指向的是 -16(%rsp)，而在退出的时候，指向的是32(%rsp)；此处使用了访问栈帧之外的空间的特性，即 GCC 不用预先分配空间再使用，而是先使用栈空间，然后在适当的时机分配，如第 11 行代码分配了空间，之后栈指针发生变化，所以同一个地址的引用偏移也相应做出调整；
注意事项
- 尽量使用 6 个以下的参数列表
- 传递大对象，尽量使用指针或者引用，鉴于寄存器只有 64 位，而且只能存储整形数值，寄存器存不下大对象
汇编相关分析
一、lea 指令
- 指令格式 : LEA 目的 源
- 指令功能 : 取源操作数地址的偏移量，并把它传送到目的操作数所在的单元
二、imul 指令
- (1) 双操作数的有符号乘指令
- 语句格式 : IMUL OPD, OPS
- 功能 : (OPD) * (OPS) ----> OPD
- 其中 OPD 可为 16/32 的寄存器，OPS 可为同类型的寄存器、存储器操作数或立即数
- (2) 3 个操作数的有符号乘指令
- 语句格式 : IMUL OPD, OPS, N
- 功能 : (OPS * N) -----> OPD
- 其中 OPD 可为 16/32 的寄存器，OPS 可为同类型的寄存器、存储器操作数，n 为立即数
- (3) 单操作数的有符号乘指令
- 语句格式 : IMUL OPS
- 功能 :
- 字节乘法 : (AL) * (OPS) ----> AX
- 字乘法 : (AX) * (OPS) ----> DX, AX
- 双字乘法 : (EAX) * (OPS) ----> EDX, EAX
三、mul 指令
- 语句格式 : MUL OPS
- 功能 :
- 字节乘法 : (AL) * (OPS) ----> AX
- 字乘法 : (AX) * (OPS) ----> DX, AX
- 双字乘法 : (EAX) * (OPS) ----> EDX, EAX
- 功能 :
1.4 参数传递范例
C 示例代码
#include#include int foo ( int arg1, int arg2, int arg3, int arg4, int arg5, int arg6, int arg7 ){ int array[] = {100, 200, 300, 400, 500, 600, 700}; int sum = array[ arg1 ] + array[ arg7 ]; return sum;} /* ----- end of function foo ----- */intmain ( int argc, char *argv[] ){ int i = 1; int j = foo(0, 1, 2, 3, 4, 5, 6); fprintf(stdout, "i = %d, j = %d\n", i, j); return EXIT_SUCCESS;} /* ---------- end of function main ---------- */
生成的汇编代码
Main 函数中，代码 31~37 准备函数 foo 的参数
- 参数 7 存储在栈上即 %rsp 指向的位置；
- 参数 6 存储在寄存器 %r9d；
- 参数 5 存储在寄存器 %r8d；
- 参数 4 对应于 %ecx；
- 参数 3 对应于 %edx；
- 参数 2 对应于 %esi；
- 参数 1 对应于 %edi；
Foo 函数中，代码 14-15，分别取出参数 7 和参数 1，参与运算；
此处数组引用，使用了最经典的寻址方式，-40(%rsp，%rdi，4) = %rsp + %rdi * 4 + (-40); 其中 %rsp 用作数组基地址，%rdi 用作了数组的下标，数字 4 表示 sizeof(int) = 4
1.5 结构体传参
C 示例代码
#include#include struct demo_s { char var8; int var32; long var64;};struct demo_s foo (struct demo_s d){ d.var8 = 8; d.var32 = 32; d.var64 = 64; return d;} /* ----- end of function foo ----- */intmain ( int argc, char *argv[] ){ struct demo_s d, result; result = foo (d); fprintf(stdout, "demo: %d, %d, %ld\n", result.var8, result.var32, result.var64); return EXIT_SUCCESS;} /* ---------- end of function main ---------- */
生成汇编代码
- 问题1 : 结构体如何传递；被分成了两个部分，var8 和 var32 合并成 8 个字节的大小，放在寄存器 %rdi 中，var64 放在寄存器的 %rsi 中，即结构体分解
- 问题2 : 结构体如何存储；如 foo 函数的第 15~17 行所示，结构体的引用变成了一个偏移量访问；类似于数组，只不过其元素大小可变；
- 问题3 : 结构体如何返回，原本 %rax 充当返回值的角色，现在添加第二个返回值角色 %rdx；同样，GCC 用两个寄存器来表示结构体；
- 即使在缺省情况下，GCC 依然想尽办法使用寄存器；随着结构变得越来越大，寄存器不够用了，则只能使用栈了；
参考与致谢
本博客为博主的学习实践总结，并参考了众多博主的博文，在此表示感谢，博主若有不足之处，请批评指正。【1】协程的实现与原理 [ M ]
【2】
【3】
【4】
【5】
【6】
【7】
【8】
转载地址：http://olmr.baihongyu.com/
你可能感兴趣的文章