非阻塞IO 判断就绪状态

teng0210

非阻塞IO读写,  如何判断轮询fd就绪. 如果不使用select (poll), epoll

ubuntuer

不用select epoll你就自己去读下看成功不成功呗！

teng0210

如果自己轮询检测
如何判断出错和断线呢
使用select时,
如果read  返回0 是断线, <0是出错
如果设置了非阻塞IO 自己轮询的话  , 出错也是这种情况么
感觉自己这样轮询判断似乎不如select的效率高

loveye2009

大家有epoll的实例吗？可以参考学习下吗？网上的例子很不好。

jiufei19

回复 4# loveye2009


    CU上的例子不够吗？

teng0210

回复 4# loveye2009

自己写的, C++的例子, 简单的http server, 访问出现hello world, 使用epoll

1. #include<iostream>
   
2. #include<string>
   
3. #include<map>
   
4. #include<sys/types.h>
   
5. #include<sys/socket.h>
   
6. #include<arpa/inet.h>
   
7. #include<unistd.h>
   
8. #include<fcntl.h>
   
9. #include<signal.h>
   
10. #include<sys/epoll.h>
    
11. #include<string.h>
    
12. #include<error.h>
    
13. #include<errno.h>
    
14. #include<stdlib.h>
    
15. #include<sys/resource.h>
    
16. #include<pthread.h>
    
17. #include<time.h>
    
18. 
    
19. using namespace std;
    
20. 
    
21. #define LISTENQ 64
    
22. #define MAX_EVENTS 10000
    
23. #define BUFSIZE 8192
    
24. 
    
25. static unsigned long countacp=0;
    
26. static unsigned long countin=0;
    
27. static unsigned long countout=0;
    
28. static unsigned long counttimeout=0;
    
29. 
    
30. struct client_t{
    
31.         int fd;
    
32.         int stat;
    
33.         string ip;
    
34.         string read_buf;
    
35.         string write_buf;
    
36.         time_t last_alive_time;
    
37.         int is_timeout;
    
38.         client_t(){
    
39.                 is_timeout = 0;
    
40.                 stat = 0;
    
41.         }
    
42. };
    
43. 
    
44. map<int, struct client_t *> fds;
    
45. pthread_mutex_t fds_mutex = PTHREAD_MUTEX_INITIALIZER;
    
46. 
    
47. static string int2str(int i){
    
48.         char buf[32];
    
49.         memset(buf, 0, 32);
    
50.         sprintf(buf, "%d", i);
    
51.         return string(buf);
    
52. }
    
53. 
    
54. //initialize server socket
    
55. //正常返回fd,否则返回-1
    
56. static int
    
57. init_servsock()
    
58. {
    
59.         int fd;
    
60.         struct sockaddr_in serveraddr;
    
61.         //socket
    
62.         if((fd=socket(AF_INET, SOCK_STREAM, 0)) < 0){
    
63.                 perror("socket");
    
64.                 return -1;
    
65.         }
    
66. 
    
67.         //可重用
    
68.         socklen_t reuse=1;
    
69.         if(setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse))<0){ //否则程序挂掉马上重启,会报端口不可用
    
70.                 perror("setsockopt()");
    
71.         }
    
72. 
    
73.         serveraddr.sin_family = AF_INET;
    
74.         serveraddr.sin_port = htons(4000);
    
75.         serveraddr.sin_addr.s_addr = htonl(INADDR_ANY);
    
76.         bzero(&(serveraddr.sin_zero), 8);
    
77.         //bind
    
78.         if(bind(fd,(struct sockaddr *)&serveraddr, sizeof(sockaddr)) < 0){
    
79.                 perror("bind");
    
80.                 return -1;
    
81.         }
    
82.         //listen
    
83.         if(listen(fd, LISTENQ)){
    
84.                 perror("listen");
    
85.                 return -1;
    
86.         }
    
87. 
    
88.         return fd;
    
89. }
    
90. 
    
91. static ssize_t
    
92. writen(int fd, const char *vptr, size_t n)
    
93. {
    
94.         size_t nleft;
    
95.         ssize_t nwritten;
    
96.         const char *ptr;
    
97. 
    
98.         ptr = vptr;
    
99.         nleft = n;
    
100.         while(nleft > 0){
     
101.                 if((nwritten = write(fd, ptr, nleft)) <= 0)
     
102.                         return(nwritten); /* error */
     
103. 
     
104.                 nleft -= nwritten;
     
105.                 ptr += nwritten;
     
106.         }
     
107.         return(n);
     
108. }
     
109. 
     
110. static ssize_t
     
111. readn(int fd, char *vptr, size_t n)
     
112. {
     
113.         size_t nleft;
     
114.         ssize_t nread;
     
115.         char *ptr;
     
116. 
     
117.         ptr = vptr;
     
118.         nleft = n;
     
119.         while(nleft > 0){
     
120.                 if((nread = read(fd, ptr, nleft) < 0))
     
121.                         return(nread);        /* error, return < 0 */
     
122.                 else if(nread ==0 ){
     
123.                         break;                /* EOF */
     
124.                 }
     
125. 
     
126.                 nleft -= nread;
     
127.                 ptr += nread;
     
128.         }
     
129.         return(n - nleft);        /* return >= 0 */
     
130. }
     
131. 
     
132. static int
     
133. set_fl(int fd, int flags)
     
134. {
     
135.         int val;
     
136. 
     
137.         if((val = fcntl(fd,F_GETFL,0)) < 0){
     
138.                 printf("fcntl F_GETFL error\n");
     
139.                 return -1;
     
140.         }
     
141. 
     
142.         val |=flags;
     
143. 
     
144.         if(fcntl(fd,F_SETFL,val) < 0){
     
145.                 printf("fcntl F_SETFL error\n");
     
146.                 return -1;
     
147.         }
     
148. 
     
149.         return 0;
     
150. }
     
151. 
     
152. //超时检测线程函数
     
153. static void *
     
154. check_time_out(void * arg){
     
155.         int epollfd=*((int *)&arg);
     
156.         printf("%d\n",epollfd);
     
157.         while(1){
     
158.                 map<int,client_t *>::iterator it = fds.begin();
     
159.                 time_t tm;
     
160.                 time_t run_tm;
     
161.                 time(&tm);
     
162.                 for(; it!=fds.end(); ++it){
     
163.                         if(it->second != NULL){
     
164.                                 run_tm = tm - it->second->last_alive_time;
     
165.                                 if(run_tm >= 30){
     
166.                                         pthread_mutex_lock(&fds_mutex);
     
167.                                         printf("fd:%d,timeout\n",it->second->fd);
     
168.                                         fflush(NULL);
     
169.                                         //counttimeout++;
     
170.                                         epoll_ctl(epollfd, EPOLL_CTL_DEL, it->second->fd, NULL);
     
171.                                         shutdown(it->second->fd, SHUT_RDWR);
     
172.                                         close(it->second->fd);
     
173.                                         delete it->second;
     
174.                                         it->second = NULL;
     
175.                                         pthread_mutex_unlock(&fds_mutex);
     
176.                                 }
     
177.                         }
     
178.                 }
     
179.                 sleep(2);
     
180.         }
     
181. }
     
182. 
     
183. int
     
184. main()
     
185. {
     
186.         struct epoll_event ev, events[MAX_EVENTS];
     
187.         int listen_sock;
     
188.         int epollfd;
     
189.         int nfds;
     
190.         int err;
     
191. 
     
192.         //更改系统打开文件限制
     
193.         struct rlimit limit;
     
194.         limit.rlim_cur = 8192;//getdtablesize();//=1024
     
195.         limit.rlim_max = limit.rlim_cur;
     
196.         if(setrlimit(RLIMIT_NOFILE, &limit) < 0){
     
197.                 perror("setrlimit");
     
198.                 exit(1);
     
199.         }
     
200.        
     
201. 
     
202.         epollfd = epoll_create(512);
     
203.         if(epollfd == -1){
     
204.                 perror("epoll_create");
     
205.                 exit(1);
     
206.         }
     
207. 
     
208.         //创建超时检测线程
     
209.         pthread_t pid;
     
210.         err = pthread_create(&pid, NULL, check_time_out, (void *)epollfd);
     
211.         if(err !=0 ){
     
212.                 printf("创建超时检测线程失败！\n");
     
213.                 exit(1);
     
214.         }
     
215. 
     
216.         if((listen_sock = init_servsock())<0){
     
217.                 exit(1);
     
218.         }
     
219. 
     
220.         //设置为非阻塞
     
221.         if(set_fl(listen_sock,O_NONBLOCK) < 0){
     
222.                 exit(1);
     
223.         }
     
224. 
     
225.         //忽略信号
     
226.         signal(SIGPIPE,SIG_IGN);
     
227. 
     
228.         //daemon
     
229.         //daemon(0,0);
     
230. 
     
231.         ev.events = EPOLLIN;
     
232.         ev.data.fd = listen_sock;
     
233.         if(epoll_ctl(epollfd, EPOLL_CTL_ADD, listen_sock, &ev) == -1){
     
234.                 perror("epoll_ctl:listen_sock");
     
235.                 exit(1);
     
236.         }
     
237. 
     
238. 
     
239. 
     
240.         while(1){
     
241.                 //printf("accept:%u\n",countacp);
     
242.                 //printf("in:%u\n", countin);
     
243.                 //printf("out:%u,\n", countout);
     
244.                 //printf("timeout:%u\n", counttimeout);
     
245.                 //printf("\n");
     
246.                 //fflush(NULL);
     
247.                 nfds = epoll_wait(epollfd, events, MAX_EVENTS, -1);
     
248.                 if(nfds == -1){
     
249.                         perror("epoll_wait");
     
250.                         exit(1);
     
251.                 }
     
252. 
     
253.                 int n;
     
254.                 for(n = 0; n < nfds; ++n){
     
255.                         if(events[n].data.fd == listen_sock){
     
256.                                 struct sockaddr_in addr_in;
     
257.                                 size_t addrlen = sizeof(struct sockaddr_in);
     
258.                                 pthread_mutex_lock(&fds_mutex);
     
259.                                 int conn_sock = accept(listen_sock, (struct sockaddr *)&addr_in, &addrlen);
     
260.                                 time_t tm=time(NULL);
     
261.                                 char tbuf[32];
     
262.                                 memset(tbuf, 0, 32);
     
263.                                 strftime(tbuf,32,"%Y-%m-%d %H:%M:%S", localtime(&tm));
     
264.                                 printf("%s\n", tbuf);
     
265.                                 fflush(NULL);
     
266.                                 if(conn_sock == -1){
     
267.                                         perror("accept");
     
268.                                         continue;
     
269.                                 }
     
270. 
     
271.                                 if(set_fl(conn_sock, O_NONBLOCK) < 0){
     
272.                                         close(conn_sock);
     
273.                                         continue;
     
274.                                 }
     
275.                                 ev.events = EPOLLIN | EPOLLET;
     
276.                                 ev.data.fd = conn_sock;
     
277.                                 if(epoll_ctl(epollfd, EPOLL_CTL_ADD, conn_sock, &ev) == -1){
     
278.                                         perror("epoll_ctl:conn_sock");
     
279.                                         close(conn_sock);
     
280.                                         continue;
     
281.                                 }
     
282. 
     
283.                                 countacp++;
     
284.                                 client_t *client_ptr = new client_t;
     
285.                                 fds[conn_sock] = client_ptr;
     
286.                                 fds[conn_sock]->fd = conn_sock;
     
287.                                 fds[conn_sock]->ip = string(inet_ntoa(addr_in.sin_addr));
     
288.                                 fds[conn_sock]->last_alive_time = time(NULL);
     
289.                                 pthread_mutex_unlock(&fds_mutex);
     
290.                         }else if(events[n].events & EPOLLIN){
     
291.                                 int fd = events[n].data.fd;
     
292.                                 int n;
     
293.                                 char buf[BUFSIZE];
     
294.                                 memset(buf, 0, BUFSIZE);
     
295. 
     
296.                                 n = readn(fd, buf, BUFSIZE);
     
297.                                 pthread_mutex_lock(&fds_mutex);
     
298.                                 fds[fd]->read_buf.append(buf,BUFSIZE);
     
299.                                 cout<<fds[fd]->read_buf<<endl<<endl;
     
300.                                 if(n >= 0){
     
301.                                         //if(fds[fd]->read_buf.find("\r\n\r\n") != string::npos){                /* 检测请求头是否结束 */
     
302.                                                 //countin++;
     
303.                                                 ev.events = EPOLLOUT | EPOLLET;
     
304.                                                 ev.data.fd = fd;
     
305.                                                 epoll_ctl(epollfd, EPOLL_CTL_MOD ,fd ,&ev);
     
306.                                         //}
     
307.                                 }else{//出错
     
308.                                         if(errno == EAGAIN){
     
309.                                                 continue;
     
310.                                         }
     
311.                                         shutdown(fds[fd]->fd, SHUT_RDWR);
     
312.                                         close(fds[fd]->fd);
     
313.                                         delete fds[fd];
     
314.                                         fds[fd] = NULL;
     
315.                                 }
     
316.                                 fds[fd]->last_alive_time=time(NULL);
     
317.                                
     
318.                                 pthread_mutex_unlock(&fds_mutex);
     
319.                                
     
320.                         }else if(events[n].events & EPOLLOUT){
     
321.                                 int fd = events[n].data.fd;
     
322.                                 pthread_mutex_lock(&fds_mutex);
     
323.                                 string wb = "<html><head><title>hello world</title></head><body>hello world!!!</body></html>";
     
324.                                 fds[fd]->write_buf = "HTTP/1.1 200 Ok\r\n";
     
325.                                 fds[fd]->write_buf += "Content-Type: text/html\r\n";
     
326.                                 fds[fd]->write_buf += "Content-Length: "+int2str(wb.length())+"\r\n";
     
327.                                 fds[fd]->write_buf += "\r\n";
     
328.                                 fds[fd]->write_buf +=wb;
     
329.                                 int n = writen(fd, fds[fd]->write_buf.data(), fds[fd]->write_buf.length());
     
330. 
     
331.                                 fds[fd]->last_alive_time=time(NULL);
     
332.                                 if(n == fds[fd]->write_buf.length() || n < 0){
     
333.                                         epoll_ctl(epollfd, EPOLL_CTL_DEL, fd, NULL);
     
334.                                         shutdown(fd,SHUT_RDWR);
     
335.                                         close(fd);
     
336.                                         delete fds[fd];
     
337.                                         fds[fd] = NULL;
     
338.                                         //countout++;
     
339.                                 }
     
340.                                 pthread_mutex_unlock(&fds_mutex);
     
341.                         }
     
342.                 }
     
343.         }
     
344. 
     
345. 
     
346.         close(listen_sock);
     
347.         close(epollfd);
     
348. 
     
349.         return 0;
     
350. }

复制代码

loveye2009

回复 6# teng0210


    好的，谢谢你。我想用c写，结合epoll和threadpool来处理打批量的客户连接。先看看你的例子，学学epoll的用法。

loveye2009

回复 5# jiufei19

从哪里找呢？不太会用。:wink: