libevent简介[翻译]9 Bufferevents的概念和基础

http://www.wangafu.net/~nickm/libevent-book/Ref6_bufferevent.html

Bufferevents的概念和基础

很多时候，一个程序想去执行一些数据缓冲，除了响应事件之外。当我们想去写入数据是，一般的操作如下：

• 决定对连接写入一些数据，把数据放到缓冲

• 等待连接可写

• 尽可能写入我们可写的数据

• 记录写入了多少数据，如果还有数据需要写入，就等待下一次连接可写

这种缓冲IO模型非常常见，玉石libevent提供了一个通用的模型。bufferevent提供了一个底层实现的传输协议，类似于socket，包括读缓冲和写缓冲。不再用标准的事件通知，二是用回调的方式，当可读或是可写时，bufferevent就会调用用户提供的回调函数通知。

bufferevent有很多类型：

1. 基于socket的bufferevent

   • bufferevent通过event_*这一类的函数从底层socket流中读取或是写入数据

2. 异步IO bufferevents

   • 在Windows平台下，bufferevent通过IOCP收发底层socket流的数据

3. 过滤 bufferevents

   • 处理到达和将要发送的数据。比如压缩、装换灯。

4. 成对的bufferevents

   • 两个bufferevents相互发送数据。

注意

目前Bufferevents只支持流式的协议，比如TCP，未来可能支持数据包式的协议，比如UDP。

Bufferevents和evbuffers

每一个bufferevent都有一个入缓冲和出缓冲。他们都是"struct evbuffer"的类型。当你有数据要写的时候，把它增加到出缓冲中，当bufferevent有数据要读时，就会从入缓冲获得数据。

回调函数和水印

每个bufferevent都有两个与数据相关的回调：一个读回调和一个写回调。默认情况下，当有数据可读时，读回调就会被调用；当有足够的数据写入到底层来清空写缓冲时，就会调用写回调。你可以通过调整读和写的水印来重载覆盖这个行为。

每个bufferevent都有四个水印：

1. 读低水位标识

   • 当有读的数据触发时，如果使得bufferevent的入缓冲区超过这个标准，就会触发。默认是0，也就是有数据，就会触发。

2. 读高水位标识

   • 当bufferevent入缓冲区到达这个标准，bufferevent就会停止读取，直到有足够多的数据可以从入缓冲区获得使得我们低于它。默认是无穷大，所以我们不会因为入缓冲区的大小而停止读取。

3. 写低水位标识

   • 当一个写触发时，使得我们到达这个标准或是更低，就会调用写回调。默认是0，所以除非出缓冲时0，所以不会触发写回调。

4. 写高水位

   • 并不是bufferevent直接使用，这个标识在特殊情况下使用，当一个bufferevent在底层用作另一个bufferevent发送数据，见下面filtering bufferevents的解释。

一个bufferevent同样有"error"和"event"的回调，用来告诉我们没有数据可操作，比如连接关闭或是错误发生。如下定义了这些标识：

1. BEV_EVENT_READING

   • 在读操作的时候，触发了一个事件。参考其他标识用来查看是哪个事件。

2. BEV_EVENT_WRITING

   • 在写的时候触发了事件，参考其他标识查看是哪个事件。

3. BEV_EVENT_ERROR

   • 在bufferevent操作时触发了错误。调用EVUTIL_SOCKET_ERROR()获取更多信息。

4. BEV_EVENT_TIMEOUT

   • 超时

5. BEV_EVENT_EOF

   • 触发了end-of-file标识

6. BEV_EVENT_CONNECTED

   • 完成了连接请求。

延迟回调

默认情况下，在特定的事件发生时，bufferevent的回调会立马执行。这种立马回调的情况在依赖比较多的时候会引起问题。比如，有一个回调是把数据放入到evbuffer A，在它是空的时候；另一个回调会把数据从evbuffer A中取出来，当它满的时候。因为这两个回调都发生在堆栈上，你可能会遇到堆栈溢出的风险当依赖变得足够差时。

为了解决这个问题，你可以告诉bufferevent让回调延迟一下。当这个条件触发时，延迟回调并不会立马执行，而是会作为event_loog()一部分进行排队，在正常的事件回调完成后再执行。

bufferevents的操作标识

你可以使用一个或是多个标识来创建bufferevent，这样就可以修改它的行为。推荐的标识有：

1. BEV_OPT_CLOSE_ON_FREE

   • 当bufferevent被释放的时候，关闭底层的传输端口。这也会关闭底层的socket，释放底层的bufferevent

2. BEV_OPT_THREADSAFE

   • 自动为bufferevent申请锁，在多线程下是安全的。

3. BEV_OPT_DEFER_CALLBACKS

   • bufferevent会延迟所有相关的回调。

4. BEV_OPT_UNLOCK_CALLBACKS

   • 默认情况下，设置bufferevent为线程安全的，任何一个用户提供的回调被执行的时候都会拥有bufferevent的锁。设置这个选项可以在用户回调的时候让libevent释放bufferevent的锁。

使用基于socket的bufferevents

最简单的bufferevents使用类型是基于socket的。基于socket的bufferevent是使用了libevent的底层事件机制，用来检测底层网络socket是否可读或是可写。并且使用底层的网络调用比如readv writev WSASend WSARecv去传输和接收数据。

创建一个基于socket的bufferevent

可以使用bufferevent_socket_new()创建一个基于socket的bufferevent

接口

struct bufferevent *bufferevent_socket_new(
    struct event_base *base,
    evutil_socket_t fd,
    enum bufferevent_options options);

base就是event_base，options是bufferevent操作，比如BEV_OPT_CLOSE_ON_FREE，的位掩码。fd是文件描述符，可以设置为-1，稍后在设置对应的文件描述符。

Tip

确保提供给bufferevent_socket_new的socket是非阻塞模式的。libevent提供更方便的接口evutil_make_socket_nonblocking用来操作这个。

成功返回bufferevent，失败返回NULL

对基于socket的bufferevents发起连接

如果bufferevent的socket没有连接，你可以发起一个新的连接。

接口

int bufferevent_socket_connect(struct bufferevent *bev,
    struct sockaddr *address, int addrlen);

address和addrlen两个参数与标准的connect()一样。如果bufferevent还没有socket，调用这个函数会申请一个新的socket流，并且设置为非阻塞。

如果bufferevent已经有一个socket了，调用bufferevent_socket_connect()会告诉libevent socket已经不再连接了，不能对socket读写，直到再次连接成功。

在连接成功之前，可以像output buffer添加数据。

成功返回0，失败返回-1

示例

#include <event2/event.h>
#include <event2/bufferevent.h>
#include <sys/socket.h>
#include <string.h>

void eventcb(struct bufferevent *bev, short events, void *ptr)
{
    if (events & BEV_EVENT_CONNECTED) {
         /* We're connected to 127.0.0.1:8080.   Ordinarily we'd do
            something here, like start reading or writing. */
    } else if (events & BEV_EVENT_ERROR) {
         /* An error occured while connecting. */
    }
}

int main_loop(void)
{
    struct event_base *base;
    struct bufferevent *bev;
    struct sockaddr_in sin;

    base = event_base_new();

    memset(&sin, 0, sizeof(sin));
    sin.sin_family = AF_INET;
    sin.sin_addr.s_addr = htonl(0x7f000001); /* 127.0.0.1 */
    sin.sin_port = htons(8080); /* Port 8080 */

    bev = bufferevent_socket_new(base, -1, BEV_OPT_CLOSE_ON_FREE);

    bufferevent_setcb(bev, NULL, NULL, eventcb, NULL);

    if (bufferevent_socket_connect(bev,
        (struct sockaddr *)&sin, sizeof(sin)) < 0) {
        /* Error starting connection */
        bufferevent_free(bev);
        return -1;
    }

    event_base_dispatch(base);
    return 0;
}

当你使用bufferevent_socket_connect()发起connect()的时候，只会得到BEV_EVENT_CONNECTED的事件。如果你自己调用connect()，会收到写的消息

如果你自己调用connect()，但仍收到BEV_EVENT_CONNECTED事件在连接成功时，在调用connect()之后调用bufferevent_socket_connect(bev, NULL, 0)会返回-1，附带一个errno，值可能是EAGAIN或者EINPROGRESS。

通过hostname启动一个连接

经常，你想把解析hostname并且创建一个连接放到一个操作步骤里面，这里有一个接口：

接口

int bufferevent_socket_connect_hostname(struct bufferevent *bev,
    struct evdns_base *dns_base, int family, const char *hostname,
    int port);
int bufferevent_socket_get_dns_error(struct bufferevent *bev);

这个函数解析DNS，获得对应协议簇的地址，允许的协议簇类型有AF_INET、AF_INET6和AF_UNSPEC。如果解析失败，会回调一个错误事件。如果成功，会启动一个连接。

dns_base参数是一个设置，如果是NULL，libevent会在等待hostname解析的过程中阻塞。如果设置了，libevent会一步的检测获得hostname。

调用了bufferevent_socket_connect()这个函数告诉libevent任何存在的socket都没有连接，不可以读或是写，知道解析完成并且连接成功。

如果出错，有可能还DNS解析的错误。可以通过bufferevent_socket_get_dns_error()获得对应的错误信息。如果错误代码是0，表示没有DNS错误发生。

简单的HTTP v0客户端

/* Don't actually copy this code: it is a poor way to implement an
   HTTP client.  Have a look at evhttp instead.
*/
#include <event2/dns.h>
#include <event2/bufferevent.h>
#include <event2/buffer.h>
#include <event2/util.h>
#include <event2/event.h>

#include <stdio.h>

void readcb(struct bufferevent *bev, void *ptr)
{
    char buf[1024];
    int n;
    struct evbuffer *input = bufferevent_get_input(bev);
    while ((n = evbuffer_remove(input, buf, sizeof(buf))) > 0) {
        fwrite(buf, 1, n, stdout);
    }
}

void eventcb(struct bufferevent *bev, short events, void *ptr)
{
    if (events & BEV_EVENT_CONNECTED) {
         printf("Connect okay.
");
    } else if (events & (BEV_EVENT_ERROR|BEV_EVENT_EOF)) {
         struct event_base *base = ptr;
         if (events & BEV_EVENT_ERROR) {
                 int err = bufferevent_socket_get_dns_error(bev);
                 if (err)
                         printf("DNS error: %s
", evutil_gai_strerror(err));
         }
         printf("Closing
");
         bufferevent_free(bev);
         event_base_loopexit(base, NULL);
    }
}

int main(int argc, char **argv)
{
    struct event_base *base;
    struct evdns_base *dns_base;
    struct bufferevent *bev;

    if (argc != 3) {
        printf("Trivial HTTP 0.x client
"
               "Syntax: %s [hostname] [resource]
"
               "Example: %s www.google.com /
",argv[0],argv[0]);
        return 1;
    }

    base = event_base_new();
    dns_base = evdns_base_new(base, 1);

    bev = bufferevent_socket_new(base, -1, BEV_OPT_CLOSE_ON_FREE);
    bufferevent_setcb(bev, readcb, NULL, eventcb, base);
    bufferevent_enable(bev, EV_READ|EV_WRITE);
    evbuffer_add_printf(bufferevent_get_output(bev), "GET %s
", argv[2]);
    bufferevent_socket_connect_hostname(
        bev, dns_base, AF_UNSPEC, argv[1], 80);
    event_base_dispatch(base);
    return 0;
}

常见的bufferevent操作

这节中的函数工作与多个bufferevent

释放bufferevent

接口

`cpp
void bufferevent_free(struct bufferevent *bev);

这个函数释放一个bufferevent资源。Bufferevents是使用引用计数的，也就是如果它在挂起等待延迟回调，你释放它并不会真的删除，直到回调执行完成。

但是`bufferevent_free()`函数确实尽可能的快的释放bufferevent。如果有挂起的数据需要写入到bufferevent，它大概并不会在释放资源之前把数据刷新到bufferevent中。

如果设置了`BEV_OPT_CLOSE_ON_FREE`标志，在释放bufferevent的时候，如果有socket或是底层相关的传输协议，它会在释放的时候关闭这些传输协议。

### 操作callbacks watermarks和enabled operations

#### 接口

```cpp
typedef void (*bufferevent_data_cb)(struct bufferevent *bev, void *ctx);
typedef void (*bufferevent_event_cb)(struct bufferevent *bev,
    short events, void *ctx);

void bufferevent_setcb(struct bufferevent *bufev,
    bufferevent_data_cb readcb, bufferevent_data_cb writecb,
    bufferevent_event_cb eventcb, void *cbarg);

void bufferevent_getcb(struct bufferevent *bufev,
    bufferevent_data_cb *readcb_ptr,
    bufferevent_data_cb *writecb_ptr,
    bufferevent_event_cb *eventcb_ptr,
    void **cbarg_ptr);

bufferevent_setcb()修改bufferevent中一个或是多个回调函数。当有足够的数据可读，readcb就会被调用；当有足够的数据可写，writecb就会被调用；当有事件触发，eventcb就会被调用。每个函数的第一个参数就是已经有事件触发的bufferevent。最后一个参数就是用户在bufferevent_callcb()提供的参数：你可以通过这个参数传递一个数据到回调函数。事件回调函数的事件参数是一个bitmask。

你可以传递一个NULL来禁止回调。记住，在bufferevent中，回调函数公用一个cbarg参数，所以修改一个就会影响所有的。

可以通过传入一个指针到bufferevent_getcb()来获得当前使用的回调函数。readcb_ptr是当前的读回调，writecb_ptr是当前的写回调，eventcb_ptr是当前的事件回调，cbarg_ptr是当前的回调参数。这几个参数，传入NULL的换，就会忽略掉。

接口

void bufferevent_enable(struct bufferevent *bufev, short events);
void bufferevent_disable(struct bufferevent *bufev, short events);

short bufferevent_get_enabled(struct bufferevent *bufev);

你可以禁用或是开启bufferevent的EV_READ EV_WRITE或EV_READ|EV_WRITE事件。当读或是写被禁用时，bufferevent就不会去读或是写数据了。

当缓冲时空的时候，没有必要禁用写数据了：bufferevent会自动的停止写，当有数据可写时在重新开启写的功能。

同样，当input buffer到达了最高的空间大小是，也没必要禁止读数据了，bufferevent会自动停止读，直到有空间可读时在开启读操作。

默认情况下，一个新的bufferevent是可以写的，但是不可以读。

可以调用bufferevent_get_enabled()查看哪些事件被开启了。

接口

void bufferevent_setwatermark(struct bufferevent *bufev, short events,
    size_t lowmark, size_t highmark);

bufferevent_setwatermark()函数可以增加读标识，写标识，或是两个都增加。如果EV_READ被设置到events参数中，那么就增加了读标识。如果EV_WRITE被设置到events参数中，就增加了写标识。

更高水平的标识0表示无限制。

示例

#include <event2/event.h>
#include <event2/bufferevent.h>
#include <event2/buffer.h>
#include <event2/util.h>

#include <stdlib.h>
#include <errno.h>
#include <string.h>

struct info {
    const char *name;
    size_t total_drained;
};

void read_callback(struct bufferevent *bev, void *ctx)
{
    struct info *inf = ctx;
    struct evbuffer *input = bufferevent_get_input(bev);
    size_t len = evbuffer_get_length(input);
    if (len) {
        inf->total_drained += len;
        evbuffer_drain(input, len);
        printf("Drained %lu bytes from %s
",
             (unsigned long) len, inf->name);
    }
}

void event_callback(struct bufferevent *bev, short events, void *ctx)
{
    struct info *inf = ctx;
    struct evbuffer *input = bufferevent_get_input(bev);
    int finished = 0;

    if (events & BEV_EVENT_EOF) {
        size_t len = evbuffer_get_length(input);
        printf("Got a close from %s.  We drained %lu bytes from it, "
            "and have %lu left.
", inf->name,
            (unsigned long)inf->total_drained, (unsigned long)len);
        finished = 1;
    }
    if (events & BEV_EVENT_ERROR) {
        printf("Got an error from %s: %s
",
            inf->name, evutil_socket_error_to_string(EVUTIL_SOCKET_ERROR()));
        finished = 1;
    }
    if (finished) {
        free(ctx);
        bufferevent_free(bev);
    }
}

struct bufferevent *setup_bufferevent(void)
{
    struct bufferevent *b1 = NULL;
    struct info *info1;

    info1 = malloc(sizeof(struct info));
    info1->name = "buffer 1";
    info1->total_drained = 0;

    /* ... Here we should set up the bufferevent and make sure it gets
       connected... */

    /* Trigger the read callback only whenever there is at least 128 bytes
       of data in the buffer. */
    bufferevent_setwatermark(b1, EV_READ, 128, 0);

    bufferevent_setcb(b1, read_callback, NULL, event_callback, info1);

    bufferevent_enable(b1, EV_READ); /* Start reading. */
    return b1;
}

操作数据

如果你不能查看数据的话，从网络读写数据对于你来说都不是好事。Bufferevents提供了给他们数据去写，然后再获得读数据的方法。

接口

struct evbuffer *bufferevent_get_input(struct bufferevent *bufev);
struct evbuffer *bufferevent_get_output(struct bufferevent *bufev);

这两个是非常有用的基本函数：可以返回输入或是输出的buffer。

注意，程序只能从input buffer删除数据，也就是读走，不可以增加；只能从output buffer写入数据，不可以删除。

如果你在写入数据的时候失速了，比如数据太少；或是读取数据的时候失速了，比如数据太多，重新想output buffer增加数据或是从input buffer删除数据都会是bufferevent重启。

接口

int bufferevent_write(struct bufferevent *bufev,
    const void *data, size_t size);
int bufferevent_write_buffer(struct bufferevent *bufev,
    struct evbuffer *buf);

这两个函数是向bufferevent的output buffer写入数据。bufferevent_write()是从内存写入size大小的数据到output buffer的最后。bufferevent_write_buffer()删除所有buf中的数据，然后方知道output buffer最后。

接口

size_t bufferevent_read(struct bufferevent *bufev, void *data, size_t size);
int bufferevent_read_buffer(struct bufferevent *bufev,
    struct evbuffer *buf);

这两个函数是从bufferevent的input buffer删除数据。bufferevent_read()从input buffer中删除size指定的大小的数据，然后把它放到data指定的内存中。返回具体删除了多少数据。bufferevent_read_buffer()清空所有input buffer中的数据，然后放入到buf中。成功返回0，失败返回-1.

注意，bufferevent_read()内存中buf指定的空间必须要大于input buffer中数据的大小。

示例

#include <event2/bufferevent.h>
#include <event2/buffer.h>

#include <ctype.h>

void
read_callback_uppercase(struct bufferevent *bev, void *ctx)
{
        /* This callback removes the data from bev's input buffer 128
           bytes at a time, uppercases it, and starts sending it
           back.

           (Watch out!  In practice, you shouldn't use toupper to implement
           a network protocol, unless you know for a fact that the current
           locale is the one you want to be using.)
         */

        char tmp[128];
        size_t n;
        int i;
        while (1) {
                n = bufferevent_read(bev, tmp, sizeof(tmp));
                if (n <= 0)
                        break; /* No more data. */
                for (i=0; i<n; ++i)
                        tmp[i] = toupper(tmp[i]);
                bufferevent_write(bev, tmp, n);
        }
}

struct proxy_info {
        struct bufferevent *other_bev;
};
void
read_callback_proxy(struct bufferevent *bev, void *ctx)
{
        /* You might use a function like this if you're implementing
           a simple proxy: it will take data from one connection (on
           bev), and write it to another, copying as little as
           possible. */
        struct proxy_info *inf = ctx;

        bufferevent_read_buffer(bev,
            bufferevent_get_output(inf->other_bev));
}

struct count {
        unsigned long last_fib[2];
};

void
write_callback_fibonacci(struct bufferevent *bev, void *ctx)
{
        /* Here's a callback that adds some Fibonacci numbers to the
           output buffer of bev.  It stops once we have added 1k of
           data; once this data is drained, we'll add more. */
        struct count *c = ctx;

        struct evbuffer *tmp = evbuffer_new();
        while (evbuffer_get_length(tmp) < 1024) {
                 unsigned long next = c->last_fib[0] + c->last_fib[1];
                 c->last_fib[0] = c->last_fib[1];
                 c->last_fib[1] = next;

                 evbuffer_add_printf(tmp, "%lu", next);
        }

        /* Now we add the whole contents of tmp to bev. */
        bufferevent_write_buffer(bev, tmp);

        /* We don't need tmp any longer. */
        evbuffer_free(tmp);
}

读写超时

作为其他事件，你可以在超过了指定时间还没有成功读写数据的时候接收到超时事件的回调。

接口

void bufferevent_set_timeouts(struct bufferevent *bufev,
    const struct timeval *timeout_read, const struct timeval *timeout_write);

设置的槽式参数可以是NULL，表示删除超时。

读超时将在等待至少timeout_read指定的秒数之后触发。。写超时将在等待至少timeout_write的秒之后触发

注意，这个超时只是在bufferevent可以操作读写的时候起作用。换句话说，读槽式不会在读被禁止或是input buffer满的情况下（在高水位）触发。同样的，写不会在写被禁止或是没数据可写的时候触发。

当读或是写的超时事件触发时，当前相关的读写操作都会被禁止。事件的回调函数会附带BEV_EVENT_TIMEOUT|BEV_EVENT_READING或BEV_EVENT_TIMEOUT|BEV_EVENT_WRITING回调。

在bufferevent上初始化一个刷新

接口

int bufferevent_flush(struct bufferevent *bufev,
    short iotype, enum bufferevent_flush_mode state);

刷新缓冲就是告诉底层传输协议强制的既可能读或是写更多的数据，忽略那些可能阻止他们读写的限制。

iotype参数的值是EV_READ EV_WRITE或EV_READ|EV_WRITE，用来标明是对读还是写还是两个都要操作。state参数的值有BEV_NORMAL BEV_FLUSH或BEV_FINISHED。BEV_FINISHED告诉另一端不要在发送数据。

出错发挥-1，0表示没数据刷新，1表示有数据刷新。

特定类型的bufferevent函数

这些bufferevent函数并不支持所有的bufferevent类型。

接口

int bufferevent_priority_set(struct bufferevent *bufev, int pri);
int bufferevent_get_priority(struct bufferevent *bufev);

这个函数调整事件的优先级用来实现把bufev调整为pri。

接口

int bufferevent_setfd(struct bufferevent *bufev, evutil_socket_t fd);
evutil_socket_t bufferevent_getfd(struct bufferevent *bufev);

这个函数设置并返回基于fd事件的文件描述符。只有socket类型的bufferevent才支持setfd()。

接口

struct event_base *bufferevent_get_base(struct bufferevent *bev);

返回bufferevent的event_base

接口

struct bufferevent *bufferevent_get_underlying(struct bufferevent *bufev);

返回另一个bufferevent使用的bufferevent。

手动锁和释放bufferevent

在使用evbuffers的时候，有时候你需要确保一些列的bufferevent操作都在一个原子内操作完成。libevent提供了方法可以让你手动的锁或是释放bufferevent。

接口

void bufferevent_lock(struct bufferevent *bufev);
void bufferevent_unlock(struct bufferevent *bufev);

注意，如果bufferevent没有设置BEV_OPT_THREADSAFE标识或是libevent的线程不支持，那么锁定bufferevent将会没任何效果。

锁定bufferevent同样会锁定相关的evbuffers。这个函数是递归的，你可以安全的锁定已经拥有锁的bufferevent。当然，你必须针对每次锁定bufferevent都要调用释放锁的函数。