state thread

 

State Thread 的官网地址：http://state-threads.sourceforge.net

The State Threads Library is a small application library which provides a foundation for writing fast and highly scalable Internet applications (such as web servers, proxy servers, mail transfer agents, and so on, really any network-data-driven application) on UNIX-like platforms. It combines the simplicity of the multithreaded programming paradigm, in which one thread supports each simultaneous connection, with the performance and scalability of an event-driven state machine architecture. In other words, this library offers a threading API for structuring an Internet application as a state machine.

传统的服务器端开发模型是：采用session异步方式调用来（调用其它好多服务时须要存状态，来完成一个完整的流程调用），这种方式稍稍复杂一点。采用”协程”的好处，经常被人提起的是：回归“同步调用型”的开发模型：程序调用api1()，api1返回以后调api2()，等api2返回以后调api3.. and so on. 这样的开发模式更加容易理解。

1、setjmp/longjmp

ST中使用了setjmp、longjmp来实现协程，内容参考另外一篇博文：http://www.cnblogs.com/ym65536/p/4984339.html

2、state threads

本文的分析主要基于state threads实现进行分析。为求简单，忽略掉一些平台强相关的逻辑。

2.1 事件系统（_st_eventsys_t）

常见的方式：poll/select/epoll/kqueue等等（默认采用select）。实现方式和其它的一些事件系统（ae/libevent）有点类型，因此此文再也不详细地作剖析。

 1 typedef struct _st_eventsys_ops {
 2  const char *    name;     /* Name of this event system */
 3  int    val;               /* Type of this event system */
 4  int    (*init)(void);     /* Initialization */
 5  void   (*dispatch)(void); /* Dispatch function */
 6  int    (*pollset_add)(struct pollfd *, int);     /* Add descriptor set */
 7  void   (*pollset_del)(struct pollfd *, int);    /* Delete descriptor set */
 8  int    (*fd_new)(int);       /* New descriptor allocated */
 9  int    (*fd_close)(int);     /* Descriptor closed */
10  int    (*fd_getlimit)(void); /* Descriptor hard limit */
11 } _st_eventsys_t;

 

2.2 协程态(_st_vp_t)

每一个vp对应一个idle协程，4个队列，切换回调等。

 1 typedef struct _st_vp {
 2   _st_thread_t *idle_thread;  /* Idle thread for this vp */
 3   st_utime_t last_clock;      /* The last time we went into vp_check_clock() */
 4 
 5   _st_clist_t run_q;          /* run queue for this vp */
 6   _st_clist_t io_q;           /* io queue for this vp */
 7   _st_clist_t zombie_q;       /* zombie queue for this vp */
 8 #ifdef DEBUG
 9   _st_clist_t thread_q;       /* all threads of this vp */
10 #endif
11   int pagesize;
12 
13   _st_thread_t *sleep_q;      /* sleep queue for this vp */
14   int sleepq_size;          /* number of threads on sleep queue */
15 
16 #ifdef ST_SWITCH_CB
17   st_switch_cb_t switch_out_cb;    /* called when a thread is switched out */
18   st_switch_cb_t switch_in_cb;    /* called when a thread is switched in */
19 #endif
20 } _st_vp_t;

 

2.3 队列(run/io/zombie/sleep queue)

st使用了4个队列, 这个在上面的结构中就已经有，为了访问的方便，定义了4组宏(debug模式下，还有另外一个)，以下。这4个队列分别对应了4种状态，由此造成本身的状态机体系。

每当建立一个微线程时，都会将其加入到RUNQ中。

1 #define _ST_RUNQ    (_st_this_vp.run_q)
2 #define _ST_IOQ     (_st_this_vp.io_q)
3 #define _ST_ZOMBIEQ (_st_this_vp.zombie_q)
4 #ifdef DEBUG
5 #define _ST_THREADQ (_st_this_vp.thread_q)
6 #endif
7 #define _ST_SLEEPQ  (_st_this_vp.sleep_q)

 

2.4 微线程（_st_thread_t）

在vp中，咱们看到有一个成员：idle_thread. 天然会有疑问：这货是干吗的？这货长的像个线程，但其实固然不是线程，就以“微线程”来称吧：即用户态下实现的线程。

主要关注点：

1. 微线程的状态机
2. 每一个微线程维护的栈空间、私有数据
3. 每一个微线程的入口点
4. 咱们熟悉的jmp_buf上下文：context

 1 typedef struct _st_thread _st_thread_t;
 2  
 3 struct _st_thread {
 4  int     state; /* Thread's state */
 5  int     flags; /* Thread's flags */
 6  
 7  void *   (*start)(void *arg); /* The start function of the thread */
 8  void *   arg;                 /* Argument of the start function */
 9  void *   retval;             /* Return value of the start function */
10  
11 _st_stack_t *   stack;       /* Info about thread's stack */
12  
13 _st_clist_t     links;       /* For putting on run/sleep/zombie queue */
14  _st_clist_t    wait_links;  /* For putting on mutex/condvar wait queue */
15 #ifdef DEBUG
16  _st_clist_t    tlink;       /* For putting on thread queue */
17 #endif
18  
19  st_utime_t     due;      /* Wakeup time when thread is sleeping */
20  _st_thread_t * left;     /* For putting in timeout heap */
21  _st_thread_t * right;    /* -- see docs/timeout_heap.txt for details */
22  int            heap_index;
23  
24 void **         private_data; /* Per thread private data */
25  
26 _st_cond_t *    term;       /* Termination condition variable for join */
27  
28 jmp_buf         context;    /* Thread's context */
29 };

 

2.5 栈（_st_stack_t）

每一个微线程自身单独都会维护本身的一个“栈空间”。栈有本身独立的内存，大小，栈底，栈顶，栈指针（程序级别、非系统内存级别），恢复点。这里的“栈”只是一个概念（实际是堆，经过malloc或mmap来实现），并不是咱们一般指的栈。

typedef struct _st_stack {
_st_clist_t     links;
char *          vaddr;              /* Base of stack's allocated memory */
int             vaddr_size;         /* Size of stack's allocated memory */
int             stk_size;           /* Size of usable portion of the stack */
char *          stk_bottom;         /* Lowest address of stack's usable portion */
char *          stk_top;            /* Highest address of stack's usable portion */
void *          sp;                 /* Stack pointer from C's point of view */
#ifdef __ia64__
void *          bsp;                /* Register stack backing store pointer */
#endif
} _st_stack_t;

 

_st_stack的可用空间大小缺省为：128k，即刚开始分配时的大小。当不够用时，会增长一个内存页的大小(getpagesize)，固然的大小必须是内存页的整数倍。(详见st_thread_create)。这部分是可用栈空间的大小，实际大小还会加上：2*REDZONE+extra。 在stack的每一端（栈顶、栈底）都有一个REDZONE，其大小也是一个分页；而extra则看是否须要（0或者一个分页大小）

在st_stack中，入栈的方式，有两种：向下增加，向上增长。以向上增加为例（向下增加的话，相似，反之便可）：

2.6 jmp_buf上下文

即然栈是本身实现的，那边对应的jmp_buf里头的sp、bsp也须要跟着变。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

... _ST_INIT_CONTEXT( thread , stack->sp, stack->bsp, _st_thread_main); ... #define MD_SETJMP(env)           _setjmp(env) #define MD_LONGJMP(env, val)     _longjmp(env, val) ... #define MD_INIT_CONTEXT(_thread, _sp, _bsp, _main) \ ST_BEGIN_MACRO \ if (MD_SETJMP((_thread)->context)) \      _main(); \ memcpy (( char *)(_bsp) - MD_STACK_PAD_SIZE, \          ( char *)(_thread)->context[0].__jmpbuf[17] - MD_STACK_PAD_SIZE, \          MD_STACK_PAD_SIZE); \ (_thread)->context[0].__jmpbuf[0] = ( long ) (_sp); \ (_thread)->context[0].__jmpbuf[17] = ( long ) (_bsp); \ ST_END_MACRO

3 微线程的建立

3.1 总体流程

微线程的创立，经过_st_thread_create()来完成。

主要过程以下：

1. 建立栈，分配空间。
2. 初始化sp/bsp、私有数据（private_data）以及微线程自身所需空间(thread)，入口函数(start)及参数。
3. 调用setjmp，若成功，则执行_st_thread_main()；
4. 将以前的sp恢复.

其中的第三步，_st_thread_main，也就是微线程自身的执行点，过程也很简单：

1. 获取当前微线程
2. 将微线程对应的栈空间位置状态变量置为0（state和flags，转型为valatile，防止longjmp切换时更改值。ps: 这里有个关于二级指针数组[1]的tricks）
3. 执行微线程的入口函数(start)
4. 退出微线程（_st_thread_exit）

退出微线程时，会发生一些很奇妙的事情：

1. 清理工做
2. 对term作检查，是否已退出。如果，添加thread到僵尸队列：zombieq中, 通知term信号，并切换context（_ST_SWITCH_CONTEXT，清理term
3. 清理stack（_st_stack_free）
4. 切换至另外一个thread执行(_ST_SWITCH_CONTEXT)

上面就是一个总体的流程。而其中有两个地方，须要特别再关注一下。

3.2 上下文切换

由于涉及到微线程的切换，即然是“切换”，那么这里确定至少涉及到两个微线程。因为微线程自己处在RUNQ中，切换的话，天然会将next。

1. 若是runq队列有在运行，切换至next; 若是没有，切换至idle_thread
2. 将切换后的当前微线程置为run状态
3. longjmp置相应微线程的jmp_buf处，恢复栈执行

 

4.小结

state thread的代码很简洁，若是知道原理的话，分析起来不困难。另外，它还简单地实现了一个http server。有兴趣的朋友能够作为参照本身实现功能更复杂的http server。