Chinaunix首页 | 论坛 | 博客
  • 博客访问: 307985
  • 博文数量: 56
  • 博客积分: 2058
  • 博客等级: 中尉
  • 技术积分: 688
  • 用 户 组: 普通用户
  • 注册时间: 2011-04-11 09:19
个人简介

code rush

文章分类

全部博文(56)

文章存档

2012年(2)

2011年(54)

分类: LINUX

2011-09-30 17:26:03

  1. 在linux的网络编程中,新的事件触发机制-epoll。

  2. 相比于select,epoll最大的好处在于它不会随着监听fd数目的增长而降低效率。因为在内核中的select实现中,它是采用轮询来处理的,轮询的fd数目越多,自然耗时越多。并且,在linux/posix_types.h头文件有这样的声明:
  3. #define __FD_SETSIZE 1024
  4. 表示select最多同时监听1024个fd,当然,可以通过修改头文件再重编译内核来扩大这个数目。

  5. epoll的接口非常简单,一共就三个函数:
  6. 1. int epoll_create(int size);
  7. 创建一个epoll的句柄,size用来告诉内核这个监听的数目一共有多大。这个参数不同于select()中的第一个参数,给出最大监听的fd+1的值。需要注意的是,当创建好epoll句柄后,它就会占用一个fd值,在linux下如果查看/proc/进程id/fd/,是能够看到这个fd的,所以在使用完epoll后,必须调用close()关闭,否则可能导致fd被耗尽。


  8. 2. int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
  9. epoll的事件注册函数,它不同与select()是在监听事件时告诉内核要监听什么类型的事件,而是在这里先注册要监听的事件类型。第一个参数是epoll_create()的返回值,第二个参数表示动作,用三个宏来表示:
  10. EPOLL_CTL_ADD:注册新的fd到epfd中;
  11. EPOLL_CTL_MOD:修改已经注册的fd的监听事件;
  12. EPOLL_CTL_DEL:从epfd中删除一个fd;
  13. 第三个参数是需要监听的fd,第四个参数是告诉内核需要监听什么事,struct epoll_event结构如下:

  14. typedef union epoll_data {
  15.     void *ptr;
  16.     int fd;
  17.     __uint32_t u32;
  18.     __uint64_t u64;
  19. } epoll_data_t;

  20. struct epoll_event {
  21.     __uint32_t events; /* Epoll events */
  22.     epoll_data_t data; /* User data variable */
  23. };

  24. events可以是以下几个宏的集合:
  25. EPOLLIN :表示对应的文件描述符可以读(包括对端SOCKET正常关闭);
  26. EPOLLOUT:表示对应的文件描述符可以写;
  27. EPOLLPRI:表示对应的文件描述符有紧急的数据可读(这里应该表示有带外数据到来);
  28. EPOLLERR:表示对应的文件描述符发生错误;
  29. EPOLLHUP:表示对应的文件描述符被挂断;
  30. EPOLLET: 将EPOLL设为边缘触发(Edge Triggered)模式,这是相对于水平触发(Level Triggered)来说的。
  31. EPOLLONESHOT:只监听一次事件,当监听完这次事件之后,如果还需要继续监听这个socket的话,需要再次把这个socket加入到EPOLL队列里


  32. 3. int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);
  33. 等待事件的产生,类似于select()调用。参数events用来从内核得到事件的集合,maxevents告之内核这个events有多大,这个 maxevents的值不能大于创建epoll_create()时的size,参数timeout是超时时间(毫秒,0会立即返回,-1将不确定,也有说法说是永久阻塞)。该函数返回需要处理的事件数目,如返回0表示已超时。


  34. 4、关于ET、LT两种工作模式:

  35. EPOLL事件有两种模型:
  36. Edge Triggered (ET)
  37. Level Triggered (LT)

  38. LT(level triggered)是缺省的工作方式,并且同时支持block和no-block socket.在这种做法中,内核告诉你一个文件描述符是否就绪了,然后你可以对这个就绪的fd进行IO操作。如果你不作任何操作,内核还是会继续通知你的,所以,这种模式编程出错误可能性要小一点。传统的select/poll都是这种模型的代表.

  39. ET(edge-triggered)是高速工作方式,只支持no-block socket。在这种模式下,当描述符从未就绪变为就绪时,内核通过epoll告诉你。然后它会假设你知道文件描述符已经就绪,并且不会再为那个文件描述符发送更多的就绪通知,直到你做了某些操作导致那个文件描述符不再为就绪状态了(比如,你在发送,接收或者接收请求,或者发送接收的数据少于一定量时导致了一个EWOULDBLOCK 错误)。但是请注意,如果一直不对这个fd作IO操作(从而导致它再次变成未就绪),内核不会发送更多的通知(only once),不过在TCP协议中,ET模式的加速效用仍需要更多的benchmark确认

  40. 另外,当使用epoll的ET模型来工作时,当产生了一个EPOLLIN事件后,
  41. 读数据的时候需要考虑的是当recv()返回的大小如果等于请求的大小,那么很有可能是缓冲区还有数据未读完,也意味着该次事件还没有处理完,所以还需要再次读取:
  42. while(rs)
  43. {
  44.   buflen = recv(activeevents[i].data.fd, buf, sizeof(buf), 0);
  45.   if(buflen < 0)
  46.   {
  47.     // 由于是非阻塞的模式,所以当errno为EAGAIN时,表示当前缓冲区已无数据可读
  48.     // 在这里就当作是该次事件已处理处.
  49.     if(errno == EAGAIN)
  50.      break;
  51.     else
  52.      return;
  53.    }
  54.    else if(buflen == 0)
  55.    {
  56.      // 这里表示对端的socket已正常关闭.
  57.    }
  58.    if(buflen == sizeof(buf)
  59.      rs = 1; // 需要再次读取
  60.    else
  61.      rs = 0;
  62. }


  63. 还有,假如发送端流量大于接收端的流量(意思是epoll所在的程序读比转发的socket要快),由于是非阻塞的socket,那么send()函数虽然返回,但实际缓冲区的数据并未真正发给接收端,这样不断的读和发,当缓冲区满后会产生EAGAIN错误(参考man send),同时,不理会这次请求发送的数据.所以,需要封装socket_send()的函数用来处理这种情况,该函数会尽量将数据写完再返回,返回-1表示出错。在socket_send()内部,当写缓冲已满(send()返回-1,且errno为EAGAIN),那么会等待后再重试.这种方式并不很完美,在理论上可能会长时间的阻塞在socket_send()内部,但暂没有更好的办法.

  64. ssize_t socket_send(int sockfd, const char* buffer, size_t buflen)
  65. {
  66.   ssize_t tmp;
  67.   size_t total = buflen;
  68.   const char *p = buffer;

  69.   while(1)
  70.   {
  71.     tmp = send(sockfd, p, total, 0);
  72.     if(tmp < 0)
  73.     {
  74.       // 当send收到信号时,可以继续写,但这里返回-1.
  75.       if(errno == EINTR)
  76.         return -1;

  77.       // 当socket是非阻塞时,如返回此错误,表示写缓冲队列已满,
  78.       // 在这里做延时后再重试.
  79.       if(errno == EAGAIN)
  80.       {
  81.         usleep(1000);
  82.         continue;
  83.       }

  84.       return -1;
  85.     }

  86.     if((size_t)tmp == total)
  87.       return buflen;

  88.     total -= tmp;
  89.     p += tmp;
  90.   }

  91.   return tmp;
  92. }


  93. 采用LT模式下, 如果accept调用有返回就可以马上建立当前这个连接了,再epoll_wait等待下次通知,和select一样。
  94. 但是对于ET而言,如果accpet调用有返回,除了建立当前这个连接外,不能马上就epoll_wait还需要继续循环accpet,直到返回-1,且errno==EAGAIN,TAF里面的示例代码:
  95. if(ev.events & EPOLLIN)
  96. {
  97.     do
  98.     {
  99.         struct sockaddr_in stSockAddr;
  100.         socklen_t iSockAddrSize = sizeof(sockaddr_in);
  101.         TC_Socket cs;
  102.         cs.setOwner(false);
  103.         //接收连接
  104.         TC_Socket s;
  105.         s.init(fd, false, AF_INET);
  106.         int iRetCode = s.accept(cs, (struct sockaddr *) &stSockAddr, iSockAddrSize);
  107.         if (iRetCode > 0)
  108.         {
  109.             ...建立连接
  110.         }
  111.         else
  112.         {
  113.             //直到发生EAGAIN才不继续accept
  114.             if(errno == EAGAIN)
  115.             {
  116.                 break;
  117.             }
  118.         }
  119.     }while(true);
  120. }
  121. 同样,recv/send等函数, 都需要到errno==EAGAIN从本质上讲:与LT相比,ET模型是通过减少系统调用来达到提高并行效率的。


  122. ET模型的逻辑:内核的读buffer有内核态主动变化时,内核会通知你, 无需再去mod。写事件是给用户使用的,最开始add之后,内核都不会通知你了,你可以强制写数据(直到EAGAIN或者实际字节数小于 需要写的字节数),当然你可以主动mod OUT,此时如果句柄可以写了(send buffer有空间),内核就通知你。
  123. 这里内核态主动的意思是:内核从网络接收了数据放入了读buffer(会通知用户IN事件,即用户可以recv数据)
  124. 并且这种通知只会通知一次,如果这次处理(recv)没有到刚才说的两种情况(EAGIN或者实际字节数小于 需要读写的字节数),则该事件会被丢弃,直到下次buffer发生变化。
  125. 与LT的差别就在这里体现,LT在这种情况下,事件不会丢弃,而是只要读buffer里面有数据可以让用户读,则不断的通知你。
  126. 另外对于ET而言,当然也不一定非send/recv到前面所述的结束条件才结束,用户可以自己随时控制,即用户可以在自己认为合适的时候去设置IN和OUT事件:
  127. 1 如果用户主动epoll_mod OUT事件,此时只要该句柄可以发送数据(发送buffer不满),则epoll
  128. _wait就会响应(有时候采用该机制通知epoll_wai醒过来)。
  129. 2 如果用户主动epoll_mod IN事件,只要该句柄还有数据可以读,则epoll_wait会响应。
  130. 这种逻辑在普通的服务里面都不需要,可能在某些特殊的情况需要。 但是请注意,如果每次调用的时候都去epoll mod将显著降低效率!
  131. 因此采用et写服务框架的时候,最简单的处理就是:
  132. 建立连接的时候epoll_add IN和OUT事件, 后面就不需要管了

  133. 每次read/write的时候,到两种情况下结束:
  134. 1 发生EAGAIN
  135. 2 read/write的实际字节数小于需要读写的字节数

  136. 对于第二点需要注意两点:
  137. A:如果是UDP服务,处理就不完全是这样,必须要recv到发生EAGAIN为止,否则就丢失事件了
  138. 因为UDP和TCP不同,是有边界的,每次接收一定是一个完整的UDP包,当然recv的buffer需要至少大于一个UDP包的大小
  139. 随便再说一下,一个UDP包到底应该多大?对于internet,由于MTU的限制,UDP包的大小不要超过576个字节,否则容易被分包,对于公司的IDC环境,建议不要超过1472,否则也比较容易分包。
  140. B 如果发送方发送完数据以后,就close连接,这个时候如果recv到数据是实际字节数小于读写字节数,根据开始所述就认为到EAGIN了从而直接返回,等待下一次事件,这样是有问题的,close事件丢失了!因此如果依赖这种关闭逻辑的服务,必须接收数据到EAGIN为止,例如lb。



  141. 那么究竟如何来使用epoll呢?
  142. 通过在包含一个头文件#include <sys/epoll.h> 以及几个简单的API将可以大大的提高你的网络服务器的支持人数。

  143. 首先通过create_epoll(int maxfds)来创建一个epoll的句柄,其中maxfds为你epoll所支持的最大句柄数。这个函数会返回一个新的epoll句柄,之后的所有操作将通过这个句柄来进行操作。在用完之后,记得用close()来关闭这个创建出来的epoll句柄。

  144. 之后在你的网络主循环里面,每一帧的调用epoll_wait(int epfd, epoll_event events, int max events, int timeout)来查询所有的网络接口,看哪一个可以读,哪一个可以写了。基本的语法为:
  145. nfds = epoll_wait(kdpfd, events, maxevents, -1);
  146. 其中kdpfd为用epoll_create创建之后的句柄,events是一个epoll_event*的指针,当epoll_wait这个函数操作成功之后,epoll_events里面将储存所有的读写事件。max_events是当前需要监听的所有socket句柄数。最后一个timeout是 epoll_wait的超时,为0的时候表示马上返回,为-1的时候表示一直等下去,直到有事件范围,为任意正整数的时候表示等这么长的时间,如果一直没有事件,则范围。一般如果网络主循环是单独的线程的话,可以用-1来等,这样可以保证一些效率,如果是和主逻辑在同一个线程的话,则可以用0来保证主循环的效率。

  147. epoll_wait范围之后应该是一个循环,遍利所有的事件。

  148. 几乎所有的epoll程序都使用下面的框架:

  149.     for( ; ; )
  150.     {
  151.         nfds = epoll_wait(epfd,events,20,500);
  152.         for(i=0;i<nfds;++i)
  153.         {
  154.             if(events[i].data.fd==listenfd) //有新的连接
  155.             {
  156.                 connfd = accept(listenfd,(sockaddr *)&clientaddr, &clilen); //accept这个连接
  157.                 ev.data.fd=connfd;
  158.                 ev.events=EPOLLIN|EPOLLET;
  159.                 epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev); //将新的fd添加到epoll的监听队列中
  160.             }
  161.             else if( events[i].events&EPOLLIN ) //接收到数据,读socket
  162.             {
  163.                 n = read(sockfd, line, MAXLINE)) < 0 //
  164.                 ev.data.ptr = md; //md为自定义类型,添加数据
  165.                 ev.events=EPOLLOUT|EPOLLET;
  166.                 epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev);//修改标识符,等待下一个循环时发送数据,异步处理的精髓
  167.             }
  168.             else if(events[i].events&EPOLLOUT) //有数据待发送,写socket
  169.             {
  170.                 struct myepoll_data* md = (myepoll_data*)events[i].data.ptr; //取数据
  171.                 sockfd = md->fd;
  172.                 send( sockfd, md->ptr, strlen((char*)md->ptr), 0 ); //发送数据
  173.                 ev.data.fd=sockfd;
  174.                 ev.events=EPOLLIN|EPOLLET;
  175.                 epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev); //修改标识符,等待下一个循环时接收数据
  176.             }
  177.             else
  178.             {
  179.                 //其他的处理
  180.             }
  181.         }
  182.     }


阅读(3578) | 评论(0) | 转发(4) |
给主人留下些什么吧!~~