[[382125]]

本文轉(zhuǎn)載自微信公眾號(hào)「編程雜技」，作者theanarkh 。轉(zhuǎn)載本文請(qǐng)聯(lián)系編程雜技公眾號(hào)。

前言：假設(shè)讀者已經(jīng)了解了協(xié)程的概念，實(shí)現(xiàn)協(xié)程的底層技術(shù)支持。本文會(huì)介紹基于底層基礎(chǔ)，如何實(shí)現(xiàn)協(xié)程以及協(xié)程的應(yīng)用(更多基礎(chǔ)可以點(diǎn)擊這里[1])。

libtask是google大佬Russ Cox(Go的核心開(kāi)發(fā)者)所寫(xiě)，本文介紹libtask的基礎(chǔ)原理。我們從libtask的main函數(shù)開(kāi)始，這個(gè)main函數(shù)就是我們?cè)赾語(yǔ)言中使用的c函數(shù)，libtask本身實(shí)現(xiàn)了main這個(gè)函數(shù)，用戶使用libtask時(shí)，要實(shí)現(xiàn)的是taskmain函數(shù)。taskmain和main的函數(shù)聲明是一樣的。下面我們看一下main函數(shù)。

int main(int argc, char **argv) 
{ 
    struct sigaction sa, osa; 
    // 注冊(cè)SIGQUIT信號(hào)處理函數(shù) 
    memset(&sa, 0, sizeof sa); 
    sa.sa_handler = taskinfo; 
    sa.sa_flags = SA_RESTART; 
    sigaction(SIGQUIT, &sa, &osa); 
 
    // 保存命令行參數(shù) 
    argv0 = argv[0]; 
    taskargc = argc; 
    taskargv = argv; 
 
    if(mainstacksize == 0) 
        mainstacksize = 256*1024; 
    // 創(chuàng)建第一個(gè)協(xié)程 
    taskcreate(taskmainstart, nil, mainstacksize); 
    // 開(kāi)始調(diào)度 
    taskscheduler(); 
    fprint(2, "taskscheduler returned in main!\n"); 
    abort(); 
    return 0; 
} 

main函數(shù)主要的兩個(gè)邏輯是taskcreate和taskscheduler函數(shù)。我們先來(lái)看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; 
    // 修改狀態(tài)為就緒，可以被調(diào)度，并且加入到就緒隊(duì)列 
    taskready(t); 
    return id; 
} 

taskcreate首先調(diào)用taskalloc分配一個(gè)表示協(xié)程的結(jié)構(gòu)體Task。我們看看這個(gè)結(jié)構(gòu)體的定義。

struct Task 
{ 
    char    name[256];    // offset known to acid 
    char    state[256]; 
    // 前后指針 
    Task    *next; 
    Task    *prev; 
    Task    *allnext; 
    Task    *allprev; 
    // 執(zhí)行上下文 
    Context    context; 
    // 睡眠時(shí)間 
    uvlong    alarmtime; 
    uint    id; 
    // 棧信息 
    uchar    *stk; 
    uint    stksize; 
    //是否退出了 
    int    exiting; 
    // 在alltask的索引 
    int    alltaskslot; 
    // 是否是系統(tǒng)協(xié)程 
    int    system; 
    // 是否就緒狀態(tài) 
    int    ready; 
    // 入口函數(shù) 
    void    (*startfn)(void*); 
    // 入口參數(shù) 
    void    *startarg; 
    // 自定義數(shù)據(jù) 
    void    *udata; 
}; 

接著看看taskalloc的實(shí)現(xiàn)。

// 分配一個(gè)協(xié)程所需要的內(nèi)存和初始化某些字段 
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 */ 
    // 結(jié)構(gòu)體本身的大小+棧大小 
    t = malloc(sizeof *t+stack); 
    memset(t, 0, sizeof *t); 
    // 棧的內(nèi)存位置 
    t->stk = (uchar*)(t+1); 
    // 棧大小 
    t->stksize = stack; 
    // 協(xié)程id 
    t->id = ++taskidgen; 
    // 協(xié)程工作函數(shù)和參數(shù) 
    t->startfn = fn; 
    t->startarg = arg; 
 
    /* do a reasonable initialization */ 
    memset(&t->context.uc, 0, sizeof t->context.uc); 
    sigemptyset(&zero); 
    // 初始化uc_sigmask字段為空，即不阻塞信號(hào) 
    sigprocmask(SIG_BLOCK, &zero, &t->context.uc.uc_sigmask); 
 
    /* must initialize with current context */ 
    // 初始化uc字段 
    getcontext(&t->context.uc)  
    // 設(shè)置協(xié)程執(zhí)行時(shí)的棧位置和大小 
    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函數(shù)代碼看起來(lái)很多，但是邏輯不算復(fù)雜，就是申請(qǐng)Task結(jié)構(gòu)體所需的內(nèi)存和執(zhí)行時(shí)棧的內(nèi)存，然后初始化各個(gè)字段。這樣，一個(gè)協(xié)程就誕生了。接著執(zhí)行taskready把協(xié)程加入就緒隊(duì)列。

// 修改協(xié)程的狀態(tài)為就緒并加入就緒隊(duì)列 
void taskready(Task *t) 
{ 
    t->ready = 1; 
    addtask(&taskrunqueue, t); 
} 
 
// 把協(xié)程插入隊(duì)列中，如果之前在其他隊(duì)列，則會(huì)被移除 
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記錄了所有就緒的協(xié)程。創(chuàng)建了協(xié)程并加入隊(duì)列后，協(xié)程還沒(méi)有開(kāi)始執(zhí)行，就像操作系統(tǒng)的進(jìn)程和線程一樣，需要有一個(gè)調(diào)度器來(lái)調(diào)度執(zhí)行。下面我們看看調(diào)度器的實(shí)現(xiàn)。

// 協(xié)程調(diào)度中心 
static void taskscheduler(void) 
{ 
    int i; 
    Task *t; 
    for(;;){ 
        // 沒(méi)有用戶協(xié)程了，則退出 
        if(taskcount == 0) 
            exit(taskexitval); 
        // 從就緒隊(duì)列拿出一個(gè)協(xié)程 
        t = taskrunqueue.head; 
        if(t == nil){ 
            fprint(2, "no runnable tasks! %d tasks stalled\n", taskcount); 
            exit(1); 
        } 
        // 從就緒隊(duì)列刪除該協(xié)程 
        deltask(&taskrunqueue, t); 
        t->ready = 0; 
        // 保存正在執(zhí)行的協(xié)程 
        taskrunning = t; 
        // 切換次數(shù)加一 
        tasknswitch++; 
        // 切換到t執(zhí)行，并且保存當(dāng)前上下文到taskschedcontext（即下面要執(zhí)行的代碼） 
        contextswitch(&taskschedcontext, &t->context); 
        // 執(zhí)行到這說(shuō)明沒(méi)有協(xié)程在執(zhí)行（t切換回來(lái)的），置空 
        taskrunning = nil; 
        // 剛才執(zhí)行的協(xié)程t退出了 
        if(t->exiting){ 
            // 不是系統(tǒng)協(xié)程，則個(gè)數(shù)減一 
            if(!t->system) 
                taskcount--; 
            // 當(dāng)前協(xié)程在alltask的索引 
            i = t->alltaskslot; 
            // 把最后一個(gè)協(xié)程換到當(dāng)前協(xié)程的位置，因?yàn)樗顺隽?nbsp;
            alltask[i] = alltask[--nalltask]; 
            // 更新被置換協(xié)程的索引 
            alltask[i]->alltaskslot = i; 
            // 釋放堆內(nèi)存 
            free(t); 
        } 
    } 
} 

調(diào)度器的代碼看起來(lái)很多，但是核心邏輯就三個(gè) 1 從就緒隊(duì)列中拿出一個(gè)協(xié)程t，并把t移出就緒隊(duì)列 2 通過(guò)contextswitch切換到協(xié)程t中執(zhí)行 3 協(xié)程t切換回調(diào)度中心，如果t已經(jīng)退出，則修改數(shù)據(jù)結(jié)構(gòu)，然后回收他占據(jù)的內(nèi)存。如果t沒(méi)退出，則繼續(xù)調(diào)度其他協(xié)程執(zhí)行。至此，協(xié)程就開(kāi)始跑起來(lái)了。并且也有了調(diào)度系統(tǒng)。這里的調(diào)度機(jī)制是比較簡(jiǎn)單的，就是按著先進(jìn)先出的方式就緒調(diào)度，并且是非搶占的。即沒(méi)有按時(shí)間片調(diào)度的概念，一個(gè)協(xié)程的執(zhí)行時(shí)間由自己決定，放棄執(zhí)行的權(quán)力也是自己控制的，當(dāng)協(xié)程不想執(zhí)行了可以調(diào)用taskyield讓出cpu。

// 協(xié)程主動(dòng)讓出cpu 
int taskyield(void) 
{ 
    int n; 
    // 當(dāng)前切換協(xié)程的次數(shù) 
    n = tasknswitch; 
    // 插入就緒隊(duì)列，等待后續(xù)的調(diào)度 
    taskready(taskrunning); 
    taskstate("yield"); 
    // 切換協(xié)程 
    taskswitch(); 
    // 等于0說(shuō)明當(dāng)前只有自己一個(gè)協(xié)程，調(diào)度的時(shí)候tasknswitch加一，所以這里減一 
    return tasknswitch - n - 1; 
} 
 
/* 
    切換協(xié)程,taskrunning是正在執(zhí)行的協(xié)程，taskschedcontext是調(diào)度協(xié)程（主線程）的上下文， 
    切換到調(diào)度中心，并保持當(dāng)前上下文到taskrunning->context 
*/ 
void taskswitch(void) 
{ 
    needstack(0); 
    contextswitch(&taskrunning->context, &taskschedcontext); 
} 
 
// 真正切換協(xié)程的邏輯 
static void contextswitch(Context *from, Context *to) 
{ 
    if(swapcontext(&from->uc, &to->uc) < 0){ 
        fprint(2, "swapcontext failed: %r\n"); 
        assert(0); 
    } 
} 

yield的邏輯也很簡(jiǎn)單，因?yàn)閰f(xié)程在執(zhí)行的時(shí)候，是不處于就緒隊(duì)列的，當(dāng)協(xié)程準(zhǔn)備讓出cpu時(shí)，協(xié)程首先把自己重新加入到就緒隊(duì)列，等待下次被調(diào)度執(zhí)行。當(dāng)然我們也可以直接調(diào)度contextswitch切換到其他協(xié)程。重點(diǎn)在于什么時(shí)候應(yīng)該讓出cpu，又什么時(shí)候應(yīng)該被調(diào)度執(zhí)行。接下來(lái)會(huì)詳細(xì)講解。至此，我們已經(jīng)有了支持協(xié)程所需要的底層基礎(chǔ)。我們看到這個(gè)實(shí)現(xiàn)的思路也不是很復(fù)雜，首先有一個(gè)隊(duì)列表示待執(zhí)行的的協(xié)程，每一個(gè)協(xié)程對(duì)應(yīng)一個(gè)Task結(jié)構(gòu)體。然后調(diào)度中心不斷地按照先進(jìn)先出的方式去調(diào)度協(xié)程的執(zhí)行就可以。因?yàn)闆](méi)有搶占機(jī)制，所以調(diào)度中心是依賴協(xié)程本身去驅(qū)動(dòng)的，協(xié)程需要主動(dòng)讓出cpu，把上下文切換回調(diào)度中心，調(diào)度中心才能進(jìn)行下一輪的調(diào)度。接下來(lái)我們看看，基于這些底層基礎(chǔ)，如果實(shí)現(xiàn)一個(gè)基于協(xié)程的服務(wù)器。下面我們通過(guò)一個(gè)例子進(jìn)行講解。

void 
taskmain(int argc, char **argv) 
{ 
    // 啟動(dòng)一個(gè)tcp服務(wù)器 
    if((fd = netannounce(TCP, 0, atoi(argv[1]))) < 0){ 
        // ... 
    } 
    // 改為非阻塞模式 
    fdnoblock(fd); 
    // accept成功后創(chuàng)建一個(gè)客戶端協(xié)程 
    while((cfd = netaccept(fd, remote, &rport)) >= 0){ 
        taskcreate(proxytask, (void*)cfd, STACK); 
    } 
} 

我們剛才講過(guò)taskmain是我們需要實(shí)現(xiàn)的函數(shù)，首先通過(guò)netannounce建立一個(gè)tcp服務(wù)器。接著把fd改成非阻塞的，這個(gè)非常重要，因?yàn)樵诤竺嬲{(diào)用accept的時(shí)候，如果是阻塞的文件描述符，那么就會(huì)引起進(jìn)程掛起，而非阻塞模式下，操作系統(tǒng)會(huì)返回EAGAIN的錯(cuò)誤碼，通過(guò)這個(gè)錯(cuò)誤碼我們可以決定下一步做什么。我們看看netaccept的實(shí)現(xiàn)。

// 處理（摘下）連接 
int 
netaccept(int fd, char *server, int *port) 
{ 
    int cfd, one; 
    struct sockaddr_in sa; 
    uchar *ip; 
    socklen_t len; 
    // 注冊(cè)事件到epoll，等待事件觸發(fā) 
    fdwait(fd, 'r'); 
    len = sizeof sa; 
    // 觸發(fā)后說(shuō)明有連接了，則執(zhí)行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就是通過(guò)調(diào)用accept逐個(gè)處理tcp連接，但是在accept之前，有一個(gè)非常重要的操作fdwait。

// 協(xié)程因?yàn)榈却齣o需要切換 
void fdwait(int fd, int rw) 
{     
    // 是否已經(jīng)初始化epoll 
    if(!startedfdtask){ 
        startedfdtask = 1; 
        epfd = epoll_create(1); 
        // 沒(méi)有初始化則創(chuàng)建一個(gè)協(xié)程，做io管理 
        taskcreate(fdtask, 0, 32768); 
    } 
    struct epoll_event ev = {0}; 
    // 記錄事件對(duì)應(yīng)的協(xié)程和感興趣的事件 
    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); 
    // 切換到其他協(xié)程，等待被喚醒 
    taskswitch(); 
    // 喚醒后函數(shù)剛才注冊(cè)的事件 
    epoll_ctl(epfd, EPOLL_CTL_DEL, fd, &ev); 
} 

fdwait首先把fd注冊(cè)到epoll中，然后把協(xié)程切換到下一個(gè)待執(zhí)行的協(xié)程。這里有個(gè)細(xì)節(jié)，當(dāng)協(xié)程X被調(diào)度執(zhí)行的時(shí)候，他是脫離了就緒隊(duì)列的，而taskswitch函數(shù)只是實(shí)現(xiàn)了切換上下文到調(diào)度中心，調(diào)度中心會(huì)從就緒隊(duì)列從選擇下一個(gè)協(xié)程執(zhí)行，那么這時(shí)候，脫離就緒隊(duì)列的協(xié)程X就處于孤島狀態(tài)，看起來(lái)再也無(wú)法給調(diào)度中心選中執(zhí)行，這個(gè)問(wèn)題的處理方式是，把協(xié)程、fd和感興趣的事件信息一起注冊(cè)到epoll中，當(dāng)epoll監(jiān)聽(tīng)到某個(gè)fd的事件發(fā)生時(shí)，就會(huì)把對(duì)應(yīng)的協(xié)程加入就緒隊(duì)列，這樣協(xié)程就可以被調(diào)度執(zhí)行了。在fdwait函數(shù)一開(kāi)始那里處理了epoll相關(guān)的邏輯。epoll的邏輯也是在一個(gè)協(xié)程中執(zhí)行的，但是epoll所在協(xié)程和一般協(xié)程不一樣，類似于操作系統(tǒng)的內(nèi)核線程一樣，epoll所在的協(xié)程成為系統(tǒng)協(xié)程，即不是用戶定義的，而是系統(tǒng)定義的。我們看一下實(shí)現(xiàn)

void fdtask(void *v) 
{ 
    int i, ms; 
    Task *t; 
    uvlong now; 
    // 變成系統(tǒng)協(xié)程 
    tasksystem(); 
    struct epoll_event events[1000]; 
    for(;;){ 
        /* let everyone else run */ 
        // 大于0說(shuō)明還有其他就緒協(xié)程可執(zhí)行，則先讓給他們執(zhí)行，否則往下執(zhí)行 
        while(taskyield() > 0) 
            ; 
        /* we're the only one runnable - poll for i/o */ 
        errno = 0; 
        // 沒(méi)有定時(shí)事件則一直阻塞 
        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; 
        // 等待事件發(fā)生，ms是等待的超時(shí)時(shí)間 
        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 */ 
        // 事件觸發(fā)，把對(duì)應(yīng)協(xié)程插入就緒隊(duì)列 
        for(i=0; i<nevents; i++){ 
            taskready((Task *)events[i].data.ptr); 
        } 
 
        now = nsec(); 
        // 處理超時(shí)事件 
        while((t=sleeping.head) && now >= t->alarmtime){ 
            deltask(&sleeping, t); 
            if(!t->system && --sleepingcounted == 0) 
                taskcount--; 
            taskready(t); 
        } 
    } 
} 

我們看到epoll的處理邏輯和一般服務(wù)器的類似，通過(guò)epoll_wait阻塞，然后epoll_wait返回時(shí)，處理每一個(gè)發(fā)生的事件，而且libtask還支持超時(shí)事件。另外libtask中當(dāng)還有其他就緒協(xié)程的時(shí)候，是不會(huì)進(jìn)入epoll_wait的，它會(huì)把cpu讓給就緒的協(xié)程(通過(guò)taskyield函數(shù))，當(dāng)就緒隊(duì)列只有epoll所在的協(xié)程時(shí)才會(huì)進(jìn)入epoll的邏輯。至此，我們看到了libtask中如何把異步變成同步的。當(dāng)用戶要調(diào)用一個(gè)可能會(huì)引起進(jìn)程掛起的接口時(shí)，就可以調(diào)用libtask提供的一個(gè)相應(yīng)的API，比如我們想讀一個(gè)文件，我們可以調(diào)用libtask的fdread。

int 
fdread(int fd, void *buf, int n) 
{ 
    int m; 
    // 非阻塞讀，如果不滿足則再注冊(cè)到epoll，參考fdread1 
    while((m=read(fd, buf, n)) < 0 && errno == EAGAIN) 
        fdwait(fd, 'r'); 
    return m; 
} 

這樣就不需要擔(dān)心進(jìn)程被掛起，同時(shí)也不需要處理epoll相關(guān)的邏輯(注冊(cè)事件，事件觸發(fā)時(shí)的處理等等)。異步轉(zhuǎn)同步，libtask的方式就是通過(guò)提供對(duì)應(yīng)的API，先把用戶的fd注冊(cè)到epoll中，然后切換到其他協(xié)程，等epoll監(jiān)聽(tīng)到事件觸發(fā)時(shí)，就會(huì)把對(duì)應(yīng)的協(xié)程插入就緒隊(duì)列，當(dāng)該協(xié)程被調(diào)度中心選中執(zhí)行時(shí)，就會(huì)繼續(xù)執(zhí)行剩下的邏輯而不會(huì)引起進(jìn)程掛起，因?yàn)檫@時(shí)候所等待的條件已經(jīng)滿足。

總結(jié)：libtask的設(shè)計(jì)思想就是把業(yè)務(wù)邏輯封裝到一個(gè)個(gè)協(xié)程中，由libtask實(shí)現(xiàn)協(xié)程的調(diào)度，在各個(gè)業(yè)務(wù)邏輯中進(jìn)行切換，從而驅(qū)動(dòng)著系統(tǒng)的運(yùn)行。另外libtask也提供了一個(gè)網(wǎng)絡(luò)和文件io異步變同步的解決方案。使得我們使用起來(lái)更加方便，高效。今天先講到這里。

References

[1] 更多基礎(chǔ)可以點(diǎn)擊這里: https://github.com/theanarkh/read-libtask-code/blob/main/README.md