協程的實現之原語操作
問題：協程的內部原語操作有哪些？分別如何實現的？
協程的核心原語操作：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) {
assert(pthread_once(&sched_key_once, nty_coroutine_sched_key_creator) == 0);
nty_schedule *sched = nty_coroutine_get_sched();
if (sched == NULL) {
nty_schedule_create(0);
sched = nty_coroutine_get_sched();
if (sched == NULL) {
printf("Failed to create schedulern");
return -1;
}
}
nty_coroutine *co = calloc(1, sizeof(nty_coroutine));
if (co == NULL) {
printf("Failed to allocate memory for new coroutinen");
return -2;
}
//
int ret = posix_memalign(&co->stack, getpagesize(), sched->stack_size);
if (ret) {
printf("Failed to allocate stack for new coroutinen");
free(co);
return -3;
}
co->sched = sched;
co->stack_size = sched->stack_size;
co->status = BIT(NTY_COROUTINE_STATUS_NEW); //
co->id = sched->spawned_coroutines ++;
co->func = func;
co->fd = -1;
co->events = 0;
co->arg = arg;
co->birth = nty_coroutine_usec_now();
*new_co = co;
TAILQ_INSERT_TAIL(&co->sched->ready, co, ready_next);
return 0;
}
yield： 讓出CPU。
void nty_coroutine_yield(nty_coroutine *co)
參數：當前運行的協程實例
調用后該函數不會立即返回，而是切換到最近執行resume的上下文。該函數返回是在執行resume的時候，會有調度器統一選擇resume的，然后再次調用yield的。resume與yield是兩個可逆過程的原子操作。
resume：恢復協程的運行權
int nty_coroutine_resume(nty_coroutine *co)
參數：需要恢復運行的協程實例
調用后該函數也不會立即返回，而是切換到運行協程實例的yield的位置。返回是在等協程相應事務處理完成后，主動yield會返回到resume的地方。
協程的實現之切換
問題：協程的上下文如何切換？切換代碼如何實現？
首先來回顧一下x86_64寄存器的相關知識。x86_64 的寄存器有16個64位寄存器，分別是：%rax, %rbx, %rcx, %esi, %edi, %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 用作數據存儲，就是使用前要先保存原值。
上下文切換，就是將CPU的寄存器暫時保存，再將即將運行的協程的上下文寄存器，分別mov到相對應的寄存器上。此時上下文完成切換。如下圖所示：
切換_switch函數定義：
int _switch(nty_cpu_ctx *new_ctx, nty_cpu_ctx *cur_ctx);
參數1：即將運行協程的上下文，寄存器列表
參數2：正在運行協程的上下文，寄存器列表
我們nty_cpu_ctx結構體的定義，為了兼容x86，結構體項命令采用的是x86的寄存器名字命名。
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返回后，執行即將運行協程的上下文。是實現上下文的切換
_switch的實現代碼：
0: __asm__ (
1: " .text n"
2: " .p2align 4,,15 n"
3: ".globl _switch n"
4: ".globl __switch n"
5: "_switch: n"
6: "__switch: n"
7: " movq %rsp, 0(%rsi) # save stack_pointer n"
8: " movq %rbp, 8(%rsi) # save frame_pointer n"
9: " movq (%rsp), %rax # save insn_pointer n"
10: " movq %rax, 16(%rsi) n"
11: " movq %rbx, 24(%rsi) # save rbx,r12-r15 n"
12: " movq %r12, 32(%rsi) n"
13: " movq %r13, 40(%rsi) n"
14: " movq %r14, 48(%rsi) n"
15: " movq %r15, 56(%rsi) n"
16: " movq 56(%rdi), %r15 n"
17: " movq 48(%rdi), %r14 n"
18: " movq 40(%rdi), %r13 # restore rbx,r12-r15 n"
19: " movq 32(%rdi), %r12 n"
20: " movq 24(%rdi), %rbx n"
21: " movq 8(%rdi), %rbp # restore frame_pointer n"
22: " movq 0(%rdi), %rsp # restore stack_pointer n"
23: " movq 16(%rdi), %rax # restore insn_pointer n"
24: " movq %rax, (%rsp) n"
25: " ret n"
26: );
按照x86_64的寄存器定義，%rdi保存第一個參數的值，即new_ctx的值，%rsi保存第二個參數的值，即保存cur_ctx的值。X86_64每個寄存器是64bit，8byte。
Movq %rsp, 0(%rsi) 保存在棧指針到cur_ctx實例的rsp項
Movq %rbp, 8(%rsi)
Movq (%rsp), %rax #將棧頂地址里面的值存儲到rax寄存器中。Ret后出棧，執行棧頂
Movq %rbp, 8(%rsi) #后續的指令都是用來保存CPU的寄存器到new_ctx的每一項中
Movq 8(%rdi), %rbp #將new_ctx的值
Movq 16(%rdi), %rax #將指令指針rip的值存儲到rax中
Movq %rax, (%rsp) # 將存儲的rip值的rax寄存器賦值給棧指針的地址的值。
Ret # 出棧，回到棧指針，執行rip指向的指令。
上下文環境的切換完成
協程的實現之定義
問題：協程如何定義? 調度器如何定義？
先來一道設計題：
設計一個協程的運行體R與運行體調度器S的結構體
1. 運行體R：包含運行狀態{就緒，睡眠，等待}，運行體回調函數，回調參數，棧指針，棧大小，當前運行體
2. 調度器S：包含執行集合{就緒，睡眠，等待}
這道設計題拆分兩個個問題，一個運行體如何高效地在多種狀態集合更換。調度器與運行體的功能界限。
運行體如何高效地在多種狀態集合更換
新創建的協程，創建完成后，加入到就緒集合，等待調度器的調度；協程在運行完成后，進行IO操作，此時IO并未準備好，進入等待狀態集合；IO準備就緒，協程開始運行，后續進行sleep操作，此時進入到睡眠狀態集合。
就緒(ready)，睡眠(sleep)，等待(wait)集合該采用如何數據結構來存儲？
就緒(ready)集合并不沒有設置優先級的選型，所有在協程優先級一致，所以可以使用隊列來存儲就緒的協程，簡稱為就緒隊列（ready_queue）。
睡眠(sleep)集合需要按照睡眠時長進行排序，采用紅黑樹來存儲，簡稱睡眠樹(sleep_tree)紅黑樹在工程實用為<key, value>, key為睡眠時長，value為對應的協程結點。
等待(wait)集合，其功能是在等待IO準備就緒，等待IO也是有時長的，所以等待(wait)集合采用紅黑樹的來存儲，簡稱等待樹(wait_tree)，此處借鑒Nginx的設計。
數據結構如下圖所示：
Coroutine就是協程的相應屬性，status表示協程的運行狀態。sleep與wait兩顆紅黑樹，ready使用的隊列，比如某協程調用sleep函數，加入睡眠樹(sleep_tree)，status |= S即可。比如某協程在等待樹(wait_tree)中，而IO準備就緒放入ready隊列中，只需要移出等待樹(wait_tree)，狀態更改status &= ~W即可。有一個前提條件就是不管何種運行狀態的協程，都在就緒隊列中，只是同時包含有其他的運行狀態。
調度器與協程的功能界限
每一協程都需要使用的而且可能會不同屬性的，就是協程屬性。每一協程都需要的而且數據一致的，就是調度器的屬性。比如棧大小的數值，每個協程都一樣的后不做更改可以作為調度器的屬性，如果每個協程大小不一致，則可以作為協程的屬性。
用來管理所有協程的屬性，作為調度器的屬性。比如epoll用來管理每一個協程對應的IO，是需要作為調度器屬性。
按照前面幾章的描述，定義一個協程結構體需要多少域，我們描述了每一個協程有自己的上下文環境，需要保存CPU的寄存器ctx；需要有子過程的回調函數func；需要有子過程回調函數的參數 arg；需要定義自己的棧空間 stack；需要有自己棧空間的大小 stack_size；需要定義協程的創建時間 birth；需要定義協程當前的運行狀態 status；需要定當前運行狀態的結點（ready_next, wait_node, sleep_node）；需要定義協程id；需要定義調度器的全局對象 sched。
協程的核心結構體如下：
typedef struct _nty_coroutine {
nty_cpu_ctx ctx;
proc_coroutine func;
void *arg;
size_t stack_size;
nty_coroutine_status status;
nty_schedule *sched;
uint64_t birth;
uint64_t id;
void *stack;
RB_ENTRY(_nty_coroutine) sleep_node;
RB_ENTRY(_nty_coroutine) wait_node;
TAILQ_ENTRY(_nty_coroutine) ready_next;
TAILQ_ENTRY(_nty_coroutine) defer_next;
} nty_coroutine;
調度器是管理所有協程運行的組件，協程與調度器的運行關系。
調度器的屬性，需要在保存CPU的寄存器上下文 ctx，可以從協程運行狀態yield到調度器運行的。從協程到調度器用yield，從調度器到協程用resume
以下為協程的定義。
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;
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;
nty_coroutine_queue ready;
nty_coroutine_rbtree_sleep sleeping;
nty_coroutine_rbtree_wait waiting;
} nty_schedule;
協程的實現之調度器
問題：協程如何被調度？
調度器的實現，有兩種方案，一種是生產者消費者模式，另一種多狀態運行。
生產者消費者模式
邏輯代碼如下：
while (1) {
//遍歷睡眠集合，將滿足條件的加入到ready
nty_coroutine *expired = NULL;
while ((expired = sleep_tree_expired(sched)) != ) {
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);
}
}
多狀態運行
實現邏輯代碼如下：
while (1) {
//遍歷睡眠集合，使用resume恢復expired的協程運行權
nty_coroutine *expired = NULL;
while ((expired = sleep_tree_expired(sched)) != ) {
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);
}
}
