Libevent分析

一、使用举例

一个简单的使用libevent创建的echosvr如下：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <errno.h>
#include <event.h>
#include "dlist.h"

#define BUF_SIZE    1024
char buf[BUF_SIZE];

void on_accept(int sock, short event, void* arg);
void on_read(int sock, short event, void* arg);

struct event_list
{
struct event ev;
struct list_head list;
};

struct event_list evlist;
struct event_base* evbase = NULL;

int main(void)
{
int listenfd = -1;
struct sockaddr_in svraddr;
int nready = 0;
int i = 0;
INIT_LIST_HEAD(&evlist.list);

if ((listenfd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
{
perror("socket error");
return -1;
}

memset(&svraddr, 0, sizeof(svraddr));
svraddr.sin_family = AF_INET;
svraddr.sin_port = htons(44888);
svraddr.sin_addr.s_addr = htonl(INADDR_ANY);

if (bind(listenfd, (struct sockaddr* )&svraddr, sizeof(svraddr)) < 0)
{
perror("bind error");
return -1;
}

if (listen(listenfd, 5) < 0)
{
perror("listen error");
return -1;
}

evbase = event_base_new();
struct event_list* evlisten = (struct event_list* )malloc(sizeof(struct event_list));
list_add_tail(&(evlisten->list), &(evlist.list));
event_set(&(evlisten->ev), listenfd, EV_READ | EV_PERSIST, on_accept, (void* )evbase);
event_base_set(evbase, &(evlisten->ev));
event_add(&(evlisten->ev), NULL);

event_base_dispatch(evbase);
printf("The End.\n");

return 0;
}

void on_accept(int sock, short event, void* arg)
{
int fd;
struct event_base* evbase = (struct event_base* )arg;
struct sockaddr_in cliaddr;
uint32_t slen = sizeof(cliaddr);
fd = accept(sock, (struct sockaddr* )&cliaddr, &slen);
if (fd < 0)
{
perror("accept");
return;
}
printf("recv sock %d: <%s:%d>\n", fd, inet_ntoa(cliaddr.sin_addr), htons(cliaddr.sin_port));

struct event_list* evread = (struct event_list* )malloc(sizeof(struct event_list));
event_set(&(evread->ev), fd, EV_READ | EV_PERSIST, on_read, (void* )evread);
event_base_set(evbase, &(evread->ev));
event_add(&(evread->ev), NULL);
}

void on_read(int sock, short event, void* arg)
{
int rsize = 0;
memset(buf, 0, BUF_SIZE);
rsize = read(sock, buf, BUF_SIZE);
if (rsize <= 0)
{
if (rsize == 0)
{
printf("client %d close socket.\n", sock);
}
else
{
perror("read:");
}
close(sock);
event_del((struct event* )arg);
free((struct event* )arg);
return;
}
printf("recv:%s\n", buf);
write(sock, buf, strlen(buf));
printf("send:%s\n", buf);
}

其中使用了双链表来管理events，相关代码如下：

#ifndef _LIST_HEAD_H
#define _LIST_HEAD_H

// 双向链表节点
struct list_head {
struct list_head *next, *prev;
};

// 初始化节点：设置name节点的前继节点和后继节点都是指向name本身。
#define LIST_HEAD_INIT(name) { &(name), &(name) }

// 定义表头(节点)：新建双向链表表头name，并设置name的前继节点和后继节点都是指向name本身。
#define LIST_HEAD(name) \
struct list_head name = LIST_HEAD_INIT(name)

// 初始化节点：将list节点的前继节点和后继节点都是指向list本身。
static inline void INIT_LIST_HEAD(struct list_head *list)
{
list->next = list;
list->prev = list;
}

// 添加节点：将new插入到prev和next之间。
static inline void __list_add(struct list_head *new,
struct list_head *prev,
struct list_head *next)
{
next->prev = new;
new->next = next;
new->prev = prev;
prev->next = new;
}

// 添加new节点：将new添加到head之后，是new称为head的后继节点。
static inline void list_add(struct list_head *new, struct list_head *head)
{
__list_add(new, head, head->next);
}

// 添加new节点：将new添加到head之前，即将new添加到双链表的末尾。
static inline void list_add_tail(struct list_head *new, struct list_head *head)
{
__list_add(new, head->prev, head);
}

// 从双链表中删除entry节点。
static inline void __list_del(struct list_head * prev, struct list_head * next)
{
next->prev = prev;
prev->next = next;
}

// 从双链表中删除entry节点。
static inline void list_del(struct list_head *entry)
{
__list_del(entry->prev, entry->next);
}

// 从双链表中删除entry节点。
static inline void __list_del_entry(struct list_head *entry)
{
__list_del(entry->prev, entry->next);
}

// 从双链表中删除entry节点，并将entry节点的前继节点和后继节点都指向entry本身。
static inline void list_del_init(struct list_head *entry)
{
__list_del_entry(entry);
INIT_LIST_HEAD(entry);
}

// 用new节点取代old节点
static inline void list_replace(struct list_head *old,
struct list_head *new)
{
new->next = old->next;
new->next->prev = new;
new->prev = old->prev;
new->prev->next = new;
}

// 双链表是否为空
static inline int list_empty(const struct list_head *head)
{
return head->next == head;
}

// 获取"MEMBER成员"在"结构体TYPE"中的位置偏移
#define offsetof(TYPE, MEMBER) ((size_t) &((TYPE *)0)->MEMBER)

// 根据"结构体(type)变量"中的"域成员变量(member)的指针(ptr)"来获取指向整个结构体变量的指针
#define container_of(ptr, type, member) ({          \
const typeof( ((type *)0)->member ) *__mptr = (ptr);    \
(type *)( (char *)__mptr - offsetof(type,member) );})

// 遍历双向链表
#define list_for_each(pos, head) \
for (pos = (head)->next; pos != (head); pos = pos->next)

#define list_for_each_safe(pos, n, head) \
for (pos = (head)->next, n = pos->next; pos != (head); \
pos = n, n = pos->next)

#define list_entry(ptr, type, member) \
container_of(ptr, type, member)

#endif

View Code
echosvr的代码流程：

1、经典的使用socket创建tcp socket。

2、创建event base：evbase = event_base_new();

3、创建listen event并加入到eventbase中：

event_set(&(evlisten->ev), listenfd, EV_READ | EV_PERSIST, on_accept, (void* )evbase);
event_base_set(evbase, &(evlisten->ev));
event_add(&(evlisten->ev), NULL);

二、Libevent基本数据结构

1、struct event_base (事件处理框架结构)

struct event_base
{
const struct eventop *evsel;    // IO复用函数集, Linux下一般是epoll的封装
void *evbase;
int event_count;        /* counts number of total events */
int event_count_active;    /* counts number of active events */

int event_gotterm;        /* Set to terminate loop */
int event_break;        /* Set to terminate loop immediately */

/* active event management */
struct event_list **activequeues;
int nactivequeues;

/* signal handling info */
struct evsignal_info sig;

struct event_list eventqueue;    // 注册事件队列
struct timeval event_tv;

struct min_heap timeheap;        // 管理定时器小根堆

struct timeval tv_cache;        // 记录时间缓存
};

evsel：libevent支持Linux、Windows等多种平台，也支持epoll、poll、select、kqueue等多种I/O多路复用模型。evsel会指向

/* In order of preference */
static const struct eventop *eventops[] = {
#ifdef HAVE_EVENT_PORTS
&evportops,
#endif
#ifdef HAVE_WORKING_KQUEUE
&kqops,
#endif
#ifdef HAVE_EPOLL
&epollops,
#endif
#ifdef HAVE_DEVPOLL
&devpollops,
#endif
#ifdef HAVE_POLL
&pollops,
#endif
#ifdef HAVE_SELECT
&selectops,
#endif
#ifdef WIN32
&win32ops,
#endif
NULL
};

linux下会指向epollops。

2、struct event(代表一个事件)

struct event
{
TAILQ_ENTRY (event) ev_next;            // 已注册事件
TAILQ_ENTRY (event) ev_active_next;     // 就绪事件
TAILQ_ENTRY (event) ev_signal_next;        // 信号
unsigned int min_heap_idx;    /* for managing timeouts */

struct event_base *ev_base;

int ev_fd;                // 对于I/O事件,是绑定的文件描述符
short ev_events;        // event关注的事件类型
short ev_ncalls;        // 事件就绪执行时，调用 ev_callback 的次数
short *ev_pncalls;    /* Allows deletes in callback */

struct timeval ev_timeout;    // timout事件的超时值

int ev_pri;        /* smaller numbers are higher priority */

void (*ev_callback)(int, short, void *arg);    // 回调函数
void *ev_arg;                                // 回调函数的参数

int ev_res;        /* result passed to event callback */
int ev_flags;                                // event的状态
};

1) ）ev_next，ev_active_next 和 ev_signal_next 都是双向链表节点指针, 定义如下：

#define TAILQ_ENTRY(type)                        \
struct {                                \
struct type *tqe_next;    /* next element */            \
struct type **tqe_prev;    /* address of previous next element */    \
}

2) ev_events：event关注的事件类型，它可以是以下3种类型：
I/O事件： EV_WRITE和EV_READ
定时事件：EV_TIMEOUT
信号： EV_SIGNAL
辅助选项：EV_PERSIST，表明是一个永久事件

三、libevent主要函数分析

1、event_base_new：创建事件处理框架

struct event_base *event_base_new(void)
{
int i;
struct event_base *base;
// 为base分配堆内存
if ((base = calloc(1, sizeof(struct event_base))) == NULL)
event_err(1, "%s: calloc", __func__);
// 信号相关
event_sigcb = NULL;
event_gotsig = 0;

detect_monotonic();
gettime(base, &base->event_tv);

min_heap_ctor(&base->timeheap);
TAILQ_INIT(&base->eventqueue);
base->sig.ev_signal_pair[0] = -1;
base->sig.ev_signal_pair[1] = -1;

// 具体底层IO操作函数, 实际使用epoll封装
base->evbase = NULL;
for (i = 0; eventops[i] && !base->evbase; i++) {
base->evsel = eventops[i];

base->evbase = base->evsel->init(base);
}

if (base->evbase == NULL)
event_errx(1, "%s: no event mechanism available", __func__);

if (evutil_getenv("EVENT_SHOW_METHOD"))
event_msgx("libevent using: %s\n",
base->evsel->name);

/* allocate a single active event queue */
event_base_priority_init(base, 1);

return (base);
}

函数首先申请base的内存，第16行TAILQ_INIT(&base->eventqueue);是初始化注册事件，TAILQ_INIT是一个宏定义：

#define    TAILQ_INIT(head) do {                        \
(head)->tqh_first = NULL;                    \
(head)->tqh_last = &(head)->tqh_first;                \
} while (0)

tqh_first为空，tqh_last指向tqh_first的地址，完成一个空的双指针的初始化。

22-26行是注册底层IO事件操作函数集，这里指向的是epollops，有如下操作：

const struct eventop epollops = {
"epoll",
epoll_init,
epoll_add,
epoll_del,
epoll_dispatch,
epoll_dealloc,
1 /* need reinit */
};

epoll相关的操作可以参考http://www.cnblogs.com/ym65536/p/4854869.html, 这里的操作只是对epoll做了简单的封装。

2、event_set

void event_set(struct event *ev, int fd, short events,
void (*callback)(int, short, void *), void *arg)
{
/* Take the current base - caller needs to set the real base later */
ev->ev_base = current_base;

ev->ev_callback = callback;
ev->ev_arg = arg;
ev->ev_fd = fd;
ev->ev_events = events;
ev->ev_res = 0;
ev->ev_flags = EVLIST_INIT;
ev->ev_ncalls = 0;
ev->ev_pncalls = NULL;

min_heap_elem_init(ev);

/* by default, we put new events into the middle priority */
if(current_base)
ev->ev_pri = current_base->nactivequeues/2;
}

这个函数很简单，只是绑定了事件event和描述符、回调函数等信息。

3、event_base_set

int
event_base_set(struct event_base *base, struct event *ev)
{
/* Only innocent events may be assigned to a different base */
if (ev->ev_flags != EVLIST_INIT)
return (-1);

ev->ev_base = base;
ev->ev_pri = base->nactivequeues/2;

return (0);
}

绑定事件event和事件处理框架event_base

4、event_add