首页 > Libev/event, TCP/IP > Libev轻网络库 源码浅析

Libev轻网络库 源码浅析

2014年3月31日 发表评论 阅读评论 20126次阅读    

忍不住先吐槽一下,Libev里面的宏真T*D难看而且很多不必要的宏,好吧,其实他是为了性能。
估计介绍Libev的文章挺多的,这里只是个人记录一下,随便写点东西,libev代码还算简洁,虽然宏多了点,还有各种预定义变量等,今天花了几个小时下载了代码看看,简单记录一下。
之前没有用过libev,一般直接裸写的epoll,总结的话,libev的功能是: 支持将SOCKET,管道, 信号,以及定时器统一为通用的变成逻辑,给开发人员提供了一个简单高效的异步网络编程库。

先看一段简单的客户端程序,标准echo服务:

#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <math.h>
#include <errno.h>
#include <netinet/in.h>
#include <strings.h>
#include "ev.h"

#define PORT 8333
#define BUFFER_SIZE 1024

//gcc test.c -lm ev.o
void accept_cb(struct ev_loop *loop, struct ev_io *watcher, int revents);
static void timeout_cb (EV_P_ ev_timer *w, int revents) ;
void read_cb(struct ev_loop *loop, struct ev_io *watcher, int revents);

int main() {
	struct ev_loop *loop = ev_default_loop(0);
	int sd;
	struct sockaddr_in addr;
	int addr_len = sizeof(addr);
	struct ev_io socket_watcher;
	ev_timer timeout_watcher;

	if( (sd = socket(PF_INET, SOCK_STREAM, 0)) < 0 )
	{
		printf("socket error. errno:%d\n", errno);
		return -1;
	}
	bzero(&addr, sizeof(addr));
	addr.sin_family = AF_INET;
	addr.sin_port = htons(PORT);
	addr.sin_addr.s_addr = INADDR_ANY;
	if (bind(sd, (struct sockaddr*) &addr, sizeof(addr)) != 0) 	{
		printf("bind error.errno:%d\n",errno);
	}
	if (listen(sd, 2) < 0) {
		printf("listen error\n");
		return -1;
	}
	printf("ev_loop beg\n") ;

	//设置cb函数,字段等
	ev_io_init(&socket_watcher, accept_cb, sd, EV_READ);
	ev_io_start(loop, &socket_watcher);

	ev_timer_init (&timeout_watcher, timeout_cb, 2, 1);
	ev_timer_start (loop, &timeout_watcher);

	while (1) {
		printf("ev_loop\n") ;
		ev_loop(loop, 0);
	}
	return 0;
}

void accept_cb(struct ev_loop *loop, struct ev_io *watcher, int revents){
	struct sockaddr_in client_addr;
	socklen_t client_len = sizeof(client_addr);
	int client_sd;

	struct ev_io *w_client = (struct ev_io*) malloc (sizeof(struct ev_io));
	if(EV_ERROR & revents){
		printf("error event in accept\n");
		return;
	}

	client_sd = accept(watcher->fd, (struct sockaddr *)&client_addr, &client_len);
	if (client_sd < 0) {
		printf("accept error\n");
		return;
	}
	printf("someone connected.\n");

	ev_io_init(w_client, read_cb, client_sd, EV_READ);
	ev_io_start(loop, w_client);
}

void read_cb(struct ev_loop *loop, struct ev_io *watcher, int revents){
	char buffer[BUFFER_SIZE];
	ssize_t read;

	if(EV_ERROR & revents) {
		printf("error event in read");
		return;
	}

	read = recv(watcher->fd, buffer, BUFFER_SIZE, 0);
	if(read < 0){
		printf("read error,errno:%d\n", errno);
		return;
	}
	if(read == 0) {
		printf("someone disconnected.errno:%d\n", errno);
		ev_io_stop(loop,watcher);
		free(watcher);
		return;
	} else {
		printf("get the message:%s\n",buffer);
	}

	send(watcher->fd, buffer, read, 0);
	bzero(buffer, read);
}

static void timeout_cb (EV_P_ ev_timer *w, int revents) {
	puts ("timeout");
	//ev_break (EV_A_ EVBREAK_ONE);
}

100行左右的代码,很简单的echo服务,这是libev程序的基本写法,下面分析一下主要的函数。由于现在大部分系统都使用epoll,所以下面假定用的是epoll模式。

一、ev_default_loop初始化epoll句柄

ev_default_loop函数返回一个struct ev_loop *的变量地址,实际上是全局变量default_loop_struct,这个结构体的内容如下,其实就是整个事件循环的总结构体,里面很多成员变量,全部都是用宏定义在ev_vars.h中。

  struct ev_loop
  {
    ev_tstamp ev_rt_now;
    #define ev_rt_now ((loop)->ev_rt_now)
    #define VAR(name,decl) decl;
      #include "ev_vars.h"
    #undef VAR
  };
  #include "ev_wrap.h"

//ev_vars.h
VAR (pendings, ANPENDING *pendings [NUMPRI])  //实际上就是ANPENDING * ev_loop->pendings ,是个指针数组,实际上当优先级队列用的,类似于linux内核里面的进程优先级队列。
VARx(ANFD *, anfds) //同上,这个用来记录每个SOCK的注册的事件列表,用fd索引。

//ev_wrap.h
 #define anfds ((loop)->anfds)   //逆天了,不就直接写一下嘛,用的这么生僻··
#define pendings ((loop)->pendings)

比如定义的总之一句,这个结构就是整个库的中心结构,里面包含各种需要的成员变量。

二、事件初始化

接下来程序用socket新建了一个SOCK句柄,并且绑定了本地端口,然后用listen设置句柄为监听状态。 这些都是比较通用的,跟libev没有太大关系。 后面就需要针对不同的句柄,设置不同的事件回调。

值得注意的是libev支持文件通知,定时器,管道等,其分别用ev_io, ev_timer等表示,记录相关的数据,然后分别用不同的函数初始化ev_io_init, ev_timer_init,实际上就是讲几份代码放在一起,稍微柔和下。从而支持不同的类型。

ev_io_init 函数初始化epoll的回调函数,以及记录事件的参数,存放在ev_io结构里面,但此时还没有加入到epoll里面的。

     ev_io_init(&socket_watcher, accept_cb, sd, EV_READ);//关注可读事件,回调函数为accept_cb,用来接收新连接。
//函数内容:
#define ev_io_init(ev,cb,fd,events)          do { ev_init ((ev), (cb)); ev_io_set ((ev),(fd),(events)); } while (0)

#define ev_init(ev,cb_) do {            \
  ((ev_watcher *)(void *)(ev))->active  =   \
  ((ev_watcher *)(void *)(ev))->pending = 0;    \
  ev_set_priority ((ev), 0);            \ //设置优先级,注意libev是能支持对不同的事件加优先级的,这个优先级决定:在同时epoll_waite返回的多个句柄后,优先级高的先得到回调。
  ev_set_cb ((ev), cb_);            \    //设置回调函数到ev->cb, 注意这个跟其他的比如redis,nginx会分别有2个函数不一样。
} while (0)

#define ev_io_set(ev,fd_,events_)            do { (ev)->fd = (fd_); (ev)->events = (events_) | EV__IOFDSET; } while (0)

可以看出ev_io_init没有做什么实质性的事情,就是记录了一下几个参数而已。

ev_io_start 函数虽然名字叫start,但实际有点骗人,因为没有start,记录这个句柄的事件,实际上还没有加入epoll的, 只是记录到了fdchanges里面。

libev里面对于所有的SOCKT句柄,都会在ev_loop->anfds数组里面按下标索引了一个槽位,其里面是个anfds[fd].head链表,每个节点代表我们上面用ev_io_init设置的事件,因此如果要同时关注可读,可写事件,需要调用2次ev_io_init,ev_io_start分别设置可读可写事件。

void noinline
ev_io_start (EV_P_ ev_io *w) EV_THROW
{//记录这个句柄的事件,实际上还没有加入epoll的, 只是记录到了fdchanges里面
  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);//为新的句柄准备空间,放到(loop)->anfds里面去的
  wlist_add (&anfds[fd].head, (WL)w);//将W插入到anfds[fd]的头部,挂一个未处理事件,比如增加,删除啥的

  /* common bug, apparently */
  assert (("libev: ev_io_start called with corrupted watcher", ((WL)w)->next != (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)
{//标记这个句柄的改动到fdchanges,后面run的时候会处理加入epoll的
  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;//因为这里根本没有将句柄加入epoll, 而是将这个有改动事件的事情记录
      //到ev_loop->fdchangeslim .意思是说:这个fdchanges数组里面的句柄都有新的事件加入或者删除,待会需要调用epoll_ctl来注册。
    }
}

上面比较关注的函数是,因为这里根本没有将句柄加入epoll, 而是将这个有改动事件的事情记录到ev_loop->fdchangeslim .意思是说:这个fdchanges数组里面的句柄都有新的事件加入或者删除,待会需要调用epoll_ctl来注册。

对于定时器事件来说,大体类似,只是不需要epoll了。就不多说了。下面看最主要的ev_loop循环。

三、ev_loop事件监听循环

实际上ev_loop是个宏观定义,内容就是ev_run   (EV_A_ flags); ev_run 函数挺大的,分步介绍。

int ev_run (EV_P_ int flags) {
  do {
      fd_reify (EV_A);//这个函数是将ev_io_start开启的事件记录的fdchanges列表设置到epoll句柄监听事件集里面去
      /* calculate blocking time */
      {
//···根据定时器的最近触发时间,计算这次的epoll等待最长可以等待多久。
      }
        //这里虽然waite了,而且解析读写事件类型了,但是还没有调用那些处理函数的,只是将他们挂入了pendings 的优先级队列
        backend_poll (EV_A_ waittime);//实际上调用的是epoll_wait等待监听事件可读可写
        time_update (EV_A_ waittime + sleeptime);
      timers_reify (EV_A); /* relative timers called last */

      EV_INVOKE_PENDING;// FUCK, 又用宏,实际上就是调用了ev_invoke_pending函数,一个个将上面backend_poll里面
      //检测到的有事件的SOCKT pendings队列里面的事件进行处理
}
  • ev_run简化一下就是这么简单,先用fd_reify将之前调用ev_io_init,ev_io_start挂到ev_loop->fdchanges数组里面的socket句柄一个个处理掉,也就是放入epoll监听事件集合中。
  • 然后调用backend_poll进行等待监听可读写事件,实际上是调用epoll_poll 函数,后者再调用epoll_wait真正去等待事件。
  • 最后用宏EV_INVOKE_PENDING 来将发生过事情的事件回调函数按优先级触发。

下面分别介绍一下。

fd_reify注册可读写事件到epoll

fd_reify将fdchanges里面在ev_io_start里面设置记录的这些新事件一个个处理,真正加入epoll里面.所谓的reify具体化。

inline_size void fd_reify (EV_P)
{//将fdchanges里面在ev_io_start里面设置记录的这些新事件一个个处理,真正加入epoll里面.所谓的reify具体化
  int i;

  for (i = 0; i < fdchangecnt; ++i)
    {
      int fd = fdchanges [i];//只需要获取里面的fd,然后就可以在anfds里面索引描述符数据的位置了
      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 |= */
        }

        //下面调用的其实是epoll_modify, 后者根据具体情况调用epoll_ctl函数,将新事件设置到epoll里面去 ,或者减少
      if (o_reify & EV__IOFDSET)
        backend_modify (EV_A_ fd, o_events, anfd->events);
    }

  fdchangecnt = 0;
}

从上面可以看出fd_reify会遍历ev_loop->fdchanges 数组,将里面有事件改动的socket的(ev_io *)anfd->head事件列表进行整合,然后看是否跟之前有变动,如果有,则调用backend_modify修改epoll注册,实际调用的其实是epoll_modify, 后者根据具体情况调用epoll_ctl函数,将新事件设置到epoll里面去 ,或者减少。

epoll_modify在ev_epoll.c里面,其整体就是个epoll_ctl调用,再熟悉不过了。

至此,一个SOCKET的epoll事件已经加入到了epollfd中了,这样只要wait,就能监听这个事件的变化。

backend_poll监听等待SOCKET可读

backend_poll实际上是epoll_poll,这个是在整个初始化的时候在epoll_init里面设置的。epoll_poll挺简单的,做2件事情:

  • 调用epoll_wait等待监听事件通知;
  • fd_event解析事件的类型,然后放到pendings的优先级队列里面,这样到后面再慢慢处理;

注意这个epoll_poll实际上还没有触发回调函数。里面对每个可读写的fd都会调用fd_event (EV_A_ fd, got); , 最终调用fd_event_nocheck, ev_feed_event。

inline_speed void
fd_event_nocheck (EV_P_ int fd, int revents)
{//对一个SOCK进行检查,看他是不是想要revents的事件,如果想要,那么将其加入到pengdings优先级队列里面,以备后续处理
  ANFD *anfd = anfds + fd;
  ev_io *w;

  for (w = (ev_io *)anfd->head; w; w = (ev_io *)((WL)w)->next)
    {//看到了吧,anfd的head链表实际上就是一个个事件,比如可读,可写分别为2个事件,发生事件时会扫描这里,去寻找需要的事件,然后处>
理
      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) EV_THROW
{//喂养事件,实际上就是将事件放到(loop)->pendings的数组里面,然后后面会一个个按照优先级去处理的
  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);
      pendings [pri][w_->pending - 1].w      = w_;
      pendings [pri][w_->pending - 1].events = revents;
    }

  pendingpri = NUMPRI - 1;
}

从上面也可以看出libev的优先级是怎么回事,就是会在本次epoll_wait返回后,优先级高的先回调。实现方法是用ev_loop->pendings[]数组来记录不同优先级的事件。从而让后面的代码按顺序处理。

ep_run里面还用timers_reify处理了一下定时器事件,periodics_reify处理绝对时间的事件。

最后调用EV_INVOKE_PENDING 而触发各种毁掉函数,其实一个宏。


# define EV_INVOKE_PENDING invoke_cb (EV_A)
#define invoke_cb ((loop)->invoke_cb)

static void noinline ecb_cold
loop_init (EV_P_ unsigned int flags) EV_THROW
{//调用epoll_create创建epoll句柄
//···
      invoke_cb          = ev_invoke_pending;//这个就是触发回调函数的函数,在ev_loop里面调用
//···

从此可见,EV_INVOKE_PENDING实际上就是ev_invoke_pending函数,其很简单,从高优先级的开始,一个个处理pendings数组。

void noinline
ev_invoke_pending (EV_P)
{//触发在epoll_wait后加入到pendings优先级队列里面的事件,一个个一次调用他们的cb回调函数
  pendingpri = NUMPRI;

  while (pendingpri) /* pendingpri possibly gets modified in the inner loop */
    {
      --pendingpri;

      while (pendingcnt [pendingpri])
        {
          ANPENDING *p = pendings [pendingpri] + --pendingcnt [pendingpri];

          p->w->pending = 0;
          EV_CB_INVOKE (p->w, p->events);//调用回调函数
          EV_FREQUENT_CHECK;
        }
    }
}
# define EV_CB_INVOKE(watcher,revents) (watcher)->cb (EV_A_ (watcher), (revents))

这样遍历从高优先级的开始处理,从而达到优先级的效果。不过其实感觉这个没啥太多用处。因为快慢都差不多反正立即处理了。

介绍到这里差不多了,从初始化到注册可读写事件,然后进入ev_loop循环,从而对事件进行可读可写的监听而得到回调。

libev的代码不多,不过为了性能大量使用了宏,所以可读性不是特别好,据说libevent好点,改天看看吧。

看代码时我对代码做了些简单的注释或者简化,放在我的github上了。

写的有点急,就话花了几个小时看的代码,所以难免有错误,发现的同学留言告诉我指正一下,谢谢了。

Share
  1. kulv
    2015年12月8日21:58 | #1

    @i4
    是的

  2. i4
    2015年1月11日21:31 | #2

    event.h/.c是为了兼容libevent的吗?

  1. 2014年11月20日10:02 | #1

注意: 评论者允许使用'@user空格'的方式将自己的评论通知另外评论者。例如, ABC是本文的评论者之一,则使用'@ABC '(不包括单引号)将会自动将您的评论发送给ABC。使用'@all ',将会将评论发送给之前所有其它评论者。请务必注意user必须和评论者名相匹配(大小写一致)。