在講協(xié)程之前，先解決上一篇文章《Linux高性能網(wǎng)絡編程十談|多進程和多線程》留下的思考題：

(1)如果在多線程程序中fork()子進程，會發(fā)生什么，我們要考慮那些問題?

• 首先我們會想到如果一個有多個線程的程序fork出來的子進程是否也是多個線程呢?不是，fork出來的子進程只有一個執(zhí)行線程，并不會把線程也復制過來;
• 其次fork出來的子進程都會繼承父進程的部分數(shù)據(jù)，包括鎖，句柄等，也就是說在父進程被鎖的臨界區(qū)，在子進程也會被鎖，這樣可能導致在子進程邏輯中繼續(xù)加鎖，導致出現(xiàn)死鎖情況;
• 最后使用pthread_atfork解決多線程下的fork問題，如下代碼注釋掉pthread_atfork這一行代碼，線程在父進程和子進程執(zhí)行process函數(shù)中重復加鎖，導致死鎖，如果使用pthread_atfork，則正常：

#include <pthread.h>
#include <stdio.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <time.h>
#include <unistd.h>

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

void *process(void *arg) {
  printf("pid = %d begin ...\n", static_cast<int>(getpid()));
  pthread_mutex_lock(&mutex);
  struct timespec ts = {2, 0};
  nanosleep(&ts, NULL);
  pthread_mutex_unlock(&mutex);
  printf("pid = %d end ...\n", static_cast<int>(getpid()));
  return NULL;
}

void prepare(void) { pthread_mutex_unlock(&mutex); }

void parent(void) { pthread_mutex_lock(&mutex); }

int main(void) {
  // pthread_atfork(prepare, parent, NULL);
  printf("pid = %d Entering main ...\n", static_cast<int>(getpid()));
  pthread_t tid;
  pthread_create(&tid, NULL, process, NULL);
  struct timespec ts = {1, 0};
  nanosleep(&ts, NULL);
  pid_t pid = fork();
  if (fork() == 0) {
    process(NULL);
  } else {
    waitpid(pid, NULL, 0);
  }
  pthread_join(tid, NULL);
  printf("pid = %d Exiting main ...\n", static_cast<int>(getpid()));
  return 0;
}

int pthread_atfork(void (*prepare)(void), void (*parent)(void), void (*child)(void))在fork()之前調用，當調用fork時，內部創(chuàng)建子進程前在父進程中會調用prepare，內部創(chuàng)建子進程成功后，父進程會調用parent，子進程會調用child;

(2)在多線程程序中，某個線程掛了，整個進程會掛么?

• 如果線程是非法訪問內存引起的崩潰，那么進程一定會崩潰，因為在進程中，各個線程的地址空間是共享的，某個線程破壞了某個地址段，其他線程也會受到到影響，這個時候操作系統(tǒng)與其保留其他線程，不如直接kill掉整個進程;
• 如果某個線程內的行為導致默認動作是停止或終止，則不管是否對其他線程是否有影響，整個進程都會停止或終止;
• 如果線程是因為自身退出(pthread_exit())或者各個線程捕獲信號可能不會掛掉整個進程，具體可以下面一個問題;

(3)如果需要將進程信號發(fā)送給某個線程，該如何處理?

• 首先線程可獨立地屏蔽某些信號，使用系統(tǒng)函數(shù)pthread_sigmask()，所以線程通常可以共享進程的信號，如果不需要則可以通過系統(tǒng)函數(shù)屏蔽;
• 其次可調用pthread_kill(pthread_t thread, int signo)，將信號發(fā)送給同一進程內指定的線程(包括自己);

第一部分：協(xié)程原理

如果您了解golang，協(xié)程應該不陌生，隨意用golang寫一個http server，性能都可能超過nginx，主要原因是內部使用輕量的協(xié)程，那下面我們就一起了解協(xié)程是什么?

協(xié)程就是 用戶態(tài)線程,協(xié)程的調度完全由開發(fā)者進行控制，因此實現(xiàn)協(xié)程的關鍵也就是 實現(xiàn)一個用戶態(tài)線程的調度器,由于協(xié)程是在用戶態(tài)中實現(xiàn)調度，避免了內核態(tài)的上下文切換造成的性能損失，從而突破了線程在IO上的性能瓶頸。

我們以ucontext庫為例子來說明協(xié)程是怎么運行的?(其他的協(xié)程實現(xiàn)方式類似)

#if defined(__APPLE__)
#define _XOPEN_SOURCE 600
#endif

#include <stdio.h>
#include <ucontext.h>

static ucontext_t ctx_main, ctx_coro;

void coroutine() {
  printf("Inside coroutine\n");
  swapcontext(&ctx_coro, &ctx_main); // 切換回主協(xié)程
  printf("Coroutine finished\n");
}

int main() {
  char coro_stack[8192];

  getcontext(&ctx_coro); // 獲取協(xié)程上下文
  ctx_coro.uc_stack.ss_sp = coro_stack;
  ctx_coro.uc_stack.ss_size = sizeof(coro_stack);
  ctx_coro.uc_link = &ctx_main; // 當協(xié)程結束時，切換回主協(xié)程
  makecontext(&ctx_coro, coroutine, 0); // 設置協(xié)程的入口點

  printf("Before coroutine\n");
  swapcontext(&ctx_main, &ctx_coro); // 切換到協(xié)程
  printf("Back in main\n");

  return 0;
}

以上代碼的輸出(mac上運行)：

Before coroutine
Inside coroutine
Back in main

以上代碼的流程是：

(1)通過getcontext保留當前棧的運行上下文到ucontext_t中;

(2)通過makecontext修改ucontext_t指向coroutine入口函數(shù);

(3)通過swapcontext切換協(xié)程;

看了如上代碼，如果之前對協(xié)程沒有了解的，還是比較懵，為什么getcontext能保留運行的上下文呢?我們先看一下內存中數(shù)據(jù)塊分布：

堆棧圖

一個函數(shù)執(zhí)行會經(jīng)過如下步驟：

(1)把參數(shù)加入棧中，如果有其他參數(shù)沒有入棧，那么使用某些寄存器傳遞;

(2)把當前指令的下一條指令地址壓入棧中;

(3)跳轉到函數(shù)體執(zhí)行:

(4)把EBP壓入棧中，指向上一個函數(shù)堆棧幀中的幀指針的位置;

(5)保存調用前后需要保存不變的寄存器的值;

(6)將局部變量壓入棧中;

從上面代碼看出，當函數(shù)執(zhí)行完需要恢復到上一次執(zhí)行入口的寄存器地址，那getcontext只需要把當前恢復入口的地址存起來和加上必要棧信息是否就能實現(xiàn)保留協(xié)程棧，getcontext的確是這么做的：

movq        (%rsp), %rcx
movq        %rcx, oRIP(%rdi)
leaq        8(%rsp), %rcx         /* Exclude the return address.  */
movq        %rcx, oRSP(%rdi)

(%rsp)中保存的即是函數(shù)返回地址，也就是執(zhí)行完getcontext這個函數(shù)之后需要執(zhí)行的下一個指令的地址，通過context保存相關寄存器的值主要是rip值，同時把當前棧的rsp值也保存，這樣便可以通過這些數(shù)據(jù)恢復context以再次繼續(xù)執(zhí)行。

同樣我們調用swapcontext取出context信息，通過恢復下一個需要執(zhí)行的函數(shù)入口實現(xiàn)協(xié)程切換：

movq        (%rsp), %rcx
movq        %rcx, oRIP(%rdi)
leaq        8(%rsp), %rcx       /* Exclude the return address.  */
movq        %rcx, oRSP(%rdi)

當前rsp指向的地址中存儲的是返回地址，即調用swapcontext后當前協(xié)程需要執(zhí)行的下一個指令地址，同時將swapcontext第二個參數(shù)的棧恢復，就進入下一個協(xié)程的入口函數(shù)。

第二部分：協(xié)程類型

目前開源有很多協(xié)程，根據(jù)運行時協(xié)程棧的分配方式分為有棧協(xié)程和無棧協(xié)程，根據(jù)調度過程中調度權的目標分為對稱協(xié)程和非對稱協(xié)程，下面我們來簡單了解一下：

1、有棧協(xié)程和無棧協(xié)程

(1)如果每個協(xié)程都有自己的調用棧，類似于線程的調用棧就是有棧協(xié)程，微信的libco、Golang中的 goroutine、Lua中的協(xié)程都是有棧協(xié)程。

實現(xiàn)方式上面應該已經(jīng)了解了，在內存中給每個協(xié)程開辟一個棧內存，當協(xié)程掛起時會將它的運行時上下文(即?？臻g)從系統(tǒng)棧中保存至其所分配的棧內存中，當協(xié)程恢復時會將其運行時上下文從棧內存中恢復至系統(tǒng)棧中;

采用有棧協(xié)程有優(yōu)點也有缺點，優(yōu)點是可以任意嵌套，只要保留了當前棧的信息，可以任意的切換到其他協(xié)程中，而缺點則是性能有一定的損失，在保留棧空間信息的拷入拷出都會影響性能，同時棧的擴大和縮小需要實現(xiàn)動態(tài)，這里會導致內存浪費;

(2)與有棧協(xié)程相反，無棧協(xié)程不會為各個協(xié)程開辟相應的調用棧。無棧協(xié)程通常是基于狀態(tài)機或閉包來實現(xiàn)，類似ES6、Dart中的await/async、Python的Generator、Kotlin中的協(xié)程、C++20中的cooroutine都是無棧協(xié)程;

使用無棧協(xié)程不需要修改調用棧，也無需額外的內存來保存調用棧，因此它的開銷會更小，同時無需要考慮棧需要動態(tài)擴大縮小的問題，但是相比于保存運行時上下文這種實現(xiàn)方式，無棧協(xié)程最大的問題它無法實現(xiàn)在任意函數(shù)調用層級的位置進行掛起，比如最簡單的無棧協(xié)程設計如下：

#include <stdio.h>

int function(void) {
  static int i, state = 0;
  switch (state) {
  case 0: /* start of function */
    for (i = 0; i < 10; i++) {
      state = 1; /* so we will come back to "case 1" */
      return i;
    case 1:; /* resume control straight after the return */
    }
  }
}

int main() {
  for (int i = 0; i < 10; i++) {
    fprintf(stdout, "%d\n", function());
  }
  return 0;
}

以上代碼通過label和goto實現(xiàn)了yield語義，從而實現(xiàn)調用function()獲得打印0~9，如果大家想詳細了解這里面的實現(xiàn)可以搜索Protothreads庫;

(3)有棧協(xié)程和無棧協(xié)程總結如下：

內存資源使用：無棧協(xié)程借助函數(shù)的棧幀來存儲一些寄存器狀態(tài)，可以調用遞歸函數(shù)，而有棧協(xié)程會要申請一個內存棧用來存儲寄存器信息，調用遞歸函數(shù)可能會爆棧;

速度：無棧協(xié)程的上下文比較少，所以能夠進行更快的用戶態(tài)上下文切換;

功能性：有棧協(xié)程能夠在嵌套的協(xié)程中進行掛起/恢復，而無棧協(xié)程只能對頂層的協(xié)程進行掛起，被調用方是不能掛起的;

2、對稱協(xié)程和非對稱協(xié)程

(1)對稱協(xié)程：任何一個協(xié)程都是相互獨立且平等的，調度權可以在任意協(xié)程之間轉移，例如go語言的協(xié)程就是對稱線程，其實現(xiàn)如下圖所示：

調度圖

CoroutineA，CoroutineB，CoroutineC之間是可以通過協(xié)程調度器可以切換到任意協(xié)程。

(2)非對稱協(xié)程：協(xié)程出讓調度權的目標只能是它的調用者，即協(xié)程之間存在調用和被調用關系，例如libco提供的協(xié)議就是非對稱協(xié)程，其實現(xiàn)如下圖所示：

調度圖

CoroutineA，CoroutineB，CoroutineC之間比如與調用者成對出現(xiàn)，比如resume的調用者返回的位置，必須是被調用者yield。

第三部分：如何使用協(xié)程實現(xiàn)高性能

以下是網(wǎng)絡IO與協(xié)程調度流程：

調度圖

(1)epoll，kqueue等IO事件觸發(fā);

(2)調度協(xié)程循環(huán)等待，如果遇到IO事件，就創(chuàng)建協(xié)程開始處理;

(3)創(chuàng)建IO協(xié)程或者定時器協(xié)程;

(4)如果是定時器協(xié)程，就加入到定時協(xié)程隊列;

(5)如果是IO協(xié)程，就加入到IO協(xié)程隊列(每一個網(wǎng)絡連接綁定一個套接字句柄，該套接字綁定一個協(xié)程);

(6)觸發(fā)的IO喚醒調度器，調度器準備協(xié)程切換;

(7)從IO協(xié)程隊列中取出對應的協(xié)程進行處理;

(8)如果當前協(xié)程遇到IO阻塞，比如處理完recv數(shù)據(jù)，需要send數(shù)據(jù)或者往下游send數(shù)據(jù)，都是IO阻塞場景;

(9)當前協(xié)程阻塞后將自己掛起;

(10)切換到調度協(xié)程或者其他協(xié)程繼續(xù)調度(如果是對稱協(xié)程直接切到調度協(xié)程，如果是非對程協(xié)程調用yield);

(11)遇到IO關閉將當前協(xié)程切換到退出狀態(tài)(可以設置退出狀態(tài));

(12)IO協(xié)程直接退出;

(13)9~12步驟中的IO觸發(fā)或者IO關閉以后，切換到下一個協(xié)程;

(14)如果調度協(xié)程執(zhí)行完，然后查詢定時協(xié)程隊列，如果有超時的處理TODO;(15)執(zhí)行完上述流程，繼續(xù)切換回調度協(xié)程，等待IO事件的觸發(fā);

以上流程的偽代碼如下(詳細的代碼后續(xù)會在https://github.com/linkxzhou/sthread這里開源，目前在完善中)：

void process(void *args) {
  ...
  /* co_read封裝監(jiān)聽io事件協(xié)程切換`yield` */
  ... = co_read(...)
  ...
  /* co_send封裝監(jiān)聽io事件協(xié)程切換`yield` */
  ... = co_send(...)
  ...
}

void co_eventloop() {
  ...
  for (;;) {
    /* 調度協(xié)程通過 epoll_wait撈出就緒事件 */
    int ret = co_epoll_wait(...);
    while (...) {
      /* 如果不存在對應句柄的協(xié)程則創(chuàng)建協(xié)程，具體process函數(shù)處理 */
      ...* co = get_co_by_fd(...);
      if (co == NULL) {
        co = co_create(...)
      }
      ...
      /* 主協(xié)程掛起當前協(xié)程，切換到對應子協(xié)程，處理到期事件和就緒事件結果 */
      co_resume(co)
    }
    ...
    /* 調度協(xié)程處理過期事件，主協(xié)程切換到定時處理協(xié)程 */
    process_alltimeout_list(...);
    ...
  }
}

如何實現(xiàn)高性能呢?

(1)首先通過IO復用結合協(xié)程，每個連接綁定一個協(xié)程，由于協(xié)程比較輕量，假設對于有棧協(xié)程占用空間8K左右，100w個連接也就是8G左右，所以對于內存開銷不大;

(2)其次協(xié)程調度是微秒或者納秒級，如果對于IO密集型應用，基本上上就是一個協(xié)程處理完以后，微秒或者納秒級內就能切換到下一個處理連接;

(3)最后對比多線程，協(xié)程減少了臨界區(qū)的處理，不需要互斥鎖，信號量等開銷較大的同步原語，所以可以更能輕松寫出高性能的server;

思考

繼續(xù)提幾個思考題(下一章會解答當前問題)：

(1)多線程情況下如何處理協(xié)程?

(2)golang的協(xié)程調度方式是怎樣的?