srs使用的開源c語言網絡協程庫state thread源碼是什么

本篇內容主要講解“srs使用的開源c語言網絡協程庫state thread源碼是什么”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“srs使用的開源c語言網絡協程庫state thread源碼是什么”吧!

state thread是一個開源的c語言網絡協程庫，它在用戶空間實現了協程調度
st最初是由網景（Netscape）公司的MSPR（Netscape Portable Runtime library）項目中剝離出來，后由SGI（Silicon Graphic Inc）和Yahoo!公司（前者是主力）共同開發維護。
2001年發布v1.0以來一直到2009年v1.9穩定版后未再變動

State Threads：回調終結者(必讀)
https://blog.csdn.net/caoshangpa/article/details/79565411

st-1.9.tar.gz 是原版, http://state-threads.sourceforge.net/
state-threads-1.9.1.tar.gz 是srs修改版, https://github.com/ossrs/state-threads

st源碼編譯
tar zxvf st-1.9.tar.gz
cd st-1.9
make linux-debug          // make命令可以查看支持的編譯選項
obj目錄有編譯生成的文件st.h, lib*.so，lib*.a
examples目錄有幾個例子lookupdns，proxy，server

需要的知識點
1 匯編語言(非必需)
2 線程的棧管理(非必需)
3 線程的調度和同步(必須)。線程不同步的測試代碼thread.c
4 setjmp/longjmp的使用(必須)。測試代碼setjmp.c
5 epoll原理和使用(必須)。測試代碼epoll_server.c 和 epoll_client.c

測試代碼以及文檔下載地址
鏈接: https://pan.baidu.com/s/1kQz3S1YIt6zUwMKScrnHaQ
提取碼: pu9z

分析state_thread源碼的目的，是為了正確的使用它
st中thread其實是協程的概念
st_xxx分為 io類 和 延遲類

一些重要的數據結構
_st_vp_t _st_this_vp;     virtual processor 虛擬處理器
_st_thread_t *_st_this_thread;
_st_clist_t  run_q, io_q, zombie_q, thread_q 
_st_thread_t  *idle_thread, *sleep_q

代碼分析
st庫自帶的example業務邏輯較為復雜，有興趣可以看下。
為了簡化問題，編寫了測試代碼st-1.9/examples/st_epoll.c，依據此代碼提出問題分析問題。
st_init()做了什么？
_st_idle_thread_start()做了什么？
st_thread_create()做了什么？
st_thread_exit()做了什么？
st_usleep()做了什么？
主業務邏輯(無限循環)協程是如何調度的？
監聽的文件描述符是如何調度的？
協程如何正常退出？
1 沒有設置終止條件變量(不可以被join)的協程直接return即可退出; 
2 設置了終止條件變量(可以被join)的協程退出時，先把自己加入到zombie_q中，然后通知等待的協程，等待的協程退出后，自己在退出。

協程的join(連接)是什么意思？
1 創建協程a的時候 st_thread_create(handle_cycle, NULL, 1, 0) 要設置為1, 表示該協程可以被join
2 協程b代碼里要掉用st_thread_join(thread, retvalp)，表示我要join到協程a上
3 join的意思是 協程a和協程b 有一定關聯行，在協程退出時，要先退出協程b 才能退出協程a
4 st中一個協程只能被另一個協程join，不能被多個協程join
5 可以被join的協程a，在沒有其他協程join時，協程a無法正常退出

st里的mutex有什么用？
通常情況下st的多協程是不需要加鎖的，但是在有些情況下需要鎖來保證原子操作，下面會詳細說明。
st_mutex_new(void); 創建鎖
st_mutex_destroy(st_mutex_t lock); 等待隊列必須為空才能銷毀鎖
st_mutex_lock(st_mutex_t lock); 第一次掉用能獲得鎖，以后掉用會加入鎖的等待隊列中(FIFO)
st_mutex_unlock(st_mutex_t lock); 釋放鎖并激活等待隊列的協程
st_mutex_trylock(st_mutex_t lock); 嘗試獲得鎖不會加入到等待隊列

st里的cond有什么用？
通常情況下st的多協程是不需要條件變量的，但是有些情況下需要條件變量來保證協程執行的先后順序，比如：協程a要先于協程b執行
st_cond_new(void); 創建條件變量
st_cond_destroy(st_cond_t cvar); 等待隊列必須為空才能銷毀條件變量
st_cond_timedwait(st_cond_t cvar, st_utime_t timeout); 限時等待條件變量，會加入條件變量的等待隊列中(FIFO)，并加入到sleep_q隊列中(可能先于FIFO的順序被調度到)
st_cond_wait(st_cond_t cvar); 阻塞等待條件變量，會加入條件變量的等待隊列中(FIFO)
st_cond_signal(st_cond_t cvar); 喚醒阻塞在條件變量上的一個協程
st_cond_broadcast(st_cond_t cvar); 喚醒阻塞在條件變量上的全部協程

這個圖要配合測試代碼 st-1.9/examples/st_epoll.c

st中與調度有關的函數
st的setjmp
#define _ST_SWITCH_CONTEXT(_thread)   \                協程切換的兩個宏函數之一，停止當前協程并運行其他協程
  ST_BEGIN_MACRO                      \
  ST_SWITCH_OUT_CB(_thread);          \                       協程切走時調用的函數，一般不管用
  if (!MD_SETJMP((_thread)->context)) \                         匯編語言實現 應該跟setjmp()一樣 首次掉用返回0
  {                                   \
    _st_vp_schedule();                \                                    核心調度函數
  }                                   \
  ST_DEBUG_ITERATE_THREADS();         \
  ST_SWITCH_IN_CB(_thread);           \                          協程切回時調用的函數，一般不管用
  ST_END_MACRO

st的longjmp
#define _ST_RESTORE_CONTEXT(_thread) \               協程切換的兩個宏函數之一，恢復線程運行
  ST_BEGIN_MACRO                     \
  _ST_SET_CURRENT_THREAD(_thread);   \                設置全局變量 _st_this_thread = _thread
  MD_LONGJMP((_thread)->context, 1); \                       匯編語言實現 應該跟longjmp()一樣, 返回值永遠為1
  ST_END_MACRO

MD_SETJMP的時候，會使用匯編把所有寄存器的信息保留下來，而MD_LONGJMP則會把所有的寄存器信息重新加載出來。兩者配合使用的時候，可以完成函數間的跳轉。

st的核心調度函數
void _st_vp_schedule(void)
{
  _st_thread_t *thread;
  printf("in _st_vp_schedule\n");
  printf("_st_active_count = %d\n", _st_active_count);
  if (_ST_RUNQ.next != &_ST_RUNQ)
  {
    printf("use runq\n");
    /* Pull thread off of the run queue */
    thread = _ST_THREAD_PTR(_ST_RUNQ.next);
    _ST_DEL_RUNQ(thread);
  }
  else
  {
    printf("use idle\n");
    /* If there are no threads to run, switch to the idle thread */
    thread = _st_this_vp.idle_thread;
  }
  ST_ASSERT(thread->state == _ST_ST_RUNNABLE);
  /* Resume the thread */
  thread->state = _ST_ST_RUNNING;
  _ST_RESTORE_CONTEXT(thread);
}

st輔助調度函數
void *_st_idle_thread_start(void *arg)
{
  printf("i'm in _st_idle_thread_start()\n");
  _st_thread_t *me = _ST_CURRENT_THREAD();

  while (_st_active_count > 0)
  {
    /* Idle vp till I/O is ready or the smallest timeout expired */
    printf("call _st_epoll_dispatch()\n");
    _ST_VP_IDLE();                                                       處理io類事件

    /* Check sleep queue for expired threads */
    _st_vp_check_clock();                                               處理延時類事件
    me->state = _ST_ST_RUNNABLE;
    _ST_SWITCH_CONTEXT(me);                                  從這里恢復運行，然后判斷_st_active_count的值
  }
/* No more threads */
  exit(0);                                                                        整個程序退出
  /* NOTREACHED */
  return NULL;
}

會觸發協程切換的函數有哪些？
sched.c:86: _ST_SWITCH_CONTEXT(me); 59 int st_poll(struct pollfd *pds, int npds, st_utime_t timeout)
sched.c:234: _ST_SWITCH_CONTEXT(me); 221 void *_st_idle_thread_start(void *arg)
sched.c:261: _ST_SWITCH_CONTEXT(thread); 244 void st_thread_exit(void *retval)
sched.c:276: _ST_SWITCH_CONTEXT(thread); 244 void st_thread_exit(void *retval)
sync.c:131: _ST_SWITCH_CONTEXT(me); 115 int st_usleep(st_utime_t usecs)
sync.c:198: _ST_SWITCH_CONTEXT(me); 180 int st_cond_timedwait(_st_cond_t *cvar, st_utime_t timeout)
sync.c:315: _ST_SWITCH_CONTEXT(me); 290 int st_mutex_lock(_st_mutex_t *lock)

sched.c:134: _ST_RESTORE_CONTEXT(thread); 115 void _st_vp_schedule(void)

st中的interrupt
顯示調用void st_thread_interrupt(_st_thread_t *thread)會對協程設置interrupt狀態，interrupt狀態會中斷協程的本次運行(可能是個循環任務)，是否導致協程退出，要看協程內部對interrupt返回值的處理。下面以st_usleep()函數為例進行說明。
[ykMac:st-1.9]# grep -nr "_ST_FL_INTERRUPT" *
common.h:311:#define _ST_FL_INTERRUPT 0x08        interrupt的宏定義
sched.c:68:  if (me->flags & _ST_FL_INTERRUPT)         59 int st_poll() ,調用函數時，判斷是否設置interrupt
sched.c:70:    me->flags &= ~_ST_FL_INTERRUPT;       如果設置就退出，退出前對interrupt取反
sched.c:107:  if (me->flags & _ST_FL_INTERRUPT)       59 int st_poll(),變為運行協程時，判斷是否設置interrupt
sched.c:109:    me->flags &= ~_ST_FL_INTERRUPT;     如果設置就退出，退出前對interrupt取反
sched.c:551:  thread->flags |= _ST_FL_INTERRUPT;     在545 void st_thread_interrupt()中設置為interrupt
sync.c:119:  if (me->flags & _ST_FL_INTERRUPT) {       115 int st_usleep(st_utime_t usecs),調用函數時
sync.c:120:    me->flags &= ~_ST_FL_INTERRUPT;
sync.c:133:  if (me->flags & _ST_FL_INTERRUPT) {       115 int st_usleep(st_utime_t usecs),變為運行協程時
sync.c:134:    me->flags &= ~_ST_FL_INTERRUPT;
sync.c:185:  if (me->flags & _ST_FL_INTERRUPT) {       180 int st_cond_timedwait(),調用函數時
sync.c:186:    me->flags &= ~_ST_FL_INTERRUPT;
sync.c:208:  if (me->flags & _ST_FL_INTERRUPT) {       180 int st_cond_timedwait(),變為運行協程時
sync.c:209:    me->flags &= ~_ST_FL_INTERRUPT;
sync.c:294:  if (me->flags & _ST_FL_INTERRUPT) {       290 int st_mutex_lock(),調用函數時
sync.c:295:    me->flags &= ~_ST_FL_INTERRUPT;
sync.c:319:  if ((me->flags & _ST_FL_INTERRUPT) && lock->owner != me) {       290 int st_mutex_lock(),變運行時
sync.c:320:    me->flags &= ~_ST_FL_INTERRUPT;

115 int st_usleep(st_utime_t usecs)
116 {
117   _st_thread_t *me = _ST_CURRENT_THREAD();
118
119   if (me->flags & _ST_FL_INTERRUPT) {
120     me->flags &= ~_ST_FL_INTERRUPT;     退出前對interrupt取反
121     errno = EINTR;
122     return -1;                                                如果不對errno或返回值做處理，循環還是會繼續的
123   }
124
125   if (usecs != ST_UTIME_NO_TIMEOUT) {
126     me->state = _ST_ST_SLEEPING;
127     _ST_ADD_SLEEPQ(me, usecs);
128   } else
129     me->state = _ST_ST_SUSPENDED;
130
131   _ST_SWITCH_CONTEXT(me);
132
133   if (me->flags & _ST_FL_INTERRUPT) {
134     me->flags &= ~_ST_FL_INTERRUPT;
135     errno = EINTR;
136     return -1;
137   }
138
139   return 0;
140 }

st的優缺點
優點：
1 用戶空間實現協程調度，降低了用戶空間和內核空間的切換，一定程度上提高了程序效率。
2 由于是在單核上的單線程多協程，同一時間只會有一個協程在運行，所以對于全局變量也不需要做協程同步。
   共享資源釋放函數只需做到可重入就行，所謂的可重入就是釋放之前先判斷是否為空值，釋放后要賦空值。
3 協程使用完，直接return即可，st會回收協程資源并做協程切換。
4 可以通過向run_q鏈表頭部加入協程，來實現優先調度。
5 st支持多個操作系統，比如 AIX,CYGWIN,DARWIN,FREEBSD,HPUX,IRIX,LINUX,NETBSD,OPENBSD,SOLARIS
缺點：
1 所有I/O操作必須使用st提供的API，只有這樣協程才能被調度器管理。
2 所有協程里不能使用sleep()，sleep()會造成整個線程sleep。
3 被調度到的協程不會被限制運行時長，如果有協程是cpu密集型或死循環，就會嚴重阻礙其他協程運行。
4 單進程單線程，只能使用單核，想要通過多個cpu提高并發能力，只能開多個程序(進程)，多進程通信較麻煩。

補充知識點
1 線程為什么要同步？
線程由內核自動調度
同一個進程上的線程共享該進程的整個虛擬地址空間
同一個進程上的線程代碼區是共享的，即不同的線程可以執行同樣的函數
所以在并發環境中，多個線程同時對同一個內存地址進行寫入，由于CPU寄存器時間調度上的問題，寫入數據會被多次的覆蓋，會造成共享數據損壞，所以就要使線程同步。

2 什么情況下需要線程同步？
線程同步指的是 不同時發生，就是線程要排隊
1 多核，單進程多線程，不同線程會對全局變量讀寫，這種情況才需要對線程做同步控制
2 單核，單進程多線程，不同線程會對全局變量讀寫，這種情況不需要對線程做同步控制
3 多核，單進程多線程，不同線程不對全局變量讀寫，這種情況不需要對線程做同步控制
4 多核，單進程多線程，不同線程會對全局變量讀，這種情況不需要對線程做同步控制

問題：對于第2條，這個應該是有點兒片面，有依賴有優先級搶占也是要同步，除非像abcde這種幾個線程干一摸一樣的事情，而且項目之間不依賴。
有依賴 可以理解為 生產消費關系，雖然是 單核 單進程 多線程 但是同一時間只能有一個線程在運行，也就是說 生產和消費不會同時發生，同樣多個生產也不會同時發生，所以不需要鎖。
線程是有優先級控制，但是不管怎么控制，只要保證同一時間只能有一個線程在運行，就不需要鎖了。
問題：對于第2條，這個應該是有點兒片面，有原子操作且原子操作過程中有線程切換，這種是需要鎖的。
比如，線程a 第一次讀取全局變量x并做處理，然后發生線程切換(線程由內和自動調度)后切回，然后第二次讀取全局變量x并做處理，我們想確保兩次讀取
x值相同，但是發生了線程切換x值可能被改變。
如何 確保 第一次讀取并處理和第二次讀取并處理是原子操作呢？ 使用st_mutex_t

3 accept()序列化
亦稱驚群效應，亦亦稱Zeeg難題
https://uwsgi-docs-zh.readthedocs.io/zh_CN/latest/articles/SerializingAccept.html
在多次fork自己之后，每個進程一般將會開始阻塞在 accept() 上
每當socket上嘗試進行一個連接，阻塞在 accept() 上的每個進程的 accept() 都會被喚醒。
只有其中一個進程能夠真正接收到這個連接，而剩余的進程將會獲得一個無聊的 EAGAIN 這導致了大量的CPU周期浪費，實際解決方法是把一個鎖放在 accept() 調用之前，來序列化它的使用

4 Internet Applications網絡程序架構
多進程架構 Multi-Process
    一個進程服務一個連接，要解決數據共享問題
單進程多線程架構 Multi-Threaded 
    一個線程服務一個連接，要解決數據同步問題
事件驅動的狀態機架構 Event-Driven State Machine 
    事件觸發回調函數(缺點是嵌套) 或 用戶空間實現協程調度
實際上 EDSM架構 用很復雜的方式模擬了多線程
st提供的就是EDSM機制，它在用戶空間實現協程調度

到此，相信大家對“srs使用的開源c語言網絡協程庫state thread源碼是什么”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！