使用State Threads实现简单的服务器

使用State Threads实现简单的服务器

关于State Threads的介绍可以参考： ​​​谈谈并发编程中的协程​​​ ​​​网络架构库：State Threads​​​ ​​​State Threads：回调终结者​​​ 一.源码编译下面是在Fedora 20（装在了虚拟机中）上的实操记录： 1.从官网 ​​​http://sourceforge-/projects/state-threads​​​-源码包，最新版是1.9 2.-完st-1.9.tar.gz，然后解压

tar zxvf st-1.9.tar.gzcd st-1.9make

此时会提示“Please specify one of the following targets”，如下图所示：

我选择的是linux-debug。

make linux-debug

此时会在目录st-1.9中产生一个新的目录LINUX_3.11.10-301.fc20.i686_DBG，里面有生成的中间文件*.o, 头文件st.h，libst.so，libst.a和example中的三个例子：lookupdns，proxy，server。

需要注意的是st.h是动态生成的，这种方法值得学习。

二.doc目录研究

在st-1.9源码中doc目录有几个文档，可以参考：

st.html——ST库概论，翻译在

​​网络架构库：State Threads​​

timeout_heap.txt——超时heap实现

notes.html——给出了编程注意点，包括移植，信号，进程内同步，进程间同步，非网络IO，超时处理，特别谈到进程内同步非常简单，不需要同步资源；非网络IO中谈到drawback和设计时需要避免的方法

reference.html——一个API接口文档介绍，需要认真阅读和熟悉，但是需要编码实战来加深理解

对于reference.html，最重要的是文尾的Program Structure部分，它给出了在一个网络应用程序中使用ST库的基本步骤：

1.如果需要，使用下面的pre-init（预初始化）函数配置ST库，设置时间，事件通知机制

st_set_utime_function()

st_set_eventsys()

2.调用初始化函数st_init()来初始化ST库

3.如果需要，调用post-init（后初始化）函数来配置ST库，设置timecache，随机化线程栈，进程resume和stop的回调函数

st_timecache_set()

st_randomize_stacks()

st_set_switch_in_cb()

st_set_switch_out_cb()

4.生成不同process之间共享的资源，创建并绑定socket，监听socket，生成共享内存段，IPC（进程内通信）channel和同步原语。

st_netfd_open_socket()

st_netfd_serialize_accept()

5.通过fork()创建多进程, 父进程退出或是watchdog

6.在每个子进程中创建thread pool来处理user connection，线程池中的每个线程可以接受客户端连接，也可以连接到其他服务器，或者执行各种network I/O等等

st_thread_create()

st_accept()

st_connect()

st_read()

st_write()

注意：在使用ST库时，只有ST库的I/O函数可以用于network I/O，其他的I/O调用（比如说fread，fwrite）都可能阻塞调用进程。

三.example目录

首先阅读里面的README，它简单介绍了这三个例子的基本情况和用法

server包含server.c和error.c

lookupdns包含lookupdns.c和res.c

proxy包含proxy.c

这里分析server的实现。server接受一个客户端连接，接收客户端数据并返给客户端一个简单的HTML网页（我会做适当修改，让server将接收到的内容原样返回）。以server为基础，我们可以很方便的实现其他的服务器。

我将源码server.c中的void handle_session(long srv_socket_index, st_netfd_t cli_nfd)函数改成如下形式，这样server会将接收到的内容原样返回，方便测试多个客户端的链接。

void handle_session(long srv_socket_index, st_netfd_t cli_nfd){ char buf[512]={'\0'}; int n = 0; struct in_addr *from = st_netfd_getspecific(cli_nfd); if (st_read(cli_nfd, buf, sizeof(buf), SEC2USEC(REQUEST_TIMEOUT)) < 0) { err_sys_report(errfd, "WARN: can't read request from %s: st_read", inet_ntoa(*from)); return; } n = strlen(buf); if (st_write(cli_nfd, buf, n, ST_UTIME_NO_TIMEOUT) != n) { err_sys_report(errfd, "WARN: can't write response to %s: st_write", inet_ntoa(*from)); return; } RQST_COUNT(srv_socket_index)++;}

作为Qt的忠实粉丝，我在st-1.9源码目录中新建一个Qt工程，pro文件如下，这样就可以愉快的调试了。

TEMPLATE = appCONFIG += consoleCONFIG -= app_bundleCONFIG -= qtTARGET = MyServerINCLUDEPATH += LINUX_3.11.10-301.fc20.i686_DBGHEADERS += LINUX_3.11.10-301.fc20.i686_DBG/st.hSOURCES += examples/server.c \ examples/error.cLIBS += LINUX_3.11.10-301.fc20.i686_DBG/libst.a

1.测试非daemon模式（即interactive模式）下的网络通信

MyServer在运行时需要解析参数，所以在Qt Creator中添加参数如下：

从源码中可以看出-i用了设置interactive模式，后边的参数不管是不是1，都会设置成功。

interactive模式比较简单，不会创建虚拟处理器（VP），也不会写日志。

从static void *handle_connections(void *arg)函数中可以看出，处理完会话后，会将当前socket关闭。该函数实际上实现了一个线程池，最小线程数是max_wait_threads，最大线程数是max_threads。

static void *handle_connections(void *arg){ st_netfd_t srv_nfd, cli_nfd; struct sockaddr_in from; int fromlen; long i = (long) arg; srv_nfd = srv_socket[i].nfd; fromlen = sizeof(from); while (WAIT_THREADS(i) <= max_wait_threads) { cli_nfd = st_accept(srv_nfd, (struct sockaddr *)&from, &fromlen, ST_UTIME_NO_TIMEOUT); if (cli_nfd == NULL) { err_sys_report(errfd, "ERROR: can't accept connection: st_accept"); continue; } /* Save peer address, so we can retrieve it later */ st_netfd_setspecific(cli_nfd, &from.sin_addr, NULL); WAIT_THREADS(i)--; BUSY_THREADS(i)++; if (WAIT_THREADS(i) < min_wait_threads && TOTAL_THREADS(i) < max_threads) { /* Create another spare thread */ if (st_thread_create(handle_connections, (void *)i, 0, 0) != NULL) WAIT_THREADS(i)++; else err_sys_report(errfd, "ERROR: process %d (pid %d): can't create" " thread", my_index, my_pid); } handle_session(i, cli_nfd); st_netfd_close(cli_nfd);//关闭socket WAIT_THREADS(i)++; BUSY_THREADS(i)--; } WAIT_THREADS(i)--; return NULL;}

客户端，可以分分钟用Qt写一个，就放在windows上吧，运行效果如下所示，我启动了两个客户端：

之所以会弹提示框，是因为检测到MyServer将socket关闭了。

2.测试daemon模式

daemon模式下，server创建一个固定数量的进程（“virtual processors”虚拟处理器或VP），并在它们死亡时管理它们。每个虚拟处理器管理它自己的独立的一组state threads（ST：状态线程），其数量各不相同，取决于server的负载。每个VP只监听一个套接字。最初的进程（即daemon进程）成为watchdog（监督者），等待其子进程（也就是VP）死亡或请求终止或重新启动的信号。收到重启信号（SIGHUP）后，所有VP关闭然后重新打开日志文件和重新加载配置。所有当前活动的连接都保留活性。这里假定新配置只影响请求处理而不是服务器参数——例如VP的数量，线程限制，绑定地址等。这些参数被通常被指定为命令行参数，所以服务器为了改变它们必须停止，然后再次启动。

每个状态线程循环处理来自单个socket的监听。一次只有一个ST在VP上运行，而VP之间不共享内存，所以任何数据都不需要互斥锁，服务器可以自由使用所有的静态变量和非重入库的函数，这大大简化了编程和调试，并提高性能（例如，对于++和---全局计数是安全的或调用inet_ntoa()不需要使用互斥）。每个VP中的当前线程负责保证那个VP的均衡，该线程可以开始一个新线程或终止自身——当备用线程数量超过了上限或者下限。

这个模式涉及到了多进程（fork）、守护进程（daemon）、进程间通信（signal）等linux知识，还是比较复杂的。在Qt Creator中使用参数“-l ./”就能以daemon模式启动MyServer，这里“-l ./”用来设置日志的路径，因为daemon模式下会打印日志。

daemon模式框架图

这个图是值得我们学习的，在进行服务器开发的时候，通常用守护进程来管理子进程，真正干活的是子进程。这样做的好处是当一个子进程挂了，不影响服务器的功能，因为守护进程不处理事务，因而挂掉的可能性要小很多。

启动守护进程

static void start_daemon(void){ pid_t pid; /* Start forking */ if ((pid = fork()) < 0) err_sys_quit(errfd, "ERROR: fork"); if (pid > 0) exit(0); /* parent */ /* First child process */ setsid(); /* become session leader */ if ((pid = fork()) < 0) err_sys_quit(errfd, "ERROR: fork"); if (pid > 0) /* first child */ exit(0); umask(022); if (chdir(logdir) < 0) err_sys_quit(errfd, "ERROR: can't change directory to %s: chdir", logdir);}

创建并管理子进程

static void start_processes(void){ int i, status; pid_t pid; sigset_t mask, omask; if (interactive_mode) { my_index = 0; my_pid = getpid(); return; } for (i = 0; i < vp_count; i++) { if ((pid = fork()) < 0) { err_sys_report(errfd, "ERROR: can't create process: fork"); if (i == 0) exit(1); err_report(errfd, "WARN: started only %d processes out of %d", i, vp_count); vp_count = i; break; } if (pid == 0) { my_index = i; my_pid = getpid(); /* Child returns to continue in main() */ return; } vp_pids[i] = pid; } /* * Parent process becomes a "watchdog" and never returns to main(). */ /* Install signal handlers */ Signal(SIGTERM, wdog_sighandler); /* terminate */ Signal(SIGHUP, wdog_sighandler); /* restart */ Signal(SIGUSR1, wdog_sighandler); /* dump info */ /* Now go to sleep waiting for a child termination or a signal */ for ( ; ; ) { if ((pid = wait(&status)) < 0) { if (errno == EINTR) continue; err_sys_quit(errfd, "ERROR: watchdog: wait"); } /* Find index of the exited child */ for (i = 0; i < vp_count; i++) { if (vp_pids[i] == pid) break; } /* Block signals while printing and forking */ sigemptyset(&mask); sigaddset(&mask, SIGTERM); sigaddset(&mask, SIGHUP); sigaddset(&mask, SIGUSR1); sigprocmask(SIG_BLOCK, &mask, &omask); if (WIFEXITED(status)) err_report(errfd, "WARN: watchdog: process %d (pid %d) exited" " with status %d", i, pid, WEXITSTATUS(status)); else if (WIFSIGNALED(status)) err_report(errfd, "WARN: watchdog: process %d (pid %d) terminated" " by signal %d", i, pid, WTERMSIG(status)); else if (WIFSTOPPED(status)) err_report(errfd, "WARN: watchdog: process %d (pid %d) stopped" " by signal %d", i, pid, WSTOPSIG(status)); else err_report(errfd, "WARN: watchdog: process %d (pid %d) terminated:" " unknown termination reason", i, pid); /* Fork another VP */ if ((pid = fork()) < 0) { err_sys_report(errfd, "ERROR: watchdog: can't create process: fork"); } else if (pid == 0) { my_index = i; my_pid = getpid(); /* Child returns to continue in main() */ err_report(errfd, "Test Information"); return; } vp_pids[i] = pid; /* Restore the signal mask */ sigprocmask(SIG_SETMASK, &omask, NULL); }}

这个函数中先创建了vp_count个子进程，然后主进程进入一个for循环中，在循环中等待进程的信号（wait(&status)），wait函数值阻塞的。如果子进程接收到SIGTERM信号时，在static void child_sighandler(int signo)中会退出该子进程。同时程序走到wait函数之后，创建出一个新的子进程，这样子进程的个数始终维持在vp_count个。

如果是主进程接收到SIGTERM信号，在static void wdog_sighandler(int signo)中会先kill所有的子进程，然后再退出主进程。此时，所有进程退出。我的电脑是四核，vp_count=4，所以一共有五个进程，如下图所示：

前四列依次是父进程ID（PPID），进程ID（PID），进程组ID（PGID）和会话ID（SID）。第一个进程的父进程ID是1，这个进程就是守护进程，下面四个进程都是子进程，它们的父进程都是第一个进程。

关于SIGTERM信号，可以通过下列指令测试：

kill -SIGTERM 进程ID号

State Threads在开源流媒体服务器Simple-RTMP-Server（SRS）中已经有了教科书般的应用，详见：​​https://github.com/winlinvip/srs​​

​

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。