Linux 高性能網絡編程協程

在講協程之前，先解決上一篇文章《Linux 高性能網絡編程十談 | 多進程和多線程》留下的思考題：
（1）如果在多線程程序中fork()子進程，會發生什麼，我們要考慮那些問題？

首先我們會想到如果一個有多個線程的程序fork出來的子進程是否也是多個線程呢？不是，fork 出來的子進程只有一個執行線程，並不會把線程也複製過來；
其次fork出來的子進程都會繼承父進程的部分數據，包括鎖，句柄等，也就是說在父進程被鎖的臨界區，在子進程也會被鎖，這樣可能導致在子進程邏輯中繼續加鎖，導致出現死鎖情況；
最後使用pthread_atfork解決多線程下的fork問題，如下代碼註釋掉pthread_atfork這一行代碼，線程在父進程和子進程執行process函數中重複加鎖，導致死鎖，如果使用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時，內部創建子進程前在父進程中會調用prepare，內部創建子進程成功後，父進程會調用parent，子進程會調用child；

（2）在多線程程序中，某個線程掛了，整個進程會掛麼？

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

（3）如果需要將進程信號發送給某個線程，該如何處理？

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

第一部分：協程原理

如果您瞭解 golang，協程應該不陌生，隨意用 golang 寫一個http server，性能都可能超過 nginx，主要原因是內部使用輕量的協程，那下面我們就一起了解協程是什麼？
協程就是用戶態線程, 協程的調度完全由開發者進行控制，因此實現協程的關鍵也就是實現一個用戶態線程的調度器, 由於協程是在用戶態中實現調度，避免了內核態的上下文切換造成的性能損失，從而突破了線程在 IO 上的性能瓶頸。
我們以ucontext庫爲例子來說明協程是怎麼運行的？（其他的協程實現方式類似）

#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); // 切換回主協程
  printf("Coroutine finished\n");
}

int main() {
  char coro_stack[8192];

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

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

  return 0;
}

以上代碼的輸出（mac 上運行）：

Before coroutine
Inside coroutine
Back in main

以上代碼的流程是：
（1）通過getcontext保留當前棧的運行上下文到ucontext_t中；
（2）通過makecontext修改ucontext_t指向coroutine入口函數；
（3）通過swapcontext切換協程；

看了如上代碼，如果之前對協程沒有了解的，還是比較懵，爲什麼getcontext能保留運行的上下文呢？我們先看一下內存中數據塊分佈：

堆棧圖

一個函數執行會經過如下步驟：
（1）把參數加入棧中，如果有其他參數沒有入棧，那麼使用某些寄存器傳遞；
（2）把當前指令的下一條指令地址壓入棧中；
（3）跳轉到函數體執行:
（4）把 EBP 壓入棧中，指向上一個函數堆棧幀中的幀指針的位置；
（5）保存調用前後需要保存不變的寄存器的值；
（6）將局部變量壓入棧中；

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

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

(%rsp) 中保存的即是函數返回地址，也就是執行完getcontext這個函數之後需要執行的下一個指令的地址，通過context保存相關寄存器的值主要是 rip 值，同時把當前棧的 rsp 值也保存，這樣便可以通過這些數據恢復context以再次繼續執行。
同樣我們調用swapcontext取出context信息，通過恢復下一個需要執行的函數入口實現協程切換：

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

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

第二部分：協程類型

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

1、有棧協程和無棧協程

（1）如果每個協程都有自己的調用棧，類似於線程的調用棧就是有棧協程，微信的 libco、Golang 中的 goroutine、Lua 中的協程都是有棧協程。
實現方式上面應該已經瞭解了，在內存中給每個協程開闢一個棧內存，當協程掛起時會將它的運行時上下文（即棧空間）從系統棧中保存至其所分配的棧內存中，當協程恢復時會將其運行時上下文從棧內存中恢復至系統棧中；

採用有棧協程有優點也有缺點，優點是可以任意嵌套，只要保留了當前棧的信息，可以任意的切換到其他協程中，而缺點則是性能有一定的損失，在保留棧空間信息的拷入拷出都會影響性能，同時棧的擴大和縮小需要實現動態，這裏會導致內存浪費；
（2）與有棧協程相反，無棧協程不會爲各個協程開闢相應的調用棧。無棧協程通常是基於狀態機或閉包來實現，類似ES6、Dart中的await/async、Python的Generator、Kotlin中的協程、C++20中的cooroutine都是無棧協程；

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

#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實現了yield語義，從而實現調用function()獲得打印0~9，如果大家想詳細瞭解這裏面的實現可以搜索Protothreads庫；

（3）有棧協程和無棧協程總結如下：

內存資源使用：無棧協程藉助函數的棧幀來存儲一些寄存器狀態，可以調用遞歸函數，而有棧協程會要申請一個內存棧用來存儲寄存器信息，調用遞歸函數可能會爆棧；
速度：無棧協程的上下文比較少，所以能夠進行更快的用戶態上下文切換；
功能性：有棧協程能夠在嵌套的協程中進行掛起 / 恢復，而無棧協程只能對頂層的協程進行掛起，被調用方是不能掛起的；

2、對稱協程和非對稱協程

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

調度圖

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

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

調度圖

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

第三部分：如何使用協程實現高性能

以下是網絡 IO 與協程調度流程：

調度圖

（1）epoll，kqueue 等 IO 事件觸發；
（2）調度協程循環等待，如果遇到 IO 事件，就創建協程開始處理；
（3）創建 IO 協程或者定時器協程；
（4）如果是定時器協程，就加入到定時協程隊列；
（5）如果是 IO 協程，就加入到 IO 協程隊列（每一個網絡連接綁定一個套接字句柄，該套接字綁定一個協程）；
（6）觸發的 IO 喚醒調度器，調度器準備協程切換；
（7）從 IO 協程隊列中取出對應的協程進行處理；
（8）如果當前協程遇到 IO 阻塞，比如處理完recv數據，需要send數據或者往下游send數據，都是 IO 阻塞場景；
（9）當前協程阻塞後將自己掛起；
（10）切換到調度協程或者其他協程繼續調度（如果是對稱協程直接切到調度協程，如果是非對程協程調用yield）；
（11）遇到 IO 關閉將當前協程切換到退出狀態（可以設置退出狀態）；
（12）IO 協程直接退出；
（13）9~12 步驟中的 IO 觸發或者 IO 關閉以後，切換到下一個協程；
（14）如果調度協程執行完，然後查詢定時協程隊列，如果有超時的處理 TODO；（15）執行完上述流程，繼續切換回調度協程，等待 IO 事件的觸發；

以上流程的僞代碼如下（詳細的代碼後續會在 https://github.com/linkxzhou/sthread 這裏開源，目前在完善中）：

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

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

如何實現高性能呢？
（1）首先通過 IO 複用結合協程，每個連接綁定一個協程，由於協程比較輕量，假設對於有棧協程佔用空間 8K 左右，100w 個連接也就是 8G 左右，所以對於內存開銷不大；
（2）其次協程調度是微秒或者納秒級，如果對於 IO 密集型應用，基本上上就是一個協程處理完以後，微秒或者納秒級內就能切換到下一個處理連接；
（3）最後對比多線程，協程減少了臨界區的處理，不需要互斥鎖，信號量等開銷較大的同步原語，所以可以更能輕鬆寫出高性能的server；

思考

繼續提幾個思考題（下一章會解答當前問題）：
（1）多線程情況下如何處理協程？
（2）golang 的協程調度方式是怎樣的？

參考

（1）https://cloud.tencent.com/developer/article/1888257
（2）https://www.zhihu.com/question/52193579/answer/156692295
（3）https://blog.csdn.net/happyAnger6/article/details/118466471
（4）https://github.com/Tencent/flare/blob/master/flare/doc/fiber.md
（5）https://www.cnblogs.com/pokpok/p/16932735.html
（6）https://www.infoq.cn/article/BOu7GVc9kP7CWkYP58jo?source=app_share
（7）https://coolshell.cn/articles/10975.html
（8）https://zhuanlan.zhihu.com/p/363775637
（9）https://zhuanlan.zhihu.com/p/52061886
（10）https://time.geekbang.org/column/article/435493

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/lNX8h2e6vfUuOnlf3EhDAg

第一部分：協程原理

第二部分：協程類型

1、有棧協程和無棧協程

2、對稱協程和非對稱協程

第三部分：如何使用協程實現高性能

思考

參考

猜你喜歡