libev & libevent简介 IntelIOT 博客园

最主要的还是看看ev_run这个部分代码。我们不打算仔细阅读只是看看梗概然后大体分析一下数据结构应该怎么样的

void ev_run (EV_P_ int flags) {   assert (("libev: ev_loop recursion during release detected", loop_done != EVBREAK_RECURSE));    loop_done = EVBREAK_CANCEL;    EV_INVOKE_PENDING; /* in case we recurse, ensure ordering stays nice and clean */     do     {       if (expect_false (loop_done))         break;        /* update fd-related kernel structures */       fd_reify (EV_A);        /* calculate blocking time */       {         ev_tstamp waittime  = 0.;         ev_tstamp sleeptime = 0.;          /* remember old timestamp for io_blocktime calculation */         ev_tstamp prev_mn_now = mn_now;          /* update time to cancel out callback processing overhead */         time_update (EV_A_ 1e100);          if (expect_true (!(flags & EVRUN_NOWAIT || idleall || !activecnt)))           {             waittime = MAX_BLOCKTIME;              if (timercnt)               {                 ev_tstamp to = ANHE_at (timers [HEAP0]) - mn_now + backend_fudge;                 if (waittime > to) waittime = to;               }              /* don't let timeouts decrease the waittime below timeout_blocktime */             if (expect_false (waittime < timeout_blocktime))               waittime = timeout_blocktime;              /* extra check because io_blocktime is commonly 0 */             if (expect_false (io_blocktime))               {                 sleeptime = io_blocktime - (mn_now - prev_mn_now);                  if (sleeptime > waittime - backend_fudge)                   sleeptime = waittime - backend_fudge;                  if (expect_true (sleeptime > 0.))                   {                     ev_sleep (sleeptime);                     waittime -= sleeptime;                   }               }           }          assert ((loop_done = EVBREAK_RECURSE, 1)); /* assert for side effect */         backend_poll (EV_A_ waittime);         assert ((loop_done = EVBREAK_CANCEL, 1)); /* assert for side effect */          /* update ev_rt_now, do magic */         time_update (EV_A_ waittime + sleeptime);       }        /* queue pending timers and reschedule them */       timers_reify (EV_A); /* relative timers called last */        EV_INVOKE_PENDING;     }   while (expect_true (     activecnt     && !loop_done     && !(flags & (EVRUN_ONCE | EVRUN_NOWAIT))   ));    if (loop_done == EVBREAK_ONE)     loop_done = EVBREAK_CANCEL; } 

我们可以总结一下大致步骤，其实和大部分的event loop写出来差不多。

• 首先触发那些已经pending的watchers.
• 判断是否loop_done
• fd_reify.这个后面会单独说。
• 计算出waittime并且进行必要的sleep.
• backend_poll开始轮询,并且整理好pending事件
• timers_reify.这个和fd_reify不同
• 调用EV_INVOKE_PENDING来触发pending的io事件

非常简单。接下来我们看看fd_reify,backend_poll,timers_reify以及EV_INVOKE_PENDING.

下面是fd_reify代码片段.可以看出，这个部分就是在修改fd关注的events。

inline_size void fd_reify (EV_P) {   int i;   for (i = 0; i < fdchangecnt; ++i)     {       int fd = fdchanges [i];       ANFD *anfd = anfds + fd;       ev_io *w;        unsigned char o_events = anfd->events;       unsigned char o_reify  = anfd->reify;        anfd->reify  = 0;        /*if (expect_true (o_reify & EV_ANFD_REIFY)) probably a deoptimisation */         {           anfd->events = 0;            for (w = (ev_io *)anfd->head; w; w = (ev_io *)((WL)w)->next)             anfd->events |= (unsigned char)w->events;            if (o_events != anfd->events)             o_reify = EV__IOFDSET; /* actually |= */         }        if (o_reify & EV__IOFDSET)         backend_modify (EV_A_ fd, o_events, anfd->events);     }    fdchangecnt = 0; } 

而这个fdchanges这个是在哪里调用的呢。我们可以看到就是在ev_io_start这个部分。也就是说如果我们想要修改 fd关注事件的话，我们必须显示地ev_io_stop掉然后修正之后重新ev_io_start.底层调用fd_change的话底层维护 数组fdchanges来保存发生events变动的fd.

void noinline ev_io_start (EV_P_ ev_io *w) {   int fd = w->fd;    if (expect_false (ev_is_active (w)))     return;    assert (("libev: ev_io_start called with negative fd", fd >= 0));   assert (("libev: ev_io_start called with illegal event mask", !(w->events & ~(EV__IOFDSET | EV_READ | EV_WRITE))));    EV_FREQUENT_CHECK;    ev_start (EV_A_ (W)w, 1);   array_needsize (ANFD, anfds, anfdmax, fd + 1, array_init_zero);   wlist_add (&anfds[fd].head, (WL)w);    fd_change (EV_A_ fd, w->events & EV__IOFDSET | EV_ANFD_REIFY);   w->events &= ~EV__IOFDSET;    EV_FREQUENT_CHECK; }  inline_size void fd_change (EV_P_ int fd, int flags) {   unsigned char reify = anfds [fd].reify;   anfds [fd].reify |= flags;    if (expect_true (!reify))     {       ++fdchangecnt;       array_needsize (int, fdchanges, fdchangemax, fdchangecnt, EMPTY2);       fdchanges [fdchangecnt - 1] = fd;     } } 

backend_poll底层支持很多poll实现，我们这里仅仅看ev_epoll.c就可以.代码在这里面我们不列举了， 如果某个fd触发事件的话那么最终会调用fd_event(EV_A_,fd,event)来进行通知。所以我们看看fd_event.

inline_speed void fd_event_nocheck (EV_P_ int fd, int revents) {   ANFD *anfd = anfds + fd;   ev_io *w;    for (w = (ev_io *)anfd->head; w; w = (ev_io *)((WL)w)->next)     {       int ev = w->events & revents;        if (ev)         ev_feed_event (EV_A_ (W)w, ev);     } } void noinline ev_feed_event (EV_P_ void *w, int revents) {   W w_ = (W)w;   int pri = ABSPRI (w_);    if (expect_false (w_->pending))     pendings [pri][w_->pending - 1].events |= revents;   else     {       w_->pending = ++pendingcnt [pri];       array_needsize (ANPENDING, pendings [pri], pendingmax [pri], w_->pending, EMPTY2);       // set the watcher and revents.       pendings [pri][w_->pending - 1].w      = w_;       pendings [pri][w_->pending - 1].events = revents;     } } 

可以看到底层是一个ANFD的数组，根据fd进行偏移。如果fd过大的话似乎会影响性能没有hpserver里面的demuxtable实现方式好。 然后得到这个fd下面所有的watcher,然后在loop->pendings里面记录所有这些触发的watcher.

其中HEAP0就是最小堆下标。如果repeat的话说明需要重复发生，那么就会重新调整时间戳，如果不是repeat的话， 那么内部会调用ev_timer_stop这个方法将这个计时器移除。所有的定时任务都通过feed_reverse添加。feed_reverse 内部是维护一个动态数组来保存所有的定时器任务，然后在feed_reverse_done里面遍历这些任务来触发这些定时器任务。

inline_size void timers_reify (EV_P) {   EV_FREQUENT_CHECK;    if (timercnt && ANHE_at (timers [HEAP0]) < mn_now)     {       do         {           ev_timer *w = (ev_timer *)ANHE_w (timers [HEAP0]);            /*assert (("libev: inactive timer on timer heap detected", ev_is_active (w)));*/            /* first reschedule or stop timer */           if (w->repeat)             {               ev_at (w) += w->repeat;               if (ev_at (w) < mn_now)                 ev_at (w) = mn_now;                assert (("libev: negative ev_timer repeat value found while processing timers", w->repeat > 0.));                ANHE_at_cache (timers [HEAP0]);               downheap (timers, timercnt, HEAP0);             }           else             ev_timer_stop (EV_A_ w); /* nonrepeating: stop timer */            EV_FREQUENT_CHECK;           feed_reverse (EV_A_ (W)w);         }       while (timercnt && ANHE_at (timers [HEAP0]) < mn_now);        feed_reverse_done (EV_A_ EV_TIMER);     } } 

这个宏最终调用的函数就是下面这个，遍历所有的pendings事件并且逐一触发。

void noinline ev_invoke_pending (EV_P) {   int pri;    for (pri = NUMPRI; pri--; )     while (pendingcnt [pri])       {         ANPENDING *p = pendings [pri] + --pendingcnt [pri];          p->w->pending = 0;         EV_CB_INVOKE (p->w, p->events);         EV_FREQUENT_CHECK;       } } 

#ifndef _COMMON_H_ #define _COMMON_H_  #include <unistd.h> #include <fcntl.h> #include <sys/types.h> #include <sys/socket.h> #include <arpa/inet.h> #include <strings.h> #include <cstdlib> #include <cstdio> #include <cstddef> #include <string>  namespace common{  #define D(exp,fmt,...) do {                     \         if(!(exp)){                             \             fprintf(stderr,fmt,##__VA_ARGS__);  \             abort();                            \         }                                       \     }while(0)  static void setnonblock(int fd){     fcntl(fd,F_SETFL,fcntl(fd,F_GETFL) | O_NONBLOCK); } static void setreuseaddr(int fd){     int ok=1;     setsockopt(fd,SOL_SOCKET,SO_REUSEADDR,&ok,sizeof(ok)); }  static void setaddress(const char* ip,int port,struct sockaddr_in* addr){     bzero(addr,sizeof(*addr));     addr->sin_family=AF_INET;     inet_pton(AF_INET,ip,&(addr->sin_addr));     addr->sin_port=htons(port); }  static std::string address_to_string(struct sockaddr_in* addr){     char ip[128];     inet_ntop(AF_INET,&(addr->sin_addr),ip,sizeof(ip));     char port[32];     snprintf(port,sizeof(port),"%d",ntohs(addr->sin_port));     std::string r;     r=r+"("+ip+":"+port+")";         return r; }  static int new_tcp_server(int port){     int fd=socket(AF_INET,SOCK_STREAM,IPPROTO_TCP);     D(fd>0,"socket failed(%m)\n");     setnonblock(fd);     setreuseaddr(fd);     sockaddr_in addr;     setaddress("0.0.0.0",port,&addr);     bind(fd,(struct sockaddr*)&addr,sizeof(addr));     listen(fd,64); // backlog = 64     return fd; }  static int new_tcp_client(const char* ip,int port){     int fd=socket(AF_INET,SOCK_STREAM,IPPROTO_TCP);     setnonblock(fd);     sockaddr_in addr;     setaddress(ip,port,&addr);     connect(fd,(struct sockaddr*)(&addr),sizeof(addr));     return fd; }  }; // namespace common  #endif // _COMMON_H_ 

#include "ev.h" #include "common.h"  static void do_connected(struct ev_loop* reactor,ev_io* w,int events){     close(w->fd);     ev_break(reactor,EVBREAK_ALL); }  int main(){     struct ev_loop* reactor=ev_loop_new(EVFLAG_AUTO);     int fd=common::new_tcp_client("127.0.0.1",34567);     ev_io io;     ev_io_init(&io,&do_connected,fd,EV_WRITE);     ev_io_start(reactor,&io);     ev_run(reactor,0);     close(fd);     ev_loop_destroy(reactor);     return 0; }  

#include "ev.h" #include "common.h"  static void do_accept(struct ev_loop* reactor,ev_io* w,int events){     struct sockaddr_in addr;     socklen_t addr_size=sizeof(addr);     int conn=accept(w->fd,(struct sockaddr*)&addr,&addr_size);     std::string r=common::address_to_string(&addr);     fprintf(stderr,"accept %s\n",r.c_str());     close(conn); }  int main(){     struct ev_loop* reactor=ev_loop_new(EVFLAG_AUTO);     int fd=common::new_tcp_server(34567);     ev_io w;     ev_io_init(&w,do_accept,fd,EV_READ);     ev_io_start(reactor,&w);     ev_run(reactor,0);     close(fd);     ev_loop_destroy(reactor);     }