最近在看SRS的原始碼，SRS是基于協程開發的，底層使用了StateThreads，所以為了充分的理解SRS原始碼，需要先學習一下StateThreads，這里對StateThreads的學習做了一些總結和記錄，
StateThreads是什么
StateThreads是一個用戶級執行緒庫，用于多執行緒編程，它提供了一種輕量級的執行緒模型，允許開發人員以更簡單的方式撰寫并發程式，
StateThreads有什么用
StateThreads 的主要目標是提供一種高效的用戶級執行緒模型，以減少執行緒切換和背景關系切換的開銷，它采用協作式調度策略，即執行緒在主動釋放執行權之前不會被搶占，這種方式可以減少執行緒切換的開銷，但也需要開發人員在適當的時機主動釋放執行權，以避免長時間的阻塞導致程式回應性下降，
StateThreads 提供了一組簡單的函式和宏，用于創建和管理執行緒、同步和通信等操作，它支持執行緒的創建、銷毀、休眠、喚醒等基本操作，以及互斥鎖、條件變數、信號量等同步機制，開發人員可以使用這些函式和宏來撰寫并發程式，而不需要直接操作作業系統提供的執行緒和同步原語，
總的來說，StateThreads是一個高性能、高并發、高擴展性和可讀性的網路服務器架構，
StateThreads怎么用
下載
git clone -b srs https://github.com/ossrs/state-threads.git
編譯
make linux-debug
編譯完成后，將頭檔案匯入需要使用到StateThreads的專案，并在編譯專案時鏈接st庫即可，
使用示例
示例一
下面是用StateThreads實作的一個簡單的服務，可以監聽客戶端的連接，
#include <iostream>
#include <stdio.h>
#include <arpa/inet.h>
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <st.h>
#define LISTEN_PORT 8000
#define ERR_EXIT(m) \
do \
{ \
perror(m); \
exit(-1); \
} while (0)
void *client_thread(void *arg)
{
st_netfd_t client_st_fd = (st_netfd_t)arg;
// 用于獲取與 st_netfd_t 物件關聯的檔案描述符（File Descriptor），它回傳一個整數值，表示檔案描述符的值，
// 將 st_netfd_t 物件轉換為普通的檔案描述符
int client_fd = st_netfd_fileno(client_st_fd);
sockaddr_in client_addr;
socklen_t client_addr_len = sizeof(client_addr);
// 獲取與套接字連接的對端的地址資訊
int ret = getpeername(client_fd, (sockaddr *)&client_addr, &client_addr_len);
if (ret == -1)
{
printf("[WARN] Failed to get client ip: %s\n", strerror(ret));
}
char ip_buf[INET_ADDRSTRLEN];
// 記憶體區域清零
memset(ip_buf, 0, sizeof(ip_buf));
inet_ntop(client_addr.sin_family, &client_addr.sin_addr, ip_buf,
sizeof(ip_buf));
while (1)
{
char buf[1024] = {0};
// 從給定的套接字中讀取指定位元組數的資料，并將其存盤在提供的緩沖區 buf 中
ssize_t ret = st_read(client_st_fd, buf, sizeof(buf), ST_UTIME_NO_TIMEOUT);
if (ret == -1)
{
printf("client st_read error\n");
break;
}
else if (ret == 0)
{
printf("client quit, ip = %s\n", ip_buf);
break;
}
printf("recv from %s, data = https://www.cnblogs.com/Azion/archive/2023/07/12/%s", ip_buf, buf);
ret = st_write(client_st_fd, buf, ret, ST_UTIME_NO_TIMEOUT);
if (ret == -1)
{
printf("client st_write error\n");
}
}
}
void *listen_thread(void *arg)
// 監聽
{
while (1)
{
st_netfd_t client_st_fd =
st_accept((st_netfd_t)arg, NULL, NULL, ST_UTIME_NO_TIMEOUT);
if (client_st_fd == NULL)
{
continue;
}
printf("get a new client, fd = %d\n", st_netfd_fileno(client_st_fd));
st_thread_t client_tid =
st_thread_create(client_thread, (void *)client_st_fd, 0, 0);
if (client_tid == NULL)
{
printf("Failed to st create client thread\n");
}
}
}
int main()
{
// 用于設定 ST 庫的事件系統，
int ret = st_set_eventsys(ST_EVENTSYS_ALT);
if (ret == -1)
{
printf("st_set_eventsys use linux epoll failed\n");
}
// st初始化
ret = st_init();
if (ret != 0)
{
printf("st_init failed. ret = %d\n", ret);
return -1;
}
// 創建套接字
int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
if (listen_fd == -1)
{
ERR_EXIT("socket");
}
int reuse_socket = 1;
// 設定套接字選項
ret = setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &reuse_socket,
sizeof(int));
if (ret == -1)
{
ERR_EXIT("setsockopt");
}
struct sockaddr_in server_addr; // 用于表示 IPv4 地址的結構體
server_addr.sin_family = AF_INET; // 地址族，一般為 AF_INET
server_addr.sin_port = htons(LISTEN_PORT); // 埠
server_addr.sin_addr.s_addr = INADDR_ANY; // ipv4地址結構
// 將套接字與特定的 IP 地址和埠號進行系結
ret =
bind(listen_fd, (struct sockaddr *)&server_addr, sizeof(struct sockaddr));
if (ret == -1)
{
ERR_EXIT("bind");
}
ret = listen(listen_fd, 128);
if (ret == -1)
{
ERR_EXIT("listen");
}
// st_netfd_open_socket() 是 State Threads (ST) 庫中的一個函式，用于創建一個 st_netfd_t 型別的檔案描述符物件，以便進行異步 I/O 操作，
st_netfd_t st_listen_fd = st_netfd_open_socket(listen_fd);
if (!st_listen_fd)
{
printf("st_netfd_open_socket open socket failed.\n");
return -1;
}
// 創建執行緒監聽來一個建立連接的請求
st_thread_t listen_tid =
st_thread_create(listen_thread, (void *)st_listen_fd, 1, 0);
if (listen_tid == NULL)
{
printf("Failed to st create listen thread\n");
}
while (1)
{
st_sleep(1);
}
return 0;
}
示例二
StateThreads創建多執行緒
#include <stdio.h>
#include <st.h>
#include <string>
void *do_calc(void *arg)
{
int sleep_ms = (int)(long int)(char *)arg * 10;
for (;;)
{
printf("in sthread #%dms\n", sleep_ms);
st_usleep(sleep_ms * 1000);
}
return NULL;
}
int main(int argc, char **argv)
{
if (argc <= 1)
{
printf("Test the concurrence of state-threads!\n"
"Usage: %s <sthread_count>\n"
"eg. %s 10000\n",
argv[0], argv[0]);
return -1;
}
if (st_init() < 0)
{
printf("error!");
return -1;
}
int i;
int count = std::stoi(argv[1]);
for (i = 1; i <= count; i++)
{
if (st_thread_create(do_calc, (void *)i, 0, 0) == NULL)
{
printf("error!");
return -1;
}
}
st_thread_exit(NULL);
return 0;
}
關于StateThreads的運行原理，可以看文章《SRS開源直播服務 - StateThreads微執行緒框架學習》
SRS中的StateThreads
使用的原始碼為SRS4.0
系統架構圖：
在SRS的原始碼中，StateThreads在srs_st_init()函式中完成初始化，具體的呼叫流程如下，
SRS的main函式在檔案srs_main_server.cpp中，
srs_main_server.cpp
......
int main(int argc, char** argv)
{
srs_error_t err = do_main(argc, argv);
......
}
srs_error_t do_main(int argc, char** argv)
{
srs_error_t err = srs_success;
// Initialize global or thread-local variables.
if ((err = srs_thread_initialize()) != srs_success) {
return srs_error_wrap(err, "thread init");
}
......
}
srs_app_threads.cpp
......
srs_error_t srs_thread_initialize()
{
srs_error_t err = srs_success;
......
// Initialize ST, which depends on pps cids.
if ((err = srs_st_init()) != srs_success) {
return srs_error_wrap(err, "initialize st failed");
}
......
}
......
srs_service_st.cpp
......
srs_error_t srs_st_init()
{
......
int r0 = 0;
if((r0 = st_init()) != 0){
return srs_error_new(ERROR_ST_INITIALIZE, "st initialize failed, r0=%d", r0);
}
......
在srs_service_st.cpp中呼叫StateThreads庫的初始化函式，完成StateThreads的初始化，
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/557140.html
標籤：其他
上一篇：LEA: Improving Sentence Similarity Robustness to Typos Using Lexical Attention Bias 論文閱讀
下一篇：返回列表