SRS开源直播服务 - StateThreads微线程框架学习

typedef struct _st_vp {
  _st_thread_t *idle_thread;  /* Idle thread for this vp */
  st_utime_t last_clock;      /* The last time we went into vp_check_clock() */

  _st_clist_t run_q;          /* run queue for this vp */
  _st_clist_t io_q;           /* io queue for this vp */
  _st_clist_t zombie_q;       /* zombie queue for this vp */
#ifdef DEBUG
  _st_clist_t thread_q;       /* all threads of this vp */
#endif
  int pagesize;

  _st_thread_t *sleep_q;      /* sleep queue for this vp */
  int sleepq_size;	          /* number of threads on sleep queue */

#ifdef ST_SWITCH_CB
  st_switch_cb_t switch_out_cb;	/* called when a thread is switched out */
  st_switch_cb_t switch_in_cb;	/* called when a thread is switched in */
#endif
} _st_vp_t;

typedef struct _st_thread {
  int state;                  /* Thread's state */
  int flags;                  /* Thread's flags */

  void *(*start)(void *arg);  /* The start function of the thread */
  void *arg;                  /* Argument of the start function */
  void *retval;               /* Return value of the start function */

  _st_stack_t *stack;	        /* Info about thread's stack */

  _st_clist_t links;          /* For putting on run/sleep/zombie queue */
  _st_clist_t wait_links;     /* For putting on mutex/condvar wait queue */

  st_utime_t due;             /* Wakeup time when thread is sleeping */
  _st_thread_t *left;         /* For putting in timeout heap */
  _st_thread_t *right;	      /* -- see docs/timeout_heap.txt for details */
  int heap_index;

  void **private_data;        /* Per thread private data */

  _st_cond_t *term;           /* Termination condition variable for join */

  jmp_buf context;            /* Thread's context */
} _st_thread_t;

// 微线程状态定义
#define _ST_ST_RUNNING      0      // 执行中 
#define _ST_ST_RUNNABLE     1      // 可执行状态，等待调度
#define _ST_ST_IO_WAIT      2      // 等待IO事件
#define _ST_ST_LOCK_WAIT    3      // 等待互斥锁
#define _ST_ST_COND_WAIT    4      // 等待条件变量
#define _ST_ST_SLEEPING     5      // sleep
#define _ST_ST_ZOMBIE       6      // 微线程已结束，待其他微线程调用st_thread_join收尸
#define _ST_ST_SUSPENDED    7      // 暂停，只能调用st_thread_interrupt唤醒

// 微线程flag定义
#define _ST_FL_PRIMORDIAL   0x01    // 原生微线程，即不是创建的微线程，没有分配私有栈资源
#define _ST_FL_IDLE_THREAD  0x02    // 空闲处理微线程，用于调用epoll，处理定时器
#define _ST_FL_ON_SLEEPQ    0x04    // 微线程在sleep队列中，需要定时器的情况：调用st_usleep、st_cond_timedwait、st_poll等待IO事件等
#define _ST_FL_INTERRUPT    0x08    // 微线程被调用st_thread_interrupt()中断
#define _ST_FL_TIMEDOUT     0x10    // 微线程定时器超时

// idle微线程执行的start函数
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 */
	// 调用IO多路复用函数（epoll，select等）等待IO事件发生，并处理所有发生的IO事件，
	// 如果微线程等待的IO事件发生，会将微线程从io等待队列st_vp_t.io_q中移除
	// 并加入到runable队列st_vp_t.run_q
        _ST_VP_IDLE();

        /* Check sleep queue for expired threads */
	// 检查微线程设置的定时器是否超时，针对超时的微线程，将其从sleep队列移除
	// 并加入到runable队列
        _st_vp_check_clock();

	// 交出运行权，并从runable队列st_vp_t.run_q中调度下一个微线程运行
        me->state = _ST_ST_RUNNABLE;
        _ST_SWITCH_CONTEXT(me);
    }

    /* No more threads */
    exit(0);

    /* NOTREACHED */
    return NULL;
}

// 框架初始化函数
int st_init(void)
{
    _st_thread_t *thread;

    if (_st_active_count) {
        /* Already initialized */
        return 0;
    }

    // 选择使用哪个IO多路复用函数，如epoll、select等
    st_set_eventsys(ST_EVENTSYS_DEFAULT);

    // 屏蔽SIGPIPE信号 并 初始化IO多路复用函数相关数据结构
    if (_st_io_init() < 0)
        return -1;

    memset(&_st_this_vp, 0, sizeof(_st_vp_t));

    // 清空runable队列、io等待队列、zombie队列
    ST_INIT_CLIST(&_ST_RUNQ);
    ST_INIT_CLIST(&_ST_IOQ);
    ST_INIT_CLIST(&_ST_ZOMBIEQ);

    // 调用IO多路复用函数对应的初始化函数
    if ((*_st_eventsys->init)() < 0)
        return -1;

    _st_this_vp.pagesize = getpagesize();
    _st_this_vp.last_clock = st_utime();

    // 创建idle微线程，idle微线程主要用于调用epoll等待IO事件和处理定时器
    _st_this_vp.idle_thread = st_thread_create(_st_idle_thread_start,
        NULL, 0, 0);
    if (!_st_this_vp.idle_thread)
        return -1;
    _st_this_vp.idle_thread->flags = _ST_FL_IDLE_THREAD;
    _st_active_count--;
    // 将idle微线程从runable队列移除，由于idle微线程只在没有微线程可以运行时，才会主动调度，
    // 所以不需要加入到run队列
    _ST_DEL_RUNQ(_st_this_vp.idle_thread);

    // 初始化primordial微线程，primordial微线程用来标记系统进程，由于可以直接使用
    // 系统进程的栈空间，故只需要为primordial微线程分配st_thread_t和私有key数据区
    thread = (_st_thread_t *)calloc(1, sizeof(_st_thread_t) +
        (ST_KEYS_MAX * sizeof(void *)));
    if (!thread)
        return -1;
    thread->private_data = (void **)(thread + 1);
    thread->state = _ST_ST_RUNNING;
    thread->flags = _ST_FL_PRIMORDIAL;
    _ST_SET_CURRENT_THREAD(thread);
    _st_active_count++;

    // 当前运行的微线程是primordial微线程，当primordial微线程退出时，整个进程也会终止
    return 0;
}

// 创建微线程函数
_st_thread_t *st_thread_create(void *(*start)(void *arg), void *arg,
	                       int joinable, int stk_size)
{
	_st_thread_t *thread;
	_st_stack_t *stack;
	void **ptds;
	char *sp;

	if (stk_size == 0)
		stk_size = ST_DEFAULT_STACK_SIZE;  // 默认栈空间为64K
	// 将栈空间调整为PAGE_SIZE的整数倍
	stk_size = ((stk_size + _ST_PAGE_SIZE - 1) / _ST_PAGE_SIZE) * _ST_PAGE_SIZE;
	// 分配微线程使用的私有数据空间：包括栈空间，私有key数据区，st_thread_t结构体等
	// 为了避免频繁调用系统分配函数分配微线程栈空间，框架中保存了stack空闲链表，
	// 会优先从空闲链表获取，当空闲链表为空时，会从系统分配
	stack = _st_stack_new(stk_size);
	if (!stack)
		return NULL;

	// 初始化分配好的空间
	sp = stack->stk_top;
	sp = sp - (ST_KEYS_MAX * sizeof(void *));
	ptds = (void **)sp;
	sp = sp - sizeof(_st_thread_t);
	thread = (_st_thread_t *)sp;

	// 保证栈可用位置一定是64字节对齐的
	if ((unsigned long)sp & 0x3f)
		sp = sp - ((unsigned long)sp & 0x3f);
	stack->sp = sp - _ST_STACK_PAD_SIZE;

	// 分配的空间重置为0，因为stack可能是从空闲链表中获取的
	memset(thread, 0, sizeof(_st_thread_t));
	memset(ptds, 0, ST_KEYS_MAX * sizeof(void *));

	thread->private_data = ptds;
	thread->stack = stack;
	thread->start = start;
	thread->arg = arg;

	// 调用setjmp保存当前的上下文信息到jmp_buf中，主要功能是初始化jmp_buf
	// 因为现在保存到jmp_buf中的信息不会在微线程的后续运行中使用到
	if (setjmp(thread->context))
		// 当微线程首次被调度运行时，会进入该分支，并调用_st_thread_main函数执行
		_st_thread_main();
	// 非常重要的一点：设置jmp_buf中的rsp指向为微线程分配的私有栈空间的可用位置
	// 这样可以保证框架中不同微线程运行在自己的私有栈空间上
	thread->context[JB_RSP] = (long)(stack->sp);

	// 如果微线程是可joinable的，就创建一个微线程终止时使用的条件变量
	// 当该微线程终止时会等待在该条件变量上，当其他微线程调用st_thread_join函数
	// 替该微线程收尸后，该微线程才会真正终止
	if (joinable) {
		thread->term = st_cond_new();
		if (thread->term == NULL) {
			_st_stack_free(thread->stack);
			return NULL;
		}
	}

	// 设置为runable状态，并 加入到runable队列中
	thread->state = _ST_ST_RUNNABLE;
	_st_active_count++;
	_ST_ADD_RUNQ(thread);

	return thread;
}

// 微线程启动后调用的main函数
void _st_thread_main(void)
{
	_st_thread_t *thread = _ST_CURRENT_THREAD();

	/*
	* Cap the stack by zeroing out the saved return address register
	* value. This allows some debugging/profiling tools to know when
	* to stop unwinding the stack. It's a no-op on most platforms.
	*/
	MD_CAP_STACK(&thread);

	// 调用微线程设置的start函数
	thread->retval = (*thread->start)(thread->arg);

	// 清理微线程使用的资源
	st_thread_exit(thread->retval);
}

// 切换_thread微线程出去，并调度下一个runable状态的微线程运行
#define _ST_SWITCH_CONTEXT(_thread)       \
    {                                     \
        if (!setjmp(_thread->context)) {  \
            _st_vp_schedule();            \
        }                                 \
    }

// 微线程调度逻辑，即调度下一个runable状态微线程运行
void _st_vp_schedule(void)
{
	_st_thread_t *thread;

	// 如果runable队列_st_this_vp.run_q非空，就选队列首的微线程
	// 否则调度idle微线程运行
	if (_ST_RUNQ.next != &_ST_RUNQ) {
		thread = _ST_THREAD_PTR(_ST_RUNQ.next);
		_ST_DEL_RUNQ(thread);
	}
	else {
		thread = _st_this_vp.idle_thread;
	}

	// 设置选中的微线程的状态为running状态，并调用longjmp函数跳转到选中的微线程，
        // 选中的微线程从最后一次的setjmp位置继续运行
	thread->state = _ST_ST_RUNNING;
	_ST_SET_CURRENT_THREAD(thread);
	longjmp(thread->context, 1)
}

// 微线程终止时调用的函数
void st_thread_exit(void *retval)
{
	_st_thread_t *thread = _ST_CURRENT_THREAD();

	thread->retval = retval;

	// 释放微线程运行期间调用st_thread_setspecific设置的私有key数据
	_st_thread_cleanup(thread);

	_st_active_count--;

	// 如果创建了term条件变量，需要通知调用st_thread_join()等待该微线程的微线程为该
	// 微线程“收尸”
	if (thread->term) {
		// 添加到zombie队列
		thread->state = _ST_ST_ZOMBIE;
		_ST_ADD_ZOMBIEQ(thread);

		// 通知等待在term条件变量上的微线程
		st_cond_signal(thread->term);

		// 交出控制权，等到为本线程收尸的微线程调用st_thread_join()返回
		// 后，本微线程才会switch回来，并恢复运行
		_ST_SWITCH_CONTEXT(thread);

		// 清理条件变量
		st_cond_destroy(thread->term);
		thread->term = NULL;
	}

	// 如果终止的不是Primordial微线程，就释放为微线程分配的私有栈空间，
	// 释放的栈空间会放到空闲链表中
	if (!(thread->flags & _ST_FL_PRIMORDIAL))
		_st_stack_free(thread->stack);

	// 交出控制权，并调度下一个runable状态的微线程，微线程生命周期终止
	_ST_SWITCH_CONTEXT(thread);
	/* Not going to land here */
}

// 调用epoll等待IO事件的函数
void _st_epoll_dispatch(void)
{
    st_utime_t min_timeout;
    _st_clist_t *q;
    _st_pollq_t *pq;
    struct pollfd *pds, *epds;
    struct epoll_event ev;
    int timeout, nfd, i, osfd, notify;
    int events, op;
    short revents;

    // sleep队列为空时，设置永不超时，否则设置为sleep队列中最早超时微线程的超时时间
    if (_ST_SLEEPQ == NULL) {
        timeout = -1;
    } else {
        min_timeout = (_ST_SLEEPQ->due <= _ST_LAST_CLOCK) ? 0 :
            (_ST_SLEEPQ->due - _ST_LAST_CLOCK);
        timeout = (int)(min_timeout / 1000);
    }

    // 调用epoll_wait函数等待IO事件发生 或者 超时 发生
    nfd = epoll_wait(_st_epoll_data->epfd, _st_epoll_data->evtlist,
        _st_epoll_data->evtlist_size, timeout);

    if (nfd > 0) {
        // 遍历发生的事件，把发生的IO事件设置到fd对应的_epoll_fd_data_t.revents中
        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);
            }
        }

        // 遍历IO等待队列_st_vp_t.io_q中的每个元素
        for (q = _ST_IOQ.next; q != &_ST_IOQ; q = q->next) {
            pq = _ST_POLLQUEUE_PTR(q);
            notify = 0;
            // 关注fd数组的结束位置
            epds = pq->pds + pq->npds;
            // 遍历关注的fd数组，查看对应fd上是否有IO事件发生
            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;
                // 关注pri事件，并检查是否发生了pri事件
                if ((events & POLLPRI) && (_ST_EPOLL_REVENTS(osfd) & EPOLLPRI))
                    revents |= POLLPRI;
                // 关注错误事件，并检查是否发生了错误事件
                if (_ST_EPOLL_REVENTS(osfd) & EPOLLERR)
                    revents |= POLLERR;
                // 始终检查是否发生了hup事件
                if (_ST_EPOLL_REVENTS(osfd) & EPOLLHUP)
                    revents |= POLLHUP;

                pds->revents = revents;
                if (revents) {
                    notify = 1;
                }
            }

            if (notify) {
                // 如果有IO事件发生，就从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);

                // 修改微线程状态 并 加入到runable队列，等待调度运行
                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;
            // 重置revents为空
            _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;
            // 修改fd在epoll中关注的IO事件集合
            if (epoll_ctl(_st_epoll_data->epfd, op, osfd, &ev) == 0 &&
                op == EPOLL_CTL_DEL) {
                _st_epoll_data->evtlist_cnt--;
            }
        }
    }
}

// 从tcp连接套接字中读数据
ssize_t st_read(_st_netfd_t *fd, void *buf, size_t nbyte, st_utime_t timeout)
{
    ssize_t n;
    // 先尝试读，如果没有读到数据，才调用st_netfd_poll等待套接字变为可读
    while ((n = read(fd->osfd, buf, nbyte)) < 0) {
        if (SocketErrno == PORT_IO_EINTR)
            continue;
        if (!_IO_NOT_READY_ERROR)
            return -1;
        
        if (st_netfd_poll(fd, POLLIN, timeout) < 0)
            return -1;
    }
    return n;
}

// 等待单个套接字fd上的IO事件
int st_netfd_poll(_st_netfd_t *fd, int how, st_utime_t timeout)
{
    struct pollfd pd;
    int n;

    pd.fd = fd->osfd;
    pd.events = (short)how;
    pd.revents = 0;

    if ((n = st_poll(&pd, 1, timeout)) < 0)
        return -1;
    if (n == 0) {
        /* Timed out */
        errno = ETIME;
        return -1;
    }
    if (pd.revents & POLLNVAL) {
        errno = EBADF;
        return -1;
    }

    return 0;
}

// 允许同时等待多个套接字的IO事件
int st_poll(struct pollfd *pds, int npds, st_utime_t timeout)
{
    struct pollfd *pd;
    struct pollfd *epd = pds + npds;
    _st_pollq_t pq;
    _st_thread_t *me = _ST_CURRENT_THREAD();
    int n;

    if (me->flags & _ST_FL_INTERRUPT) {
        me->flags &= ~_ST_FL_INTERRUPT;
        errno = EINTR;
        return -1;
    }

    // 将数组中fd关注的IO事件添加到epoll中
    if ((*_st_eventsys->pollset_add)(pds, npds) < 0)
        return -1;

    pq.pds = pds;
    pq.npds = npds;
    pq.thread = me;
    pq.on_ioq = 1;
    // 将调用微线程添加到IO等待队列中
    _ST_ADD_IOQ(pq);
    // 如果设置了超时时间，就加入到sleep队列中
    if (timeout != ST_UTIME_NO_TIMEOUT)
        _ST_ADD_SLEEPQ(me, timeout);
    me->state = _ST_ST_IO_WAIT;
    // 交出执行权，并调度其他微线程运行，等到关注的任何一个fd上发生了关注的IO事件时，
    // 会从这里返回
    _ST_SWITCH_CONTEXT(me);

    n = 0;
    if (pq.on_ioq) {
        // 超时情况，从IO等待队列中就移除
        _ST_DEL_IOQ(pq);
        (*_st_eventsys->pollset_del)(pds, npds);
    } else {
        // 计算发生IO事件的fd数，这里没有把发生IO事件的fd组织成一个数组
        for (pd = pds; pd < epd; pd++) {
            if (pd->revents)
                n++;
        }
    }

    if (me->flags & _ST_FL_INTERRUPT) {
        me->flags &= ~_ST_FL_INTERRUPT;
        errno = EINTR;
        return -1;
    }

    return n;
}

// StateThreads条件变量相关函数
int st_cond_wait(_st_cond_t *cvar);
int st_cond_timedwait(_st_cond_t *cvar, st_utime_t timeout);
int st_cond_signal(st_cond_t cvar);

// linux条件变量相关系统调用
int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
int pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex, struct timespec *timeout);