文章目录
- 需求
- 介绍
- 协程
- 注意
- 调度
- 部署
- 环境搭建
- 测试
- 测试库是否正常
- 测试TCP服务器
需求
最近在对网关模型的并发性能进行验证，
核心目标不仅是让服务器能够承载更多的请求，
还力求在协议栈的解析阶段实现极致的轻量化和无锁操作，从而大幅提升处理效率和系统响应速度。
为实现这一目标，选取了业界内多个领先的并发处理框架作为候选对象进行测试，以评估它们在处理高并发请求时的表现和稳定性。以下是关于其中一个框架——StateThreads的一些相关内容
介绍
https://state-threads.sourceforge.net/
协程
通常我们把协程理解为是一种程序自己实现调度、用于提高运行效率、降低开发复杂度的东西。提高运行效率很好理解，因为在程序层自己完成了部分的调度，降低了对系统调度的依赖，减少了大量的中断和换页操作。
很多语言都拥有协程，例如python或者golang。而对于c/c++而言，通常实现协程的常见方式，通常是依赖于glibc提供的setjump&longjump或者基于汇编语言
注意
协程不是用来提高服务器性能的，是用来提高代码的编写速度和可维护性的
比如协议栈解析中最常见的结构 length-data ， 其中length字段（假设为2字节）用于指示紧随其后的data部分的长度
解析：
- 使用epoll来监听文件描述符（fd）的可读事件，当检测到fd可读事件时，程序需要首先读取这2字节的length，然后根据这个长度去读取相应长度的data。如果首次读取的data不完整，程序需要继续监听fd的可读事件，并在后续读取中维护状态，确保所有数据被完整接收。
- 使用协程后上述 过程被极大地简化和抽象化，复杂的逻辑被封装在协程库中，调度由协程库实现，代码逻辑简化为，读取2个字节解析出length，读取length长度数据，然后进行业务处理
调度
ST调度工作原理，ST运行环境维护了四种队列，分别是IOQ、RUNQ、SLEEPQ以及ZOMBIEQ，当每个thread处于不同队列中对应不同的状态（ST顾名思义所谓thread状态机）。比如polling请求的时候，当前thread就加入IOQ表示等待事件（如果有timeout同时会被放到SLEEPQ中），当事件触发时，thread就从IOQ（如果有timeout同时会从SLEEPQ）移除并转移到RUNQ等待被调度，成为当前的running
thread，相当于操作系统的就绪队列，跟传统EDSM对应起来就是注册回调以及激活回调。再比如模拟同步控制wait/sleep/lock的时候，当前thread会被放入SLEEPQ，直到被唤醒或者超时再次进入RUNQ以待调度。
参考 浅析State-Thread
部署
下载
https://sourceforge.net/projects/state-threads/files/
解压
编译
# 根据具体环境选择参数
make linux-debug
参数包括
aix-debug
aix-optimized
cygwin-debug
cygwin-optimized
darwin-debug
darwin-optimized
freebsd-debug
freebsd-optimized
…… 等
环境搭建
头文件和库都在obj目录下 拷贝到开发环境
obj目录下还有配套的测试文件 proxy server lookupdns
.
测试
测试环境目录结构
测试库是否正常
编译命令
gcc test.c -o test -l st -L .
#include <stdio.h>
#include "st.h"
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");
printf("Usage: %s <sthread_count>\n");
printf("eg. %s 10000\n", argv[0], argv[0]);
return -1;
}
if(st_init() < 0){
printf("st_init error!");
return -1;
}
int i;
int count = atoi(argv[1]);
for(i = 1; i <= count; i++){
if(st_thread_create(do_calc, (void*)i, 0, 0) == NULL){
printf("st_thread_create error!");
return -1;
}
}
st_thread_exit(NULL);
return 0;
}
测试TCP服务器
编译命令
gcc -g tcp-srv.c -o tcp-srv -l st -L. -I .
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <st.h>
#define BACKLOG 5
#define IOBUFSIZE 8192
#define PORT 7474
static void *handle_request(void *arg)
{
st_netfd_t cli_nfd = (st_netfd_t) arg;
struct pollfd pd;
pd.fd = st_netfd_fileno(cli_nfd);
pd.events = POLLIN;
char buf[IOBUFSIZE];
int nw, nr;
for ( ; ; ) {
/*pd.revents = 0;
if (st_poll(&pd, 1, ST_UTIME_NO_TIMEOUT) <= 0) {
printf("st_poll<=0\n");
break;
}*/
//if (pd.revents & POLLIN) {
nr = (int) st_read(cli_nfd, buf, IOBUFSIZE, ST_UTIME_NO_TIMEOUT);
if (nr <= 0)break;
printf("[recv][%d] %s\n",nr,buf);
nw = st_write(cli_nfd, buf, nr, ST_UTIME_NO_TIMEOUT);
printf("[write] %d\n",nw);
if (nw == nr)break;
//}
}
done:
st_netfd_close(cli_nfd);
return NULL;
}
int main() {
int sock, client_fd;
struct sockaddr_in address;
struct sockaddr_in cli_addr;
socklen_t addrlen = sizeof(address);
st_netfd_t cli_nfd, srv_nfd;
int n =0;
// 初始化StateThreads库
if (st_init() != 0) {
perror("st_init");
return 0;
}
if ((sock = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
printf("socket");
exit(1);
}
n = 1;
if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (char *)&n, sizeof(n)) < 0) {
printf("setsockopt");
exit(1);
}
// 绑定socket到端口
address.sin_family = AF_INET;
address.sin_addr.s_addr = INADDR_ANY;
address.sin_port = htons(PORT);
if (bind(sock, (struct sockaddr *)&address, sizeof(address)) < 0) {
printf("bind");
exit(1);
}
listen(sock, 128);
if ((srv_nfd = st_netfd_open_socket(sock)) == NULL) {
printf("st_netfd_open");
exit(1);
}
printf("tcp srv start\n");
for ( ; ; ) {
n = sizeof(cli_addr);
cli_nfd = st_accept(srv_nfd, (struct sockaddr *)&cli_addr, &n,ST_UTIME_NO_TIMEOUT);
if (cli_nfd == NULL) {
printf("st_accept\n");
continue;
}
if (st_thread_create(handle_request, cli_nfd, 0, 0) == NULL) {
printf("st_thread_create\n");
continue;
}
}
close(sock);
st_netfd_close(srv_nfd);
return 0;
}
