从头编写高性能服务程序(二)

从头编写高性能服务程序(二)

[转]http://blog.chinaunix.net/u/6889/showart.php?id=2296857

从头编写高性能服务程序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)

来进行处理

下载:

1. #include <sys/socket.h>
2. #include <sys/wait.h>
3. #include <netinet/in.h>
4. #include <sys/epoll.h>
5. #include <sys/sendfile.h> 
6. #include <unistd.h>
7. #include <stdio.h>
8. #include <stdlib.h>
9. #include <string.h>
10. #include <strings.h>
11. #include <fcntl.h>
13. int create_listen_fd(int port){
14.     int listen_fd;
15.     struct sockaddr_in my_addr;
16.     if ((listen_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1){
17.         perror("create socket error");
18.         exit(1);
19.     }
20.     int flag;
21.     if (setsockopt(listen_fd,SOL_SOCKET,SO_REUSEADDR
22.     ,(char *)&flag,sizeof(flag)) == -1){
23.         perror("setsockopt error");
24.     }
25.     int flags = fcntl(listen_fd, F_GETFL, 0);
26.     fcntl(listen_fd, F_SETFL, flags|O_NONBLOCK);
27.     my_addr.sin_family = AF_INET;
28.     my_addr.sin_port = htons(port);
29.     my_addr.sin_addr.s_addr = INADDR_ANY;
30.     bzero(&(my_addr.sin_zero), 8);
31.     if (bind(listen_fd, (struct sockaddr *)&my_addr,
32.     sizeof(struct sockaddr_in)) == -1) {
33.         perror("bind error");
34.         exit(1);
35.     }
36.     if (listen(listen_fd,1) == -1){
37.         perror("listen error");
38.         exit(1);
39.     }
40.     return listen_fd;
41. }
43. int create_accept_fd(int listen_fd){
44.     int addr_len = sizeof( struct sockaddr_in );
45.     struct sockaddr_in remote_addr;
46.     int accept_fd = accept( listen_fd,
47.         (struct sockaddr *)&remote_addr, &addr_len );
48.     int flags = fcntl(accept_fd, F_GETFL, 0);
49.     fcntl(accept_fd, F_SETFL, flags|O_NONBLOCK);
50.     return accept_fd;
51. }
53. int fork_process(int process_num){
54.     int i;
55.     int pid=-1;
56.     for(i = 0; i < process_num; i++){
57.         if(pid != 0){
58.             pid = fork();
59.         }
60.     }
61.     return pid;
62. }
64. int handle_normal(int socket_fd){
65.     char in_buf[1024];
66.     memset(in_buf, 0, 1024);
67.     int recv_num = recv( socket_fd, &in_buf, 1024, 0 );
68.     if( recv_num ==0 ){
69.         close(socket_fd);
70.         printf("ProcessID:%d,EPOLLIN,fd:%d,closed\n", getpid(), socket_fd);
71.     }
72.     else{
73.         printf("ProcessID:%d,EPOLLIN,fd:%d,recv:%s\n", getpid(), socket_fd, in_buf);
74.     }
75.     return recv_num;
76. }
78. int handle_hook(int socket_fd){
79.     char in_buf[1024];
80.     memset(in_buf, 0, 1024);
81.     int recv_num = recv( socket_fd, &in_buf, 1024, 0 );
82.     if( recv_num ==0 ){
83.         close(socket_fd);
84.         printf("ProcessID:%d,EPOLLIN,fd:%d,closed\n", getpid(), socket_fd);
85.     }
86.     else{
87.         printf("ProcessID:%d,EPOLLIN,fd:%d,recv_num:%d;recv:", getpid(), socket_fd, recv_num);
88.         for (int i = 0; i<recv_num; i++){
89.         printf("%02x ",in_buf[i]);
90.         }
91.         printf("\n");
92.     }
93.     return recv_num;
94. }
96. struct event_handle{
97.     int fd;
98.     int (* handle)(int fd);
99. };
100. typedef int (* EVENT_HANDLE)(int);
101. typedef struct event_handle * EH;
103. int main(){
104.     int listen_fd = create_listen_fd(3389);
105.     int pid = fork_process(3);
106.     if(pid == 0){
107.         int accept_handles = 0;
108.         struct epoll_event ev,events[20];
109.         int epfd = epoll_create(256);
110.         int ev_s = 0;
111.        
112.         ev.data.fd = listen_fd;
113.         ev.events = EPOLLIN|EPOLLET;
114.         epoll_ctl(epfd,EPOLL_CTL_ADD,listen_fd,&ev);
115.         struct event_handle ev_handles[256];
116.         for(;;){
117.             ev_s = epoll_wait( epfd,events, 20, 500 );
118.             int i = 0;
119.             for(i = 0; i<ev_s; i++){
120.                 if(events[i].data.fd == listen_fd){
121.                     int max_process_accept = 3;
122.                     if(accept_handles < max_process_accept){
123.                         accept_handles++;
124.                         int accept_fd = create_accept_fd(listen_fd);
125.                         ev_handles[accept_handles].fd = accept_fd;
126.                         ev_handles[accept_handles].handle = handle_hook;
127.                         ev.data.ptr = &ev_handles[accept_handles];
128.                         ev.events = EPOLLIN|EPOLLET;
129.                         epoll_ctl(epfd,EPOLL_CTL_ADD,accept_fd,&ev);
130.                         printf("ProcessID:%d,EPOLLIN,fd:%d,accept:%d\n", getpid(), listen_fd, accept_fd);
131.                     }
132.                 }
133.                 else if(events[i].events&EPOLLIN){
134.                     EVENT_HANDLE current_handle = ((EH)(events[i].data.ptr))->handle;
135.                     int current_fd = ((EH)(events[i].data.ptr))->fd;
136.                     if( (*current_handle)(current_fd)  == 0){ 
137.                         accept_handles--;
138.                     }
139.                 }
140.                 else if(events[i].events&EPOLLOUT){
141.                     //need add write event process
142.                 }
143.             }
144.         }
145.     }
146.     else{
147.         //manager the process
148.         int child_process_status;
149.         wait(&child_process_status);
150.     }
151.    
152.     return 0;
153. }

从头编写高性能服务程序10-请求解析

最终的形态基本上在上次的结构中就定型了
当然有些细节需要完善
不过基本上用这个结构来写service已经是OK了
那么现在就是继续细化这个结构.用来写个比较靠近实际的应用

有了链接的管理
接下来就是对通讯协议的实现
由于是从头开始写
所以协议也由我们自己来实现
先是对请求的解析
从客户端telnet传送过来的数据
回行是用/r/n结尾的
所以我们不停的接受数据
然后判断数据的最后是否是/r/n
如果是的话.就把它和以前的数据一起拼接起来
然后调用请求分析来解析指令

在event_handle结构中
我们加入了command数组
用来存放每次传输过来的数据
直至遇到以/r/n结尾的数据.然后拼接起来,输出,再清空这个数组
从头再接受新的指令

由于使用了epoll和非阻塞accept_fd
所以每次接受到的数据是零散的
需要将每次recv的数据连续的拼接到一个变量中
这就是command数组存在的理由
而command_pos用来保存的是每次拼接后数组的实际存放数据的量
也可以认为是最后一个数据所在数组中的位置
便于下次拼接

下载:

1. #include <sys/socket.h>
2. #include <sys/wait.h>
3. #include <netinet/in.h>
4. #include <sys/epoll.h>
5. #include <sys/sendfile.h> 
6. #include <unistd.h>
7. #include <stdio.h>
8. #include <stdlib.h>
9. #include <string.h>
10. #include <strings.h>
11. #include <fcntl.h>
13. typedef struct event_handle{
14.     int fd;
15.     int ( * handle )( struct event_handle * ev );
16.     char command[1024];
17.     int command_pos;
18. } EV,* EH;
19. typedef int ( * EVENT_HANDLE )( struct event_handle * ev );
22. int create_listen_fd( int port ){
23.     int listen_fd;
24.     struct sockaddr_in my_addr;
25.     if ( ( listen_fd = socket( AF_INET, SOCK_STREAM, 0 ) ) == -1 ){
26.         perror( "create socket error" );
27.         exit( 1 );
28.     }
29.     int flag;
30.     if ( setsockopt( listen_fd, SOL_SOCKET, SO_REUSEADDR
31.     , ( char * )&flag, sizeof( flag ) ) == -1 ){
32.         perror( "setsockopt error" );
33.     }
34.     int flags = fcntl( listen_fd, F_GETFL, 0 );
35.     fcntl( listen_fd, F_SETFL, flags|O_NONBLOCK );
36.     my_addr.sin_family = AF_INET;
37.     my_addr.sin_port = htons( port );
38.     my_addr.sin_addr.s_addr = INADDR_ANY;
39.     bzero( &( my_addr.sin_zero ), 8 );
40.     if ( bind( listen_fd, ( struct sockaddr * )&my_addr,
41.     sizeof( struct sockaddr_in ) ) == -1 ) {
42.         perror( "bind error" );
43.         exit( 1 );
44.     }
45.     if ( listen( listen_fd, 1 ) == -1 ){
46.         perror( "listen error" );
47.         exit( 1 );
48.     }
49.     return listen_fd;
50. }
52. int create_accept_fd( int listen_fd ){
53.     int addr_len = sizeof( struct sockaddr_in );
54.     struct sockaddr_in remote_addr;
55.     int accept_fd = accept( listen_fd,
56.         ( struct sockaddr * )&remote_addr, &addr_len );
57.     int flags = fcntl( accept_fd, F_GETFL, 0 );
58.     fcntl( accept_fd, F_SETFL, flags|O_NONBLOCK );
59.     return accept_fd;
60. }
62. int fork_process( int process_num ){
63.     int i;
64.     int pid=-1;
65.     for( i = 0; i < process_num; i++ ){
66.         if( pid != 0 ){
67.             pid = fork();
68.         }
69.     }
70.     return pid;
71. }
73. int handle_hook_v2( EH ev ){
74.     char in_buf[1024];
75.     memset( in_buf, 0, 1024 );
76.     int recv_num = recv( ev->fd, &in_buf, 1024, 0 );
77.     if( recv_num ==0 ){
78.         printf( "ProcessID:%d, EPOLLIN, fd:%d, closed\n", getpid(), ev->fd );
79.         printf( "  recved:%s\n", ev->command );
80.         close( ev->fd );
81.     }
82.     else{
83.         printf( "ProcessID:%d, EPOLLIN, fd:%d, recv_num:%d;recv:", getpid(), ev->fd, recv_num );
84.         int i;
85.         for( i = 0; i<recv_num; i++ ){
86.             printf( "%02x ", in_buf[i] );
87.         }
88.         printf( "\n" );
89.         memcpy( ev->command + ev->command_pos, in_buf, recv_num );
90.         ev->command_pos += recv_num;
91.         if( recv_num == 2 && ( !memcmp( &in_buf[recv_num-2], "\r\n", 2 ) ) ){
92.             printf( "  recved:%s\n", ev->command );
93.             memset( ev->command, 0, 1024 );
94.             ev->command_pos = 0;
95.         }
96.     }
97.     return recv_num;
98. }
102. int main(){
103.     int listen_fd = create_listen_fd( 3389 );
104.     int pid = fork_process( 3 );
105.     if( pid == 0 ){
106.         int accept_handles = 0;
107.         struct epoll_event ev, events[20];
108.         int epfd = epoll_create( 256 );
109.         int ev_s = 0;
110.        
111.         ev.data.fd = listen_fd;
112.         ev.events = EPOLLIN|EPOLLET;
113.         epoll_ctl( epfd, EPOLL_CTL_ADD, listen_fd, &ev );
114.         struct event_handle ev_handles[256];
115.         for( ;; ){
116.             ev_s = epoll_wait( epfd, events, 20, 500 );
117.             int i = 0;
118.             for( i = 0; i<ev_s; i++ ){
119.                 if( events[i].data.fd == listen_fd ){
120.                     int max_process_accept = 3;
121.                     if( accept_handles < max_process_accept ){
122.                         accept_handles++;
123.                         int accept_fd = create_accept_fd( listen_fd );
124.                         ev_handles[accept_handles].fd = accept_fd;
125.                         ev_handles[accept_handles].handle = handle_hook_v2;
126.                         ev_handles[accept_handles].command_pos = 0;
127.                         memset( ev_handles[accept_handles].command, 0, 1024 );
128.                         ev.data.ptr = &ev_handles[accept_handles];
129.                         ev.events = EPOLLIN|EPOLLET;
130.                         epoll_ctl( epfd, EPOLL_CTL_ADD, accept_fd, &ev );
131.                         printf( "ProcessID:%d, EPOLLIN, fd:%d, accept:%d\n", getpid(), listen_fd, accept_fd );
132.                     }
133.                 }
134.                 else if( events[i].events&EPOLLIN ){
135.                     EVENT_HANDLE current_handle = ( ( EH )( events[i].data.ptr ) )->handle;
136.                     EH current_event = ( EH )( events[i].data.ptr );
137.                     if( ( *current_handle )( current_event )  == 0 ){ 
138.                         accept_handles--;
139.                     }
140.                 }
141.                 else if( events[i].events&EPOLLOUT ){
142.                     //need add write event process
143.                 }
144.             }
145.         }
146.     }
147.     else{
148.         //manager the process
149.         int child_process_status;
150.         wait( &child_process_status );
151.     }
152.    
153.     return 0;
154. }

从头编写高性能服务程序11-指令处理&sendfile

实现命令的获取之后
现在是增加对command的解析以及对应的反馈
为了做些稍微有意义的事情.我把这个service做的工作定位在以下内容
1.查询文件大小
2.返回文件内容
3.删除文件

协议的格式如下
请求模式:文件名称
例如
1:new.txt

请求模式见源代码中的宏定义

这次代码也对原来的程序作了一定的修改
从丑陋的代码渐渐修改.
希望在最终程序完成的时候
能够有比较好的代码风格

这个版本中有一个bug
就是sendfile只是调用一次
而实际上如果是较大的文件
需要在判断EPLLOUT之后不停的sendfile
直到EAGAIN遇见accept_fd阻塞为止
这样直至下次EPOLLOUT发生
再从上次暂停的位置继续发送
下一个版本中将会有这个BUG的修正

下载:

1. #include <sys/socket.h>
2. #include <sys/wait.h>
3. #include <netinet/in.h>
4. #include <netinet/tcp.h>
5. #include <sys/epoll.h>
6. #include <sys/sendfile.h> 
7. #include <sys/stat.h>
8. #include <unistd.h>
9. #include <stdio.h>
10. #include <stdlib.h>
11. #include <string.h>
12. #include <strings.h>
13. #include <fcntl.h>
15. #define HANDLE_INFO   1
16. #define HANDLE_SEND   2
17. #define HANDLE_DEL    3
18. #define HANDLE_CLOSE  4
20. #define MAX_REQLEN          1024
21. #define MAX_PROCESS_CONN    3
22. #define FIN_CHAR            0x00
23. #define SUCCESS  0
24. #define ERROR   -1
26. typedef struct event_handle{
27.     int socket_fd;
28.     int file_fd;
29.     char request[MAX_REQLEN];
30.     int request_len;
31.     int ( * handle )( struct event_handle * ev );
32.     int handle_method;
34. } EV,* EH;
35. typedef int ( * EVENT_HANDLE )( struct event_handle * ev );
37. int create_listen_fd( int port ){
38.     int listen_fd;
39.     struct sockaddr_in my_addr;
40.     if( ( listen_fd = socket( AF_INET, SOCK_STREAM, 0 ) ) == -1 ){
41.         perror( "create socket error" );
42.         exit( 1 );
43.     }
44.     int flag;
45.     int olen = sizeof(int);
46.     if( setsockopt( listen_fd, SOL_SOCKET, SO_REUSEADDR
47.                         , (const void *)&flag, olen ) == -1 ){
48.         perror( "setsockopt error" );
49.     }
50.     flag = 1;
51.     if( setsockopt( listen_fd, IPPROTO_TCP, TCP_DEFER_ACCEPT, &flag, olen ) == -1 ){
52.         perror( "setsockopt error" );
53.     }
54.     int flags = fcntl( listen_fd, F_GETFL, 0 );
55.     fcntl( listen_fd, F_SETFL, flags|O_NONBLOCK );
56.     my_addr.sin_family = AF_INET;
57.     my_addr.sin_port = htons( port );
58.     my_addr.sin_addr.s_addr = INADDR_ANY;
59.     bzero( &( my_addr.sin_zero ), 8 );
60.     if( bind( listen_fd, ( struct sockaddr * )&my_addr,
61.     sizeof( struct sockaddr_in ) ) == -1 ) {
62.         perror( "bind error" );
63.         exit( 1 );
64.     }
65.     if( listen( listen_fd, 1 ) == -1 ){
66.         perror( "listen error" );
67.         exit( 1 );
68.     }
69.     return listen_fd;
70. }
72. int create_accept_fd( int listen_fd ){
73.     int addr_len = sizeof( struct sockaddr_in );
74.     struct sockaddr_in remote_addr;
75.     int accept_fd = accept( listen_fd,
76.         ( struct sockaddr * )&remote_addr, &addr_len );
77.     int flags = fcntl( accept_fd, F_GETFL, 0 );
78.     fcntl( accept_fd, F_SETFL, flags|O_NONBLOCK );
79.     return accept_fd;
80. }
82. int fork_process( int process_num ){
83.     int i;
84.     int pid=-1;
85.     for( i = 0; i < process_num; i++ ){
86.         if( pid != 0 ){
87.             pid = fork();
88.         }
89.     }
90.     return pid;
91. }
93. int init_evhandle(EH ev,int socket_fd,EVENT_HANDLE handle){
94.     ev->socket_fd = socket_fd;
95.     ev->handle = handle;
96.     ev->request_len = 0;
97.     ev->handle_method = 0;
98.     memset( ev->request, 0, 1024 );
99. }
100. //accept->accept_queue->request->request_queue->output->output_queue
101. //multi process sendfile
102. int parse_request(EH ev){
103.     ev->request_len--;
104.     *( ev->request + ev->request_len - 1 ) = 0x00;
105.     int i;
106.     for( i=0; i<ev->request_len; i++ ){
107.         if( ev->request[i] == ':' ){
108.             ev->request_len = ev->request_len-i-1;
109.             char temp[MAX_REQLEN];
110.             memcpy( temp, ev->request, i );
111.             ev->handle_method = atoi( temp );
112.             memcpy( temp, ev->request+i+1, ev->request_len );
113.             memcpy( ev->request, temp, ev->request_len );
114.             break;
115.         }
116.     }
117.     handle_request( ev );
118.     return SUCCESS;
119. }
121. int handle_request(EH ev){
122.     struct stat file_info;
123.     switch( ev->handle_method ){
124.         case HANDLE_INFO:
125.             ev->file_fd = open( ev->request, O_RDONLY );
126.             if( ev->file_fd == -1 ){
127.                 send( ev->socket_fd, "open file failed\n", strlen("open file failed\n"), 0 );
128.                 return -1;
129.             }
130.             fstat(ev->file_fd, &file_info);
131.             char info[MAX_REQLEN];
132.             sprintf(info,"file len:%d\n",file_info.st_size);
133.             send( ev->socket_fd, info, strlen( info ), 0 );
134.             break;
135.         case HANDLE_SEND:
136.             ev->file_fd = open( ev->request, O_RDONLY );
137.             if( ev->file_fd == -1 ){
138.                 send( ev->socket_fd, "open file failed\n", strlen("open file failed\n"), 0 );
139.                 return -1;
140.             }
141.             fstat(ev->file_fd, &file_info);
142.             sendfile( ev->socket_fd, ev->file_fd, 0, file_info.st_size );
143.             break;
144.         case HANDLE_DEL:
145.             break;
146.         case HANDLE_CLOSE:
147.             break;
148.     }
149.     finish_request( ev );
150.     return SUCCESS;
151. }
153. int finish_request(EH ev){
154.     close(ev->socket_fd);
155.     close(ev->file_fd);
156.     ev->handle_method = -1;
157.     clean_request( ev );
158.     return SUCCESS;
159. }
161. int clean_request(EH ev){
162.     memset( ev->request, 0, MAX_REQLEN );
163.     ev->request_len = 0;
164. }
166. int handle_hook_v2( EH ev ){
167.     char in_buf[MAX_REQLEN];
168.     memset( in_buf, 0, MAX_REQLEN );
169.     int recv_num = recv( ev->socket_fd, &in_buf, MAX_REQLEN, 0 );
170.     if( recv_num ==0 ){
171.         close( ev->socket_fd );
172.         return ERROR;
173.     }
174.     else{
175.         //check ifoverflow
176.         if( ev->request_len > MAX_REQLEN-recv_num ){
177.             close( ev->socket_fd );
178.             clean_request( ev );
179.         }
180.         memcpy( ev->request + ev->request_len, in_buf, recv_num );
181.         ev->request_len += recv_num;
182.         if( recv_num == 2 && ( !memcmp( &in_buf[recv_num-2], "\r\n", 2 ) ) ){
183.             parse_request(ev);
184.         }
185.     }
186.     return recv_num;
187. }
189. int main(){
190.     int listen_fd = create_listen_fd( 3389 );
191.     int pid = fork_process( 3 );
192.     if( pid == 0 ){
193.         int accept_handles = 0;
194.         struct epoll_event ev, events[20];
195.         int epfd = epoll_create( 256 );
196.         int ev_s = 0;
197.        
198.         ev.data.fd = listen_fd;
199.         ev.events = EPOLLIN|EPOLLET;
200.         epoll_ctl( epfd, EPOLL_CTL_ADD, listen_fd, &ev );
201.         struct event_handle ev_handles[256];
202.         for( ;; ){
203.             ev_s = epoll_wait( epfd, events, 20, 500 );
204.             int i = 0;
205.             for( i = 0; i<ev_s; i++ ){
206.                 if( events[i].data.fd == listen_fd ){
207.                     if( accept_handles < MAX_PROCESS_CONN ){
208.                         accept_handles++;
209.                         int accept_fd = create_accept_fd( listen_fd );
210.                         init_evhandle(&ev_handles[accept_handles],accept_fd,handle_hook_v2);
211.                         ev.data.ptr = &ev_handles[accept_handles];
212.                         ev.events = EPOLLIN|EPOLLET;
213.                         epoll_ctl( epfd, EPOLL_CTL_ADD, accept_fd, &ev );
214.                     }
215.                 }
216.                 else if( events[i].events&EPOLLIN ){
217.                     EVENT_HANDLE current_handle = ( ( EH )( events[i].data.ptr ) )->handle;
218.                     EH current_event = ( EH )( events[i].data.ptr );
219.                     if( ( *current_handle )( current_event )  == 0 ){ 
220.                         accept_handles--;
221.                     }
222.                 }
223.                 else if( events[i].events&EPOLLOUT ){
224.                     //need add write event process
225.                 }
226.             }
227.         }
228.     }
229.     else{
230.         //manager the process
231.         int child_process_status;
232.         wait( &child_process_status );
233.     }
234.    
235.     return SUCCESS;
236. }