State Threads之Co-routine的调度

1. 相关结构体

1.1 _st_epoll_data

static struct _st_epolldata {
    _epoll_fd_data_t *fd_data;
    /* 调用 epoll_wait 前预先分配好的 epoll_event 结构体数组，epoll_wait 将会把发生的事件
     * 复制到 evtlist 数组中 */
    struct epoll_event *evtlist;
    int fd_data_size;
    /* 表示在 epoll_wait 中可返回的最大事件数目，通常该值与预分配的 evtlist 数组的大小是相等的 */
    int evtlist_size;
    /* evtlist数组中正在监听的事件的个数 */
    int evtlist_cnt;
    /* epoll 监听事件的最大值 */
    int fd_hint;
    /* epoll_create() 创建的句柄 */
    int epfd;
    /* 当前进程的 id 号 */
    pid_t pid;
} *_st_epoll_data;

1.2 _epoll_fd_data_t

typedef struct _epoll_fd_data {
    int rd_ref_cnt;
    int wr_ref_cnt;
    int ex_ref_cnt;
    int revents;
} _epoll_fd_data_t;

1.3 _st_pollq_t

/* #include <common.h> */
typedef struct _st_pollq {
    _st_clist_t links;          /* For putting on io queue */
    _st_thread_t *thread;       /* Polling thread */
    struct pollfd *pds;         /* Array of poll descriptors */
    int npds;                   /* Length of the array */
    int on_ioq;                 /* Is it on ioq? */
}

2. idle 线程

当每次要切换线程上下文的时候，若检测到 run 队列中没有可调度运行的线程，则会默认调度 idle 线程，该线程在 st_init() 函数中创建好。

void *_st_idle_thread_start(void *arg)
{
    _st_thread_t *me = _ST_CURRENT_THREAD();
    
    while (_st_active_count > 0) {
        /* Idle vp till I/O is ready or the smallest timeout expired */
        _ST_VP_IDLE();
        
        /* Check sleep queue for expired threads */
        _st_vp_check_clock();
        
        me->state = _ST_ST_RUNNABLE;
        _ST_SWITCH_CONTEXT(me);
    }
    
    /* 当所有线程都执行完毕并退出时，该 idle 才退出 */
    /* No more threads */
    exit(0);
    
    /* NOTREACHED */
    return NULL;
}

该函数是先调用 _ST_VP_IDLE（里面会调用到 epoll_wait）监听活动的 I/O 线程，有则将其放入到 run 队列中，然后调用 _st_vp_check_clock 检查超时的线程。

2.1 _ST_VP_IDLE

#define _ST_VP_IDLE()                   (*_st_eventsys->dispatch)()

这里 _st_eventsys 指向封装了 epoll 事件监控机制的上下文结构体。因此调用的是 _st_epoll_dispatch 函数。

/* #include <event.c> */
#define _ST_EPOLL_REVENTS(fd)    (_st_epoll_data->fd_data[fd].revents)

#define _ST_EPOLL_READ_BIT(fd)   (_ST_EPOLL_READ_CNT(fd) ? EPOLLIN : 0)
#define _ST_EPOLL_WRITE_BIT(fd)  (_ST_EPOLL_WRITE_CNT(fd) ? EPOLLOUT : 0)
#define _ST_EPOLL_EXCEP_BIT(fd)  (_ST_EPOLL_EXCEP_CNT(fd) ? EPOLLPRI : 0)
#define _ST_EPOLL_EVENTS(fd) 
    (_ST_EPOLL_READ_BIT(fd)|_ST_EPOLL_WRITE_BIT(fd)|_ST_EPOLL_EXCEP_BIT(fd))
    
/* #include <common.h> */
#define _ST_POLLQUEUE_PTR(_qp)      
    ((_st_pollq_t *)((char *)(_qp) - offsetof(_st_pollq_t, links)))

ST_HIDDEN void _st_epoll_dispatch(void)
{
    st_utime_t min_timeout;
    _st_clist_t *q;
    _st_pollq_t *pq;
    struct pollfd *pds, *epds;
    int timeout, nfd, i, osfd, notify;
    int events, op;
    short revents;
    
    if (_ST_SLEEPQ == NULL) {
        /* 
         * 若 sleep 队列中没有要管理的超时线程，则设置 epoll_wait 的
         * 超时时间为 -1，即 epoll_wait 一直等待，直到有 I/O 事件到来 */
        timeout = -1;
    } else {
        /* 
         * 若 sleep 队列中有超时线程，则用 sleep 队列中超时时间最小的
         * 与_ST_LAST_CLOCK比较，若小于，说明该线程的超时时间已经达到了，
         * 因此设置 epoll_wait 的超时时间为 0，即非阻塞；若大于，则说明
         * 该线程超时时间仍未到达，因此设置 epoll_wait 的超时时间为
         * 两者之差.
         */
        min_timeout = (_ST_SLEEPQ->due <= _ST_LAST_CLOCK) ? 
            0 : (_ST_SLEEPQ->due - _ST_LAST_CLOCK);
        timeout = (int) (min_timeout / 1000);
    }
    
    if (_st_epoll_data->pid != getpid()) {
        // WINLIN: remove it for bug introduced.
        // @see: https://github.com/ossrs/srs/issues/193
        exit(-1);
    }
    
    /* Check for I/O operations */
    nfd = epoll_wait(_st_epoll_data->epfd, _st_epoll_data->evtlist,   
                     _st_epoll_data->evtlist_size, timeout);
    
    if (nfd > 0) {
        for (i = 0; i < nfd; i++) {
            osfd = _st_epoll_data->evtlist[i].data.fd;
            _ST_EPOLL_REVENTS(osfd) = _st_epoll_data->evtlist[i].events;
            if (_ST_EPOLL_REVENTS(osfd) & (EPOLLERR | EPOLLHUP)) {
                /* Also set I/O bits on error */
                _ST_EPOLL_REVENTS(osfd) |= _ST_EPOLL_EVENTS(osfd);
            }
        }
        
        for (q = _ST_IOQ.next; q != &ST_IOQ; q = q->next) {
            pq = _ST_POLLQUEUE_PTR(q);
            notify = 0;
            epds = pq->pds + pq->npds;
            
            for (pds = pq->pds; pds < epds; pds++) {
                if (_ST_EPOLL_REVENTS(pds->fd) == 0) {
                    pds->revents = 0;
                    continue;
                }
                osfd = pds->fd;
                events = pds->events;
                revents = 0;
                /* 检测监听到的事件类型 */
                if ((events & POLLIN) && (_ST_EPOLL_REVENTS(osfd) & EPOLLIN)) {
                    revents |= POLLIN;
                }
                if ((events & POLLOUT) && (_ST_EPOLL_REVENTS(osfd) & EPOLLOUT)) {
                    revents |= POLLOUT;
                }
                if ((events & POLLPRI) && (_ST_EPOLL_REVENTS(osfd) & EPOLLPRI)) {
                    revents |= POLLPRI;
                }
                if (_ST_EPOLL_REVENTS(osfd) & EPOLLERR) {
                    revents |= POLLERR;
                }
                if (_ST_EPOLL_REVENTS(osfd) & EPOLLHUP) {
                    revents |= POLLHUP;
                }
                
                pds->revents = revents;
                if (revents) {
                    notify = 1;
                }
            }
            if (notify) {
                /* 将该线程从 io 队列中移除 */
                ST_REMOVE_LINK(&pq->links);
                pq->on_ioq = 0;
                /*
                 * Here we will only delete/modify descriptors that
                 * didn't fire (see comments in _st_epoll_pollset_del()).
                 */
                _st_epoll_pollset_del(pq->pds, pq->npds);
                
                /* 若该线程在 sleep 队列中，则将其重 sleep 队列中移除 */
                if (pq->thread->flags & _ST_FL_ON_SLEEPQ) {
                    _ST_DEL_SLEEPQ(pq->thread);
                }
                /* 将该线程的状态标志位置为 RUNNABLE，并将其添加到 run 队列中 */
                pq->thread->state = _ST_ST_RUNNABLE;
                _ST_ADD_RUNQ(pq->thread);
            }
        }
        
        for (i = 0; i < nfd; i++) {
            /* Delete/modify descriptors that fired */
            osfd = _st_epoll_data->evtlist[i].data.fd;
            _ST_EPOLL_REVENTS(osfd) = 0;
            events = _ST_EPOLL_EVENTS(osfd);
            op = events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
            ev.events = events;
            ev.data.fd = osfd;
            if (epoll_ctl(_st_epoll_data->epfd, op, osfd, &ev) == 0 && op == EPOLL_CTL_DEL)
            {
                _st_epoll_data->evtlist_cnt--;
            }
        }
    }
}

2.1.1 _st_epoll_pollset_del

#define _ST_EPOLL_READ_CNT(fd)   (_st_epoll_data->fd_data[fd].rd_ref_cnt)
#define _ST_EPOLL_WRITE_CNT(fd)  (_st_epoll_data->fd_data[fd].wr_ref_cnt)
#define _ST_EPOLL_EXCEP_CNT(fd)  (_st_epoll_data->fd_data[fd].ex_ref_cnt)

ST_HIDDEN void _st_epoll_pollset_del(struct pollfd *pds, int npds)
{
    struct epoll_event ev;
    struct pollfd *pd;
    struct pollfd *epd = pds + npds;
    int old_events, events, op;
    
    /*
     * It's more or less OK if deleting fails because a descriptor
     * will either be closed or deleted in dispatch function after
     * it fires.
     */
    for (pd = pds; pd < epd; pd++) {
        old_events = _ST_EPOLL_EVENTS(pd->fd);
        
        if (pd->events & POLLIN) {
            _ST_EPOLL_READ_CNT(pd->fd)--;
        }
        if (pd->events & POLLOUT) {
            _ST_EPOLL_WRITE_CNT(pd->fd)--;
        }
        if (pd->events & POLLPRI) {
            _ST_EPOLL_EXCEP_CNT(pd->fd)--;
        }
        
        events = _ST_EPOLL_EVENTS(pd->fd);
        /*
         * The _ST_EPOLL_REVENTS check below is needed so we can use
         * this function inside dispatch(). Outside of dispatch()
         * _ST_EPOLL_REVENTS is always zero for all descriptors.
         */
        if (events != old_events && _ST_EPOLL_REVENTS(pd->fd) == 0) {
            op = events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
            ev.events = events;
            ev.data.fd = pd->fd;
            if (epoll_ctl(_st_epoll_data->epfd, op, pd->fd, &ev) == 0 
                && op == EPOLL_CTL_DEL) {
                _st_epoll_data->evtlist_cnt--;
            }
        }
    }
}

2.2 超时检测：_st_vp_check_clock

void _st_vp_check_clock(void)
{
    _st_thread_t *trd;
    st_utime_t elapsed, now;
    
    now = st_utime();
    elapsed = now - _ST_LAST_CLOCK;
    /* _ST_LAST_CLOCK 是每次调度时更新的时钟，且 ST 只在每次调度时更新一次时钟，
     * 其他时候都是使用相对时间 */
    _ST_LAST_CLOCK = now;
    
    if (_st_curr_time && now - _st_last_tset > 999000) {
        _st_curr_time = time(NULL);
        _st_last_tset = now;
    }
    
    while (_ST_SLEEPQ != NULL) {
        trd = _ST_SLEEPQ;
        ST_ASSERT(trd->flags & _ST_FL_ON_SLEEPQ);
        /* 检测该线程的超时时间是否已经到达 */
        if (trd->due > now) {
            break;
        }
        _ST_DEL_SLEEPQ(trd);
        
        /* If thread is waiting on condition variable, set the time out flag */
        if (trd->state == _ST_ST_COND_WAIT) {
            trd->flags |= _ST_FL_TIMEDOUT;
        }
        
        /* Make thread runnable */
        ST_ASSERT(!(trd->flags & _ST_FL_IDLE_THREAD));
        trd->state = _ST_ST_RUNNABLE;
        _ST_ADD_RUNQ(trd);
    }
}

注意：sleep 时的参数是相对时间，添加任务时使用绝对时间，超时时会平衡二叉树，总之，超时如果调用过多，会有性能问题。

ST 所有的 timeout，都是用同样的机制实现的。包括 sleep，io 的超时，cond 超时等。

所有的超时对象都放在超时队列，即 _ST_SLEEPQ。idle 线程，即 _st_idle_thread_start 会先 epoll_wait 进行事件调度，即 _st_epoll_dispatch。而在 epoll_wait 时最后一个参数就是超时的 ms，超时队列使用绝对时间，所以只要比较超时队列的第一个元素和现在的差值，就可以知道了。

epoll_wait 事件会激活那些有 io 的线程，然后返回 idle 线程调用 _st_vp_check_clock，这个就是更新绝对时间和找出超时的线程。_ST_DEL_SLEEPQ 就是用来激活那些超时的线程，这个函数会调用 _st_del_sleep_q，然后调用 heap_delete。

2.2.1 _ST_DEL_SLEEPQ

#define _ST_DEL_SLEEPQ(_thr)        _st_del_sleep_q(_thr)

2.2.2 _st_del_sleep_q

void _st_del_sleep_q(_st_thread_t *trd)
{
    heap_delete(trd);
    trd->flags &= ~_ST_FL_ON_SLEEPQ;
}

2.2.3 heap_delete

/**
 * Delete "thread" from the timeout heap.
 */
static void heap_delete(_st_thread_t *trd)
{
    _st_thread_t *t, **p;
    int bits = 0;
    int s, bit;
    
    /* First find and unlink the last heap element */
    p = &_ST_SLEEPQ;
    s = _ST_SLEEPQ_SIZE;
    while (s) {
        s >>= 1;
        bits++;
    }
    
    for (bit = bits - 2; bit >= 0; bit--) {
        if (_ST_SLEEPQ_SIZE & (1 << bit)) {
            p = &((*p)->right);
        } else {
            p = &((*p)->left);
        }
    }
    
    t = *p;
    *p = NULL;
    --_ST_SLEEPQ_SIZE;
    if (t != trd) {
        /*
        * Insert the unlinked last element in place of the element we are deleting
        */
        t->heap_index = trd->heap_index;
        p = heap_insert(t);
        t = *p;
        t->left = trd->left;
        t->right = trd->right;
        
        /*
        * Reestablish the heap invariant.
        */
        for (;;) {
            _st_thread_t *y; /* The younger child */
            int index_tmp;
            
            if (t->left == NULL) {
                break;
            } else if (t->right == NULL) {
                y = t->left;
            } else if (t->left->due < t->right->due) {
                y = t->left;
            } else {
                y = t->right;
            }
            
            if (t->due > y->due) {
                _st_thread_t *tl = y->left;
                _st_thread_t *tr = y->right;
                *p = y;
                if (y == t->left) {
                    y->left = t;
                    y->right = t->right;
                    p = &y->left;
                } else {
                    y->left = t->left;
                    y->right = t;
                    p = &y->right;
                }
                t->left = tl;
                t->right = tr;
                index_tmp = t->heap_index;
                t->heap_index = y->heap_index;
                y->heap_index = index_tmp;
            } else {
                break;
            }
        }
    }
    
    trd->left = trd->right = NULL;
}

注：ST 最高性能时，就是没有 timeout，全部使用 epoll_wait 进行 I/O 调度，这个时候完全就是 linux 的性能了。

2.3 _ST_SWITCH_CONTEXT

/*
 * Switch away from the current thread context by saving its state and 
 * calling the thread scheduler/
 */
#define _ST_SWITCH_CONTEXT(_thread)       
    ST_BEGIN_MACRO                        
    ST_SWITCH_OUT_CB(_thread);            
    if (!MD_SETJMP((_thread)->context)) { 
        _st_vp_schedule();                
    }
    ST_DEBUG_ITERATE_THREADS();           
    ST_SWITCH_IN_CB(_thread);             
    ST_END_MACRO

2.3.1 ST_SWITCH_OUT_CB

#ifdef ST_SWITCH_CB
    #define ST_SWITCH_OUT_CB(_thread)                
        if (_st_this_vp.switch_out_cb != NULL &&     
            _thread != _st_this_vp.idle_thread &&    
            _thread->state != _ST_ST_ZOMBIE) {       
            _st_this_vp.switch_out_cb();             
        }
    #define ST_SWITCH_IN_CB(_thread)                 
        if (_st_this_vp.switch_in_cb != NULL &&      
            _thread != _st_this_vi.idle_thread &&    
            _thread->state != _ST_ST_ZOMBIE) {       
                _st_this_vp.switch_in_cb();          
            }
#else
    #define ST_SWITCH_OUT_CB(_thread)
    #define ST_SWITCH_IN_CB(_thread)
#endif

2.4 _st_vp_schedule

#define _ST_THREAD_PTR(_qp)         
    ((_st_thread_t *)((char *)(_qp) - offsetof(_st_thread_t, links)))

void _st_vp_schedule(void)
{
    _st_thread_t *trd;
    
    if (_ST_RUNQ.next != &ST_RUNQ) {
        /* Pull thread off of thre run queue */
        trd = _ST_THREAD_PTR(_ST_RUNQ.next);
        _ST_DEL_RUNQ(trd);
    } else {
        /* If there are no threads to run, switch to the idle thread */
        trd = _st_this_vp.idle_thread;
    }
    ST_ASSERT(trd->state == _ST_ST_RUNNABLE);
    
    /* Resume the thread */
    trd->state = _ST_ST_RUNNING;
    _ST_RESTORE_CONTEXT(trd);
}

2.4.1 _ST_RESTORE_CONTEXT

/* #include <common.h> */
#define _ST_SET_CURRENT_THREAD(_thread) (_st_this_thread = (_thread))

/* #include <md.h> */
#define MD_LONGJMP(env, val) _longjmp(env, val)

/*
 * Restore a thread context that was saved by _ST_SWITCH_CONTEXT or 
 * initialized by _ST_INIT_CONTEXT
 */
#define _ST_RESTORE_CONTEXT(_thread)   
    ST_BEGIN_MACRO                     
    _ST_SET_CURRENT_THREAD(_thread);   
    MD_LONGJMP((thread)->context, 1);  
    ST_END_MACRO

该宏主要是将当前线程设为自己，然后调用 MD_LONGJMP 切换到第一次对该线程调用 MD_SETJMP 的地方。