从头编写高性能服务程序9-多进程非阻塞epoll-prefork-hook
整个基础结构已经基本确定了
接下来做一些细节工作
首先把一些函数抽取出来.
例如prefork独立出来.socket->bind->listen独立出来
这里我们引入一个新的思路
原先由统一的函数在epoll_wait之后对events里面的fd进行处理
但是每个fd可能需要处理的方式都不同.
怎么样针对不同的fd来调用特定的函数呢?
首先在epoll_event结构中有data成员
而data的定义如下
typedef union epoll_data {
void *ptr;
int fd;
__uint32_t u32;
__uint64_t u64;
} epoll_data_t;
struct epoll_event {
__uint32_t events; /* Epoll events */
epoll_data_t data; /* User data variable */
};
可见既可以在events里面放data.fd
也可以使用data.ptr来指向一个指针
当fd有消息时内核将对应的ev变量塞入events数组的时候
如果我们只是用fd来指向注册的,那么获取数据的时候只能得到对应的fd
这样使用什么函数来处理这个fd就需要另行判断
那么如果使用ptr来指向一个结构
而结构内保存了fd以及处理这个fd所使用的函数指针
那当我们得到events数组内的事件时
就可以直接调用ptr指向的函数指针了.
这就类似Nginx中的hook函数.
在Nginx中几乎任何一种事件都会绑定其处理函数
而由模块实现距离的函数,然后在hook上去.
那么下面的代码我们就模拟这个方法:
我们建立一个数据结构来保存每个fd以及对应的处理函数
struct event_handle{
int fd;
int (* handle)(int fd);
};
handle_hook是我们为每个fd注册的处理函数
当accept获得新的accept_fd之后
我们使用
ev_handles[accept_handles].handle = handle_hook
来将对应的函数注册到对应的events内
在fd得到通知的时候
使用
(*current_handle)(current_fd)
来进行处理
下载:
- #include <sys/socket.h>
- #include <sys/wait.h>
- #include <netinet/in.h>
- #include <sys/epoll.h>
- #include <sys/sendfile.h>
- #include <unistd.h>
- #include <stdio.h>
- #include <stdlib.h>
- #include <string.h>
- #include <strings.h>
- #include <fcntl.h>
-
- int create_listen_fd(int port){
- int listen_fd;
- struct sockaddr_in my_addr;
- if ((listen_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1){
- perror("create socket error");
- exit(1);
- }
- int flag;
- if (setsockopt(listen_fd,SOL_SOCKET,SO_REUSEADDR
- ,(char *)&flag,sizeof(flag)) == -1){
- perror("setsockopt error");
- }
- int flags = fcntl(listen_fd, F_GETFL, 0);
- fcntl(listen_fd, F_SETFL, flags|O_NONBLOCK);
- my_addr.sin_family = AF_INET;
- my_addr.sin_port = htons(port);
- my_addr.sin_addr.s_addr = INADDR_ANY;
- bzero(&(my_addr.sin_zero), 8);
- if (bind(listen_fd, (struct sockaddr *)&my_addr,
- sizeof(struct sockaddr_in)) == -1) {
- perror("bind error");
- exit(1);
- }
- if (listen(listen_fd,1) == -1){
- perror("listen error");
- exit(1);
- }
- return listen_fd;
- }
-
- int create_accept_fd(int listen_fd){
- int addr_len = sizeof( struct sockaddr_in );
- struct sockaddr_in remote_addr;
- int accept_fd = accept( listen_fd,
- (struct sockaddr *)&remote_addr, &addr_len );
- int flags = fcntl(accept_fd, F_GETFL, 0);
- fcntl(accept_fd, F_SETFL, flags|O_NONBLOCK);
- return accept_fd;
- }
-
- int fork_process(int process_num){
- int i;
- int pid=-1;
- for(i = 0; i < process_num; i++){
- if(pid != 0){
- pid = fork();
- }
- }
- return pid;
- }
-
- int handle_normal(int socket_fd){
- char in_buf[1024];
- memset(in_buf, 0, 1024);
- int recv_num = recv( socket_fd, &in_buf, 1024, 0 );
- if( recv_num ==0 ){
- close(socket_fd);
- printf("ProcessID:%d,EPOLLIN,fd:%d,closed\n", getpid(), socket_fd);
- }
- else{
- printf("ProcessID:%d,EPOLLIN,fd:%d,recv:%s\n", getpid(), socket_fd, in_buf);
- }
- return recv_num;
- }
-
- int handle_hook(int socket_fd){
- char in_buf[1024];
- memset(in_buf, 0, 1024);
- int recv_num = recv( socket_fd, &in_buf, 1024, 0 );
- if( recv_num ==0 ){
- close(socket_fd);
- printf("ProcessID:%d,EPOLLIN,fd:%d,closed\n", getpid(), socket_fd);
- }
- else{
- printf("ProcessID:%d,EPOLLIN,fd:%d,recv_num:%d;recv:", getpid(), socket_fd, recv_num);
- for (int i = 0; i<recv_num; i++){
- printf("%02x ",in_buf[i]);
- }
- printf("\n");
- }
- return recv_num;
- }
-
- struct event_handle{
- int fd;
- int (* handle)(int fd);
- };
- typedef int (* EVENT_HANDLE)(int);
- typedef struct event_handle * EH;
-
- int main(){
- int listen_fd = create_listen_fd(3389);
- int pid = fork_process(3);
- if(pid == 0){
- int accept_handles = 0;
- struct epoll_event ev,events[20];
- int epfd = epoll_create(256);
- int ev_s = 0;
-
- ev.data.fd = listen_fd;
- ev.events = EPOLLIN|EPOLLET;
- epoll_ctl(epfd,EPOLL_CTL_ADD,listen_fd,&ev);
- struct event_handle ev_handles[256];
- for(;;){
- ev_s = epoll_wait( epfd,events, 20, 500 );
- int i = 0;
- for(i = 0; i<ev_s; i++){
- if(events[i].data.fd == listen_fd){
- int max_process_accept = 3;
- if(accept_handles < max_process_accept){
- accept_handles++;
- int accept_fd = create_accept_fd(listen_fd);
- ev_handles[accept_handles].fd = accept_fd;
- ev_handles[accept_handles].handle = handle_hook;
- ev.data.ptr = &ev_handles[accept_handles];
- ev.events = EPOLLIN|EPOLLET;
- epoll_ctl(epfd,EPOLL_CTL_ADD,accept_fd,&ev);
- printf("ProcessID:%d,EPOLLIN,fd:%d,accept:%d\n", getpid(), listen_fd, accept_fd);
- }
- }
- else if(events[i].events&EPOLLIN){
- EVENT_HANDLE current_handle = ((EH)(events[i].data.ptr))->handle;
- int current_fd = ((EH)(events[i].data.ptr))->fd;
- if( (*current_handle)(current_fd) == 0){
- accept_handles--;
- }
- }
- else if(events[i].events&EPOLLOUT){
- //need add write event process
- }
- }
- }
- }
- else{
- //manager the process
- int child_process_status;
- wait(&child_process_status);
- }
-
- return 0;
- }
从头编写高性能服务程序10-请求解析
最终的形态基本上在上次的结构中就定型了
当然有些细节需要完善
不过基本上用这个结构来写service已经是OK了
那么现在就是继续细化这个结构.用来写个比较靠近实际的应用
有了链接的管理
接下来就是对通讯协议的实现
由于是从头开始写
所以协议也由我们自己来实现
先是对请求的解析
从客户端telnet传送过来的数据
回行是用/r/n结尾的
所以我们不停的接受数据
然后判断数据的最后是否是/r/n
如果是的话.就把它和以前的数据一起拼接起来
然后调用请求分析来解析指令
在event_handle结构中
我们加入了command数组
用来存放每次传输过来的数据
直至遇到以/r/n结尾的数据.然后拼接起来,输出,再清空这个数组
从头再接受新的指令
由于使用了epoll和非阻塞accept_fd
所以每次接受到的数据是零散的
需要将每次recv的数据连续的拼接到一个变量中
这就是command数组存在的理由
而command_pos用来保存的是每次拼接后数组的实际存放数据的量
也可以认为是最后一个数据所在数组中的位置
便于下次拼接
下载:
- #include <sys/socket.h>
- #include <sys/wait.h>
- #include <netinet/in.h>
- #include <sys/epoll.h>
- #include <sys/sendfile.h>
- #include <unistd.h>
- #include <stdio.h>
- #include <stdlib.h>
- #include <string.h>
- #include <strings.h>
- #include <fcntl.h>
-
- typedef struct event_handle{
- int fd;
- int ( * handle )( struct event_handle * ev );
- char command[1024];
- int command_pos;
- } EV,* EH;
- typedef int ( * EVENT_HANDLE )( struct event_handle * ev );
-
-
- int create_listen_fd( int port ){
- int listen_fd;
- struct sockaddr_in my_addr;
- if ( ( listen_fd = socket( AF_INET, SOCK_STREAM, 0 ) ) == -1 ){
- perror( "create socket error" );
- exit( 1 );
- }
- int flag;
- if ( setsockopt( listen_fd, SOL_SOCKET, SO_REUSEADDR
- , ( char * )&flag, sizeof( flag ) ) == -1 ){
- perror( "setsockopt error" );
- }
- int flags = fcntl( listen_fd, F_GETFL, 0 );
- fcntl( listen_fd, F_SETFL, flags|O_NONBLOCK );
- my_addr.sin_family = AF_INET;
- my_addr.sin_port = htons( port );
- my_addr.sin_addr.s_addr = INADDR_ANY;
- bzero( &( my_addr.sin_zero ), 8 );
- if ( bind( listen_fd, ( struct sockaddr * )&my_addr,
- sizeof( struct sockaddr_in ) ) == -1 ) {
- perror( "bind error" );
- exit( 1 );
- }
- if ( listen( listen_fd, 1 ) == -1 ){
- perror( "listen error" );
- exit( 1 );
- }
- return listen_fd;
- }
-
- int create_accept_fd( int listen_fd ){
- int addr_len = sizeof( struct sockaddr_in );
- struct sockaddr_in remote_addr;
- int accept_fd = accept( listen_fd,
- ( struct sockaddr * )&remote_addr, &addr_len );
- int flags = fcntl( accept_fd, F_GETFL, 0 );
- fcntl( accept_fd, F_SETFL, flags|O_NONBLOCK );
- return accept_fd;
- }
-
- int fork_process( int process_num ){
- int i;
- int pid=-1;
- for( i = 0; i < process_num; i++ ){
- if( pid != 0 ){
- pid = fork();
- }
- }
- return pid;
- }
-
- int handle_hook_v2( EH ev ){
- char in_buf[1024];
- memset( in_buf, 0, 1024 );
- int recv_num = recv( ev->fd, &in_buf, 1024, 0 );
- if( recv_num ==0 ){
- printf( "ProcessID:%d, EPOLLIN, fd:%d, closed\n", getpid(), ev->fd );
- printf( " recved:%s\n", ev->command );
- close( ev->fd );
- }
- else{
- printf( "ProcessID:%d, EPOLLIN, fd:%d, recv_num:%d;recv:", getpid(), ev->fd, recv_num );
- int i;
- for( i = 0; i<recv_num; i++ ){
- printf( "%02x ", in_buf[i] );
- }
- printf( "\n" );
- memcpy( ev->command + ev->command_pos, in_buf, recv_num );
- ev->command_pos += recv_num;
- if( recv_num == 2 && ( !memcmp( &in_buf[recv_num-2], "\r\n", 2 ) ) ){
- printf( " recved:%s\n", ev->command );
- memset( ev->command, 0, 1024 );
- ev->command_pos = 0;
- }
- }
- return recv_num;
- }
-
-
-
- int main(){
- int listen_fd = create_listen_fd( 3389 );
- int pid = fork_process( 3 );
- if( pid == 0 ){
- int accept_handles = 0;
- struct epoll_event ev, events[20];
- int epfd = epoll_create( 256 );
- int ev_s = 0;
-
- ev.data.fd = listen_fd;
- ev.events = EPOLLIN|EPOLLET;
- epoll_ctl( epfd, EPOLL_CTL_ADD, listen_fd, &ev );
- struct event_handle ev_handles[256];
- for( ;; ){
- ev_s = epoll_wait( epfd, events, 20, 500 );
- int i = 0;
- for( i = 0; i<ev_s; i++ ){
- if( events[i].data.fd == listen_fd ){
- int max_process_accept = 3;
- if( accept_handles < max_process_accept ){
- accept_handles++;
- int accept_fd = create_accept_fd( listen_fd );
- ev_handles[accept_handles].fd = accept_fd;
- ev_handles[accept_handles].handle = handle_hook_v2;
- ev_handles[accept_handles].command_pos = 0;
- memset( ev_handles[accept_handles].command, 0, 1024 );
- ev.data.ptr = &ev_handles[accept_handles];
- ev.events = EPOLLIN|EPOLLET;
- epoll_ctl( epfd, EPOLL_CTL_ADD, accept_fd, &ev );
- printf( "ProcessID:%d, EPOLLIN, fd:%d, accept:%d\n", getpid(), listen_fd, accept_fd );
- }
- }
- else if( events[i].events&EPOLLIN ){
- EVENT_HANDLE current_handle = ( ( EH )( events[i].data.ptr ) )->handle;
- EH current_event = ( EH )( events[i].data.ptr );
- if( ( *current_handle )( current_event ) == 0 ){
- accept_handles--;
- }
- }
- else if( events[i].events&EPOLLOUT ){
- //need add write event process
- }
- }
- }
- }
- else{
- //manager the process
- int child_process_status;
- wait( &child_process_status );
- }
-
- return 0;
- }
从头编写高性能服务程序11-指令处理&sendfile
实现命令的获取之后
现在是增加对command的解析以及对应的反馈
为了做些稍微有意义的事情.我把这个service做的工作定位在以下内容
1.查询文件大小
2.返回文件内容
3.删除文件
协议的格式如下
请求模式:文件名称
例如
1:new.txt
请求模式见源代码中的宏定义
这次代码也对原来的程序作了一定的修改
从丑陋的代码渐渐修改.
希望在最终程序完成的时候
能够有比较好的代码风格
这个版本中有一个bug
就是sendfile只是调用一次
而实际上如果是较大的文件
需要在判断EPLLOUT之后不停的sendfile
直到EAGAIN遇见accept_fd阻塞为止
这样直至下次EPOLLOUT发生
再从上次暂停的位置继续发送
下一个版本中将会有这个BUG的修正
下载:
- #include <sys/socket.h>
- #include <sys/wait.h>
- #include <netinet/in.h>
- #include <netinet/tcp.h>
- #include <sys/epoll.h>
- #include <sys/sendfile.h>
- #include <sys/stat.h>
- #include <unistd.h>
- #include <stdio.h>
- #include <stdlib.h>
- #include <string.h>
- #include <strings.h>
- #include <fcntl.h>
-
- #define HANDLE_INFO 1
- #define HANDLE_SEND 2
- #define HANDLE_DEL 3
- #define HANDLE_CLOSE 4
-
- #define MAX_REQLEN 1024
- #define MAX_PROCESS_CONN 3
- #define FIN_CHAR 0x00
- #define SUCCESS 0
- #define ERROR -1
-
- typedef struct event_handle{
- int socket_fd;
- int file_fd;
- char request[MAX_REQLEN];
- int request_len;
- int ( * handle )( struct event_handle * ev );
- int handle_method;
-
- } EV,* EH;
- typedef int ( * EVENT_HANDLE )( struct event_handle * ev );
-
- int create_listen_fd( int port ){
- int listen_fd;
- struct sockaddr_in my_addr;
- if( ( listen_fd = socket( AF_INET, SOCK_STREAM, 0 ) ) == -1 ){
- perror( "create socket error" );
- exit( 1 );
- }
- int flag;
- int olen = sizeof(int);
- if( setsockopt( listen_fd, SOL_SOCKET, SO_REUSEADDR
- , (const void *)&flag, olen ) == -1 ){
- perror( "setsockopt error" );
- }
- flag = 1;
- if( setsockopt( listen_fd, IPPROTO_TCP, TCP_DEFER_ACCEPT, &flag, olen ) == -1 ){
- perror( "setsockopt error" );
- }
- int flags = fcntl( listen_fd, F_GETFL, 0 );
- fcntl( listen_fd, F_SETFL, flags|O_NONBLOCK );
- my_addr.sin_family = AF_INET;
- my_addr.sin_port = htons( port );
- my_addr.sin_addr.s_addr = INADDR_ANY;
- bzero( &( my_addr.sin_zero ), 8 );
- if( bind( listen_fd, ( struct sockaddr * )&my_addr,
- sizeof( struct sockaddr_in ) ) == -1 ) {
- perror( "bind error" );
- exit( 1 );
- }
- if( listen( listen_fd, 1 ) == -1 ){
- perror( "listen error" );
- exit( 1 );
- }
- return listen_fd;
- }
-
- int create_accept_fd( int listen_fd ){
- int addr_len = sizeof( struct sockaddr_in );
- struct sockaddr_in remote_addr;
- int accept_fd = accept( listen_fd,
- ( struct sockaddr * )&remote_addr, &addr_len );
- int flags = fcntl( accept_fd, F_GETFL, 0 );
- fcntl( accept_fd, F_SETFL, flags|O_NONBLOCK );
- return accept_fd;
- }
-
- int fork_process( int process_num ){
- int i;
- int pid=-1;
- for( i = 0; i < process_num; i++ ){
- if( pid != 0 ){
- pid = fork();
- }
- }
- return pid;
- }
-
- int init_evhandle(EH ev,int socket_fd,EVENT_HANDLE handle){
- ev->socket_fd = socket_fd;
- ev->handle = handle;
- ev->request_len = 0;
- ev->handle_method = 0;
- memset( ev->request, 0, 1024 );
- }
- //accept->accept_queue->request->request_queue->output->output_queue
- //multi process sendfile
- int parse_request(EH ev){
- ev->request_len--;
- *( ev->request + ev->request_len - 1 ) = 0x00;
- int i;
- for( i=0; i<ev->request_len; i++ ){
- if( ev->request[i] == ':' ){
- ev->request_len = ev->request_len-i-1;
- char temp[MAX_REQLEN];
- memcpy( temp, ev->request, i );
- ev->handle_method = atoi( temp );
- memcpy( temp, ev->request+i+1, ev->request_len );
- memcpy( ev->request, temp, ev->request_len );
- break;
- }
- }
- handle_request( ev );
- return SUCCESS;
- }
-
- int handle_request(EH ev){
- struct stat file_info;
- switch( ev->handle_method ){
- case HANDLE_INFO:
- ev->file_fd = open( ev->request, O_RDONLY );
- if( ev->file_fd == -1 ){
- send( ev->socket_fd, "open file failed\n", strlen("open file failed\n"), 0 );
- return -1;
- }
- fstat(ev->file_fd, &file_info);
- char info[MAX_REQLEN];
- sprintf(info,"file len:%d\n",file_info.st_size);
- send( ev->socket_fd, info, strlen( info ), 0 );
- break;
- case HANDLE_SEND:
- ev->file_fd = open( ev->request, O_RDONLY );
- if( ev->file_fd == -1 ){
- send( ev->socket_fd, "open file failed\n", strlen("open file failed\n"), 0 );
- return -1;
- }
- fstat(ev->file_fd, &file_info);
- sendfile( ev->socket_fd, ev->file_fd, 0, file_info.st_size );
- break;
- case HANDLE_DEL:
- break;
- case HANDLE_CLOSE:
- break;
- }
- finish_request( ev );
- return SUCCESS;
- }
-
- int finish_request(EH ev){
- close(ev->socket_fd);
- close(ev->file_fd);
- ev->handle_method = -1;
- clean_request( ev );
- return SUCCESS;
- }
-
- int clean_request(EH ev){
- memset( ev->request, 0, MAX_REQLEN );
- ev->request_len = 0;
- }
-
- int handle_hook_v2( EH ev ){
- char in_buf[MAX_REQLEN];
- memset( in_buf, 0, MAX_REQLEN );
- int recv_num = recv( ev->socket_fd, &in_buf, MAX_REQLEN, 0 );
- if( recv_num ==0 ){
- close( ev->socket_fd );
- return ERROR;
- }
- else{
- //check ifoverflow
- if( ev->request_len > MAX_REQLEN-recv_num ){
- close( ev->socket_fd );
- clean_request( ev );
- }
- memcpy( ev->request + ev->request_len, in_buf, recv_num );
- ev->request_len += recv_num;
- if( recv_num == 2 && ( !memcmp( &in_buf[recv_num-2], "\r\n", 2 ) ) ){
- parse_request(ev);
- }
- }
- return recv_num;
- }
-
- int main(){
- int listen_fd = create_listen_fd( 3389 );
- int pid = fork_process( 3 );
- if( pid == 0 ){
- int accept_handles = 0;
- struct epoll_event ev, events[20];
- int epfd = epoll_create( 256 );
- int ev_s = 0;
-
- ev.data.fd = listen_fd;
- ev.events = EPOLLIN|EPOLLET;
- epoll_ctl( epfd, EPOLL_CTL_ADD, listen_fd, &ev );
- struct event_handle ev_handles[256];
- for( ;; ){
- ev_s = epoll_wait( epfd, events, 20, 500 );
- int i = 0;
- for( i = 0; i<ev_s; i++ ){
- if( events[i].data.fd == listen_fd ){
- if( accept_handles < MAX_PROCESS_CONN ){
- accept_handles++;
- int accept_fd = create_accept_fd( listen_fd );
- init_evhandle(&ev_handles[accept_handles],accept_fd,handle_hook_v2);
- ev.data.ptr = &ev_handles[accept_handles];
- ev.events = EPOLLIN|EPOLLET;
- epoll_ctl( epfd, EPOLL_CTL_ADD, accept_fd, &ev );
- }
- }
- else if( events[i].events&EPOLLIN ){
- EVENT_HANDLE current_handle = ( ( EH )( events[i].data.ptr ) )->handle;
- EH current_event = ( EH )( events[i].data.ptr );
- if( ( *current_handle )( current_event ) == 0 ){
- accept_handles--;
- }
- }
- else if( events[i].events&EPOLLOUT ){
- //need add write event process
- }
- }
- }
- }
- else{
- //manager the process
- int child_process_status;
- wait( &child_process_status );
- }
-
- return SUCCESS;
- }
阅读(1071) | 评论(0) | 转发(1) |