協程庫Libtask源碼分析之架構篇
本文轉載自微信公眾號「編程雜技」,作者theanarkh 。轉載本文請聯系編程雜技公眾號。
前言:假設讀者已經了解了協程的概念,實現協程的底層技術支持。本文會介紹基于底層基礎,如何實現協程以及協程的應用(更多基礎可以點擊這里[1])。
libtask是google大佬Russ Cox(Go的核心開發者)所寫,本文介紹libtask的基礎原理。我們從libtask的main函數開始,這個main函數就是我們在c語言中使用的c函數,libtask本身實現了main這個函數,用戶使用libtask時,要實現的是taskmain函數。taskmain和main的函數聲明是一樣的。下面我們看一下main函數。
- int main(int argc, char **argv)
- {
- struct sigaction sa, osa;
- // 注冊SIGQUIT信號處理函數
- memset(&sa, 0, sizeof sa);
- sa.sa_handler = taskinfo;
- sa.sa_flags = SA_RESTART;
- sigaction(SIGQUIT, &sa, &osa);
- // 保存命令行參數
- argv0 = argv[0];
- taskargc = argc;
- taskargv = argv;
- if(mainstacksize == 0)
- mainstacksize = 256*1024;
- // 創建第一個協程
- taskcreate(taskmainstart, nil, mainstacksize);
- // 開始調度
- taskscheduler();
- fprint(2, "taskscheduler returned in main!\n");
- abort();
- return 0;
- }
main函數主要的兩個邏輯是taskcreate和taskscheduler函數。我們先來看taskcreate。
- int taskcreate(void (*fn)(void*), void *arg, uint stack)
- {
- int id;
- Task *t;
- t = taskalloc(fn, arg, stack);
- taskcount++;
- id = t->id;
- // 記錄位置
- t->alltaskslot = nalltask;
- // 保存到alltask中
- alltask[nalltask++] = t;
- // 修改狀態為就緒,可以被調度,并且加入到就緒隊列
- taskready(t);
- return id;
- }
taskcreate首先調用taskalloc分配一個表示協程的結構體Task。我們看看這個結構體的定義。
- struct Task
- {
- char name[256]; // offset known to acid
- char state[256];
- // 前后指針
- Task *next;
- Task *prev;
- Task *allnext;
- Task *allprev;
- // 執行上下文
- Context context;
- // 睡眠時間
- uvlong alarmtime;
- uint id;
- // 棧信息
- uchar *stk;
- uint stksize;
- //是否退出了
- int exiting;
- // 在alltask的索引
- int alltaskslot;
- // 是否是系統協程
- int system;
- // 是否就緒狀態
- int ready;
- // 入口函數
- void (*startfn)(void*);
- // 入口參數
- void *startarg;
- // 自定義數據
- void *udata;
- };
接著看看taskalloc的實現。
- // 分配一個協程所需要的內存和初始化某些字段
- static Task*
- taskalloc(void (*fn)(void*), void *arg, uint stack)
- {
- Task *t;
- sigset_t zero;
- uint x, y;
- ulong z;
- /* allocate the task and stack together */
- // 結構體本身的大小+棧大小
- t = malloc(sizeof *t+stack);
- memset(t, 0, sizeof *t);
- // 棧的內存位置
- t->stk = (uchar*)(t+1);
- // 棧大小
- t->stksize = stack;
- // 協程id
- t->id = ++taskidgen;
- // 協程工作函數和參數
- t->startfn = fn;
- t->startarg = arg;
- /* do a reasonable initialization */
- memset(&t->context.uc, 0, sizeof t->context.uc);
- sigemptyset(&zero);
- // 初始化uc_sigmask字段為空,即不阻塞信號
- sigprocmask(SIG_BLOCK, &zero, &t->context.uc.uc_sigmask);
- /* must initialize with current context */
- // 初始化uc字段
- getcontext(&t->context.uc)
- // 設置協程執行時的棧位置和大小
- t->context.uc.uc_stack.ss_sp = t->stk+8;
- t->context.uc.uc_stack.ss_size = t->stksize-64;
- z = (ulong)t;
- y = z;
- z >>= 16; /* hide undefined 32-bit shift from 32-bit compilers */
- x = z>>16;
- // 保存信息到uc字段
- makecontext(&t->context.uc, (void(*)())taskstart, 2, y, x);
- return t;
- }
taskalloc函數代碼看起來很多,但是邏輯不算復雜,就是申請Task結構體所需的內存和執行時棧的內存,然后初始化各個字段。這樣,一個協程就誕生了。接著執行taskready把協程加入就緒隊列。
- // 修改協程的狀態為就緒并加入就緒隊列
- void taskready(Task *t)
- {
- t->ready = 1;
- addtask(&taskrunqueue, t);
- }
- // 把協程插入隊列中,如果之前在其他隊列,則會被移除
- void addtask(Tasklist *l, Task *t)
- {
- if(l->tail){
- l->tail->next = t;
- t->prev = l->tail;
- }else{
- l->head = t;
- t->prev = nil;
- }
- l->tail = t;
- t->next = nil;
- }
taskrunqueue記錄了所有就緒的協程。創建了協程并加入隊列后,協程還沒有開始執行,就像操作系統的進程和線程一樣,需要有一個調度器來調度執行。下面我們看看調度器的實現。
- // 協程調度中心
- static void taskscheduler(void)
- {
- int i;
- Task *t;
- for(;;){
- // 沒有用戶協程了,則退出
- if(taskcount == 0)
- exit(taskexitval);
- // 從就緒隊列拿出一個協程
- t = taskrunqueue.head;
- if(t == nil){
- fprint(2, "no runnable tasks! %d tasks stalled\n", taskcount);
- exit(1);
- }
- // 從就緒隊列刪除該協程
- deltask(&taskrunqueue, t);
- t->ready = 0;
- // 保存正在執行的協程
- taskrunning = t;
- // 切換次數加一
- tasknswitch++;
- // 切換到t執行,并且保存當前上下文到taskschedcontext(即下面要執行的代碼)
- contextswitch(&taskschedcontext, &t->context);
- // 執行到這說明沒有協程在執行(t切換回來的),置空
- taskrunning = nil;
- // 剛才執行的協程t退出了
- if(t->exiting){
- // 不是系統協程,則個數減一
- if(!t->system)
- taskcount--;
- // 當前協程在alltask的索引
- i = t->alltaskslot;
- // 把最后一個協程換到當前協程的位置,因為他要退出了
- alltask[i] = alltask[--nalltask];
- // 更新被置換協程的索引
- alltask[i]->alltaskslot = i;
- // 釋放堆內存
- free(t);
- }
- }
- }
調度器的代碼看起來很多,但是核心邏輯就三個 1 從就緒隊列中拿出一個協程t,并把t移出就緒隊列 2 通過contextswitch切換到協程t中執行 3 協程t切換回調度中心,如果t已經退出,則修改數據結構,然后回收他占據的內存。如果t沒退出,則繼續調度其他協程執行。至此,協程就開始跑起來了。并且也有了調度系統。這里的調度機制是比較簡單的,就是按著先進先出的方式就緒調度,并且是非搶占的。即沒有按時間片調度的概念,一個協程的執行時間由自己決定,放棄執行的權力也是自己控制的,當協程不想執行了可以調用taskyield讓出cpu。
- // 協程主動讓出cpu
- int taskyield(void)
- {
- int n;
- // 當前切換協程的次數
- n = tasknswitch;
- // 插入就緒隊列,等待后續的調度
- taskready(taskrunning);
- taskstate("yield");
- // 切換協程
- taskswitch();
- // 等于0說明當前只有自己一個協程,調度的時候tasknswitch加一,所以這里減一
- return tasknswitch - n - 1;
- }
- /*
- 切換協程,taskrunning是正在執行的協程,taskschedcontext是調度協程(主線程)的上下文,
- 切換到調度中心,并保持當前上下文到taskrunning->context
- */
- void taskswitch(void)
- {
- needstack(0);
- contextswitch(&taskrunning->context, &taskschedcontext);
- }
- // 真正切換協程的邏輯
- static void contextswitch(Context *from, Context *to)
- {
- if(swapcontext(&from->uc, &to->uc) < 0){
- fprint(2, "swapcontext failed: %r\n");
- assert(0);
- }
- }
yield的邏輯也很簡單,因為協程在執行的時候,是不處于就緒隊列的,當協程準備讓出cpu時,協程首先把自己重新加入到就緒隊列,等待下次被調度執行。當然我們也可以直接調度contextswitch切換到其他協程。重點在于什么時候應該讓出cpu,又什么時候應該被調度執行。接下來會詳細講解。至此,我們已經有了支持協程所需要的底層基礎。我們看到這個實現的思路也不是很復雜,首先有一個隊列表示待執行的的協程,每一個協程對應一個Task結構體。然后調度中心不斷地按照先進先出的方式去調度協程的執行就可以。因為沒有搶占機制,所以調度中心是依賴協程本身去驅動的,協程需要主動讓出cpu,把上下文切換回調度中心,調度中心才能進行下一輪的調度。接下來我們看看,基于這些底層基礎,如果實現一個基于協程的服務器。下面我們通過一個例子進行講解。
- void
- taskmain(int argc, char **argv)
- {
- // 啟動一個tcp服務器
- if((fd = netannounce(TCP, 0, atoi(argv[1]))) < 0){
- // ...
- }
- // 改為非阻塞模式
- fdnoblock(fd);
- // accept成功后創建一個客戶端協程
- while((cfd = netaccept(fd, remote, &rport)) >= 0){
- taskcreate(proxytask, (void*)cfd, STACK);
- }
- }
我們剛才講過taskmain是我們需要實現的函數,首先通過netannounce建立一個tcp服務器。接著把fd改成非阻塞的,這個非常重要,因為在后面調用accept的時候,如果是阻塞的文件描述符,那么就會引起進程掛起,而非阻塞模式下,操作系統會返回EAGAIN的錯誤碼,通過這個錯誤碼我們可以決定下一步做什么。我們看看netaccept的實現。
- // 處理(摘下)連接
- int
- netaccept(int fd, char *server, int *port)
- {
- int cfd, one;
- struct sockaddr_in sa;
- uchar *ip;
- socklen_t len;
- // 注冊事件到epoll,等待事件觸發
- fdwait(fd, 'r');
- len = sizeof sa;
- // 觸發后說明有連接了,則執行accept
- if((cfd = accept(fd, (void*)&sa, &len)) < 0){
- return -1;
- }
- // 和客戶端通信的fd也改成非阻塞模式
- fdnoblock(cfd);
- one = 1;
- setsockopt(cfd, IPPROTO_TCP, TCP_NODELAY, (char*)&one, sizeof one);
- return cfd;
- }
netaccept就是通過調用accept逐個處理tcp連接,但是在accept之前,有一個非常重要的操作fdwait。
- // 協程因為等待io需要切換
- void fdwait(int fd, int rw)
- {
- // 是否已經初始化epoll
- if(!startedfdtask){
- startedfdtask = 1;
- epfd = epoll_create(1);
- // 沒有初始化則創建一個協程,做io管理
- taskcreate(fdtask, 0, 32768);
- }
- struct epoll_event ev = {0};
- // 記錄事件對應的協程和感興趣的事件
- ev.data.ptr = taskrunning;
- switch(rw){
- case 'r':
- ev.events |= EPOLLIN | EPOLLPRI;
- break;
- case 'w':
- ev.events |= EPOLLOUT;
- break;
- }
- int r = epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev);
- // 切換到其他協程,等待被喚醒
- taskswitch();
- // 喚醒后函數剛才注冊的事件
- epoll_ctl(epfd, EPOLL_CTL_DEL, fd, &ev);
- }
fdwait首先把fd注冊到epoll中,然后把協程切換到下一個待執行的協程。這里有個細節,當協程X被調度執行的時候,他是脫離了就緒隊列的,而taskswitch函數只是實現了切換上下文到調度中心,調度中心會從就緒隊列從選擇下一個協程執行,那么這時候,脫離就緒隊列的協程X就處于孤島狀態,看起來再也無法給調度中心選中執行,這個問題的處理方式是,把協程、fd和感興趣的事件信息一起注冊到epoll中,當epoll監聽到某個fd的事件發生時,就會把對應的協程加入就緒隊列,這樣協程就可以被調度執行了。在fdwait函數一開始那里處理了epoll相關的邏輯。epoll的邏輯也是在一個協程中執行的,但是epoll所在協程和一般協程不一樣,類似于操作系統的內核線程一樣,epoll所在的協程成為系統協程,即不是用戶定義的,而是系統定義的。我們看一下實現
- void fdtask(void *v)
- {
- int i, ms;
- Task *t;
- uvlong now;
- // 變成系統協程
- tasksystem();
- struct epoll_event events[1000];
- for(;;){
- /* let everyone else run */
- // 大于0說明還有其他就緒協程可執行,則先讓給他們執行,否則往下執行
- while(taskyield() > 0)
- ;
- /* we're the only one runnable - poll for i/o */
- errno = 0;
- // 沒有定時事件則一直阻塞
- if((t=sleeping.head) == nil)
- ms = -1;
- else{
- /* sleep at most 5s */
- now = nsec();
- if(now >= t->alarmtime)
- ms = 0;
- else if(now+5*1000*1000*1000LL >= t->alarmtime)
- ms = (t->alarmtime - now)/1000000;
- else
- ms = 5000;
- }
- int nevents;
- // 等待事件發生,ms是等待的超時時間
- if((nevents = epoll_wait(epfd, events, 1000, ms)) < 0){
- if(errno == EINTR)
- continue;
- fprint(2, "epoll: %s\n", strerror(errno));
- taskexitall(0);
- }
- /* wake up the guys who deserve it */
- // 事件觸發,把對應協程插入就緒隊列
- for(i=0; i<nevents; i++){
- taskready((Task *)events[i].data.ptr);
- }
- now = nsec();
- // 處理超時事件
- while((t=sleeping.head) && now >= t->alarmtime){
- deltask(&sleeping, t);
- if(!t->system && --sleepingcounted == 0)
- taskcount--;
- taskready(t);
- }
- }
- }
我們看到epoll的處理邏輯和一般服務器的類似,通過epoll_wait阻塞,然后epoll_wait返回時,處理每一個發生的事件,而且libtask還支持超時事件。另外libtask中當還有其他就緒協程的時候,是不會進入epoll_wait的,它會把cpu讓給就緒的協程(通過taskyield函數),當就緒隊列只有epoll所在的協程時才會進入epoll的邏輯。至此,我們看到了libtask中如何把異步變成同步的。當用戶要調用一個可能會引起進程掛起的接口時,就可以調用libtask提供的一個相應的API,比如我們想讀一個文件,我們可以調用libtask的fdread。
- int
- fdread(int fd, void *buf, int n)
- {
- int m;
- // 非阻塞讀,如果不滿足則再注冊到epoll,參考fdread1
- while((m=read(fd, buf, n)) < 0 && errno == EAGAIN)
- fdwait(fd, 'r');
- return m;
- }
這樣就不需要擔心進程被掛起,同時也不需要處理epoll相關的邏輯(注冊事件,事件觸發時的處理等等)。異步轉同步,libtask的方式就是通過提供對應的API,先把用戶的fd注冊到epoll中,然后切換到其他協程,等epoll監聽到事件觸發時,就會把對應的協程插入就緒隊列,當該協程被調度中心選中執行時,就會繼續執行剩下的邏輯而不會引起進程掛起,因為這時候所等待的條件已經滿足。
總結:libtask的設計思想就是把業務邏輯封裝到一個個協程中,由libtask實現協程的調度,在各個業務邏輯中進行切換,從而驅動著系統的運行。另外libtask也提供了一個網絡和文件io異步變同步的解決方案。使得我們使用起來更加方便,高效。今天先講到這里。
References
[1] 更多基礎可以點擊這里: https://github.com/theanarkh/read-libtask-code/blob/main/README.md