简单的回射服务器实现

SoforthHe

说明一下：内核比较老，epoll有惊群现象，据说在较新的内核上(2.6.24以后)，性能会有进一步提升，只是没有验证下，详见这个帖子：http://bbs.chinaunix.net/thread-1091745-3-1.html
你还在用JAVA吗？快回到C的阵营来吧。。。:wink:
16核CPU

webbench测试:

服务端：

客户端

ab测试:
1.非keepalive模式

服务端：

客户端：

2.Keepalive模式

服务端：

客户端：

源代码：
worker.c

1. #include <stdio.h>
   
2. #include <stdlib.h>
   
3. #include <string.h>
   
4. #include <pthread.h>
   
5. #include <sys/epoll.h>
   
6. #include <errno.h>
   
7. #include "thread.h"
   
8. #include "list.h"
   
9. 
   
10. #include <netinet/in.h>
    
11. #include <netinet/tcp.h>
    
12. #include <sys/socket.h>
    
13. #include <sys/types.h>
    
14. #include <sys/stat.h>
    
15. #include <sys/signal.h>
    
16. #include <sys/ioctl.h>
    
17. #include <fcntl.h>
    
18. 
    
19. 
    
20. typedef int BOOL;
    
21. 
    
22. typedef struct _WORKER {
    
23.         int epfd;
    
24.         int nextepfd;
    
25.         int waitevents;       
    
26. }WORKER, *PWORKER;
    
27. 
    
28. static int   g_thread_num;
    
29. static int   g_max_backlogs;
    
30. static int   g_max_events;
    
31. static int   g_listenfd;
    
32. static short g_listnport;
    
33. static BOOL  g_keepalive;
    
34. static BOOL  g_stop;
    
35. 
    
36. static inline void epoll_add(int epfd, int fd, int events) {
    
37.         struct epoll_event ev;
    
38.         ev.events = events | EPOLLERR | EPOLLET ;
    
39.         ev.data.fd = fd;
    
40.         epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev);       
    
41. }
    
42. static inline void epoll_mod(int epfd, int fd, int events) {
    
43.         struct epoll_event ev;
    
44.         ev.events = events | EPOLLERR | EPOLLET;
    
45.         ev.data.fd = fd;
    
46.         epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev);       
    
47. }
    
48. static inline void epoll_del(int epfd, int fd) {
    
49.         struct epoll_event ev;
    
50.         ev.data.fd = fd;
    
51.         epoll_ctl(epfd, EPOLL_CTL_DEL, fd, &ev);       
    
52. }
    
53. 
    
54. static void set_tcpnodelay(int sockfd, BOOL set){
    
55.         int v;
    
56.         struct linger linger = {0, 0};
    
57.         int optval = 1;
    
58.         if( set ) {
    
59.                 v=1;
    
60.                 setsockopt(sockfd, SOL_TCP, TCP_NODELAY, &v, sizeof(v));
    
61.                 v=0;
    
62.                 setsockopt(sockfd, SOL_TCP, TCP_CORK, &v, sizeof(v));
    
63. 
    
64.                 setsockopt(sockfd, SOL_SOCKET, SO_LINGER, (int *)&linger, sizeof(linger));
    
65.         }
    
66. }
    
67. 
    
68. static void set_nonblock(int sockfd, BOOL set){
    
69.         if(set) {
    
70.                 fcntl(sockfd, F_SETFL, fcntl(sockfd, F_GETFL)|O_NONBLOCK);
    
71.         }
    
72. }
    
73. 
    
74. static void set_reuseable(int sockfd, BOOL set) {
    
75.         int optval = 1;
    
76.         if(set) {
    
77.                 setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));
    
78.         }
    
79. }
    
80. 
    
81. void HandleRecv(int epfd, int clifd, int nextepfd, int keepalive) {
    
82.         int rc;
    
83.         char buf[4096] = { 0 };
    
84.         while(1) {
    
85.                 rc = recv(clifd, buf, sizeof(buf), MSG_DONTWAIT);
    
86.                 if( rc < 0 ) {
    
87.                         if( errno == EINTR ) {
    
88.                                 continue;
    
89.                         }else {
    
90.                                 break;
    
91.                         }
    
92.                 }else if( rc == 0 ) {
    
93.                         send(clifd, buf, rc, MSG_DONTWAIT|MSG_NOSIGNAL);
    
94.                         epoll_del(epfd, clifd);
    
95.                         close(clifd);
    
96.                 }else if( rc > 0 ) {
    
97.                         send(clifd, buf, rc, MSG_DONTWAIT|MSG_NOSIGNAL);
    
98.                         if(keepalive) {
    
99.                                 epoll_del(epfd, clifd);
    
100.                                 epoll_add(nextepfd, clifd, EPOLLIN);
     
101.                         }else {
     
102.                                 epoll_del(epfd, clifd);
     
103.                                 close(clifd);
     
104.                         }
     
105.                 }
     
106.         }
     
107. }
     
108. 
     
109. int worker_loop(void *para){
     
110.         PWORKER wk = (PWORKER)para;
     
111.         struct epoll_event *pev = NULL;
     
112.         struct sockaddr_in addr;
     
113.         int socklen = sizeof(addr);
     
114. 
     
115.         int i;
     
116.         int n;
     
117.         int clifd;
     
118.         int v = 0;       
     
119. 
     
120. 
     
121.         pev = (struct epoll_event*)malloc(sizeof(struct epoll_event) * g_max_events);
     
122. 
     
123.         while(!g_stop) {
     
124.                 n = epoll_wait(wk->epfd, pev, wk->waitevents, 1);
     
125.                 if( n <= 0 ) {
     
126.                         continue;
     
127.                 }
     
128. 
     
129.                 for( i = 0; i < n; ++i ) {
     
130.                         if(g_listenfd == pev[i].data.fd) {
     
131.                                 while(!g_stop){
     
132.                                         clifd = accept(g_listenfd, (struct sockaddr*)&addr, &socklen);
     
133. 
     
134.                                         if(clifd < 0 ) break;
     
135. 
     
136.                                         set_nonblock(clifd, 1);
     
137.                                         set_tcpnodelay(clifd, 1);
     
138. 
     
139.                                         epoll_add(wk->epfd, clifd, EPOLLIN);
     
140.                                         HandleRecv(wk->epfd, clifd, wk->nextepfd, g_keepalive);
     
141.                                 }
     
142.                         } else {
     
143.                                 if( pev[i].events & EPOLLIN ){
     
144.                                         HandleRecv(wk->epfd, pev[i].data.fd, wk->nextepfd, g_keepalive);
     
145.                                 }
     
146.                         }
     
147.                 }
     
148.         }
     
149. 
     
150.         free(pev);
     
151.         pthread_exit(NULL);
     
152. 
     
153. }
     
154. 
     
155. PWORKER init_worker(int threadnum, int maxevents, int listenfd){
     
156.         int i;
     
157.         char thread_name[THREAD_NAME_LEN];
     
158.         PWORKER wk;
     
159. 
     
160.         wk = (PWORKER)malloc(sizeof(WORKER)*threadnum);
     
161. 
     
162.         wk[0].epfd = epoll_create(maxevents);               
     
163.         wk[0].waitevents = maxevents-(maxevents/threadnum*(threadnum-1));
     
164.         epoll_add(wk[0].epfd, listenfd, EPOLLIN);
     
165.         snprintf(thread_name, sizeof(thread_name), "Worker%d", 0);
     
166.         register_thread(thread_name, worker_loop, &wk[0], 0);
     
167. 
     
168.         for( i = 1; i < threadnum; ++i ) {
     
169. 
     
170.                 wk[i].epfd = epoll_create(maxevents);
     
171.                 wk[i].waitevents = maxevents/threadnum;
     
172.                 wk[i-1].nextepfd = wk[i].epfd;
     
173.                 epoll_add(wk[i].epfd, listenfd, EPOLLIN);
     
174.                 snprintf(thread_name, sizeof(thread_name), "worker%d", i);
     
175.                 register_thread(thread_name, worker_loop, &wk[i], 0);
     
176.         }
     
177. 
     
178.         wk[i-1].nextepfd = wk[0].epfd;
     
179. 
     
180.         return wk;
     
181. }
     
182. 
     
183. void fini_worker(PWORKER wk){
     
184.         free(wk);
     
185. }
     
186. 
     
187. int make_socket(uint16_t port, int backlogs) {
     
188.         struct sockaddr_in addr;       
     
189.         int s;       
     
190.         int optval = 1;
     
191. 
     
192.         s = socket(AF_INET, SOCK_STREAM, 0);
     
193.         if(s < 0) {
     
194.                 perror("socket");
     
195.                 return -1;
     
196.         }
     
197. 
     
198.         set_reuseable(s, 1);
     
199.         set_nonblock(s, 1);
     
200.         set_tcpnodelay(s, 1);
     
201. 
     
202.         memset(&addr, 0, sizeof(struct sockaddr_in));
     
203.         addr.sin_family = AF_INET;
     
204.         addr.sin_addr.s_addr = 0;
     
205.         addr.sin_port = htons(port);
     
206. 
     
207.         if(bind(s, (struct sockaddr*)&addr, sizeof(addr))<0){
     
208.                 perror("bind");
     
209.                 return -1;
     
210.         }
     
211. 
     
212.         if( listen(s, backlogs) < 0 ) {
     
213.                 perror("listen");
     
214.                 return -1;
     
215.         }
     
216. 
     
217.         setsockopt(s, SOL_TCP, TCP_DEFER_ACCEPT, &optval, sizeof(optval));
     
218. 
     
219.         return s;
     
220. }
     
221. 
     
222. static void sigusr1_handler(int signo) {
     
223.         g_stop = 1;
     
224. }
     
225. 
     
226. void set_stop_signal(){
     
227.         struct sigaction sa;
     
228.         sigset_t sset;
     
229.         memset(&sa, 0, sizeof(sa));
     
230.         sa.sa_handler = sigusr1_handler;
     
231.         sigaction(SIGUSR1, &sa, NULL);
     
232. }
     
233. int main(int argc, char** argv) {
     
234.         PWORKER wk;
     
235.         int status = -1;
     
236. 
     
237.         if( argc < 3 ) {
     
238.                 printf("%s <port> <keepalive>\n", argv[0]);
     
239.                 return -1;
     
240.         }
     
241. 
     
242.         set_stop_signal();
     
243. 
     
244.         g_stop = 0;
     
245.         g_keepalive = atoi(argv[2]);
     
246.         g_listnport = atoi(argv[1]);
     
247.         g_max_backlogs = 10000;
     
248.         g_thread_num = 16;
     
249.         g_max_events = 65535;
     
250.         g_listenfd = make_socket(g_listnport, g_max_backlogs);
     
251.         wk = init_worker(g_thread_num, g_max_events, g_listenfd);
     
252. 
     
253.         start_threads();
     
254. 
     
255.         while(!g_stop) {
     
256.                 sleep(1);
     
257.         }
     
258. 
     
259.         sleep(1);
     
260.         stop_threads();       
     
261.         fini_worker(wk);
     
262.         close(g_listenfd);       
     
263. 
     
264.         return 0;
     
265. }
     
266. 
     

复制代码

lhy0416

到底是给什么公司做广告的？

SoforthHe

源代码.rar (6.72 KB, 下载次数: 41)

2011-12-05 17:05 上传

点击文件名下载附件

SoforthHe

其实只是闲得蛋疼而已。。。