libevent简介[翻译]10 Bufferevents的高级操作

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

这一章介绍了一些高级的用法，一般情况使用不到，如果你仅仅是学习如何使用bufferevent，请跳过这章，阅读evbuffer的章节。

成对的bufferevent

有时候你需要一个可以告诉自己的网络程序。比如，你可能有一个程序写了一个隧道，这个隧道使用一些用户连接协议，有时候呢，你也想让它连接自己。你可以通过打开一个连接自己监听端口的连接并且让自己程序使用这种方式来实现，这是肯定可以的，但是，这样做会消耗更多的资源，因为你的程序访问自己还需要经过网络

你可以创建一个成对的bufferevent来代替这种方案。所有在一个上面写入的数据，在另一个都可以接收到，反之亦然，但是没有具体的平台的socket使用。

接口

int bufferevent_pair_new(struct event_base *base, int options,
    struct bufferevent *pair[2]);

调用bufferevent_pair_new()，会把参数中pair[0]和pair[1]设置为成对的bufferevents，它们之间相互连接。所有常用的操作都可以使用，除了BEV_OPT_CLOSE_ON_FREE没有效果。BEV_OPT_DEFER_CALLBACKS一直是打开的。

为什么使用成对的bufferevent需要使用延迟回调呢？因为对于一对bufferevent，经常会遇到，一个在回调操作数据，另一个也在回调。如果回调不延迟，这个回调链会很容易的溢出或是饥饿使得所有的回调重新进入。

成对的bufferevent也支持刷新缓冲；设置参数位BEV_NORMAL或BEV_FLUSH都将强制所有相关的数据从一个传输到另一个，忽略可能限制它的watermark。把模式设置为BEV_FINISHED将会是另外一个相对的bufferevent获得EOF事件。

释放其中一个bufferevent并不会自动释放另一个或是使另一个获得EOF的事件；它只会使另一个变成非连接。一旦bufferevent变成非连接，它将不能再读和写或者接收事件。

接口

struct bufferevent *bufferevent_pair_get_partner(struct bufferevent *bev)

有时候你需要通过给定的一个获取另一个bufferevent。你可以调用bufferevent_pair_get_partner()来实现这个功能。如果另一个存在，将会返回与bev成对的另一个bufferevent。否则，返回NULL

过滤型bufferevents

有时候你想把经过bufferevent传输的数据解析一次。你可以增加一个压缩层或是封装层在另一个传输协议中。

接口

enum bufferevent_filter_result {
        BEV_OK = 0,
        BEV_NEED_MORE = 1,
        BEV_ERROR = 2
};
typedef enum bufferevent_filter_result (*bufferevent_filter_cb)(
    struct evbuffer *source, struct evbuffer *destination, ev_ssize_t dst_limit,
    enum bufferevent_flush_mode mode, void *ctx);

struct bufferevent *bufferevent_filter_new(struct bufferevent *underlying,
        bufferevent_filter_cb input_filter,
        bufferevent_filter_cb output_filter,
        int options,
        void (*free_context)(void *),
        void *ctx);

bufferevent_filter_new()创建一个新的过滤bufferevent，包装于存在的底层bufferevent。所有通过底层bufferevent接收的数据都会通过input过滤器，然后传输到过滤bufferevent；所有发送的数据都会先经过过滤bufferevent，然后传送到底层的bufferevent。

位底层bufferevent增加过滤器会替换底层的回调函数。你仍可以位底层的bufferevent的evbuffer设置回调，但是你不能为bufferevent设置回调，不然你的过滤器就不工作了。

所有常用的操作都支持，可以通过options设置。如果BEV_OPT_CLOSE_ON_FREE设置了，释放过滤bufferevnet的时候同样会释放底层的bufferevent。ctx是用于过滤函数优先级的参数。如果free_context提供了，会在过滤bufferevent关闭的时候会掉ctx。

input过滤函数会在底层input buffer有任何可读的数据时调用。output过滤函数会在过滤器的output buffer有新的可写数据时调用。每个都收到一堆evbuffer：一个用来读数据的源evbuffer和一个用来写数据的目的evbuffer。dst_limit指定了增加到目标空间的最大字节数。过滤函数可以忽略这个参数，但是这样会使high-water marks或者速率限制失效。如果dst_limit是-1，表示没限制。mode参数告诉过滤器写入数据的积极性。如果是BEV_NORMAL，应当尽可能写入多的数据方便转换。BEV_FLUSH表示尽可能多的写入，BEV_FINISHED表示过滤函数应该在流结束的时候额外的增加清理工作。最后过滤函数的ctx参数是一个void指针，用来传送bufferevent_filter_new()的构造器中。

过滤函数必须返回BEV_OK如果任何数据成功的写入到目标缓冲中，返回BEV_NEED_MORE如果没有更多的数据写入到目标缓冲，没有获取更多的数据从input或是使用另一个刷新模式，返回BEV_ERROR如果发生了无法修复的问题。

创建过滤器可以使底层的bufferevent同时可读和可写。你不需要自己操作读写：过滤器会在你不想读的时候延迟底层的bufferevent的读操作。2.0.8-rc以后的版本，允许单独的打开禁止读和写对于底层的bufferevent。如果你这样做，保持过滤器成功的获取到它想要的数据。

你不用必须设置input filter和output filter：任何一个没哟设置，都会被一个直接传输数据不做任何其他操作的过滤器替代。

限制一次最大读写数据

默认情况下，bufferevent不会在每个事件循环调用中读写超过最大字节数的数据；这样做会导致不可预料的行为，并且会出现资源饥饿的现象。另外一方面，默认的设置有时候并不是适合所有的情况。

接口

int bufferevent_set_max_single_read(struct bufferevent *bev, size_t size);
int bufferevent_set_max_single_write(struct bufferevent *bev, size_t size);

ev_ssize_t bufferevent_get_max_single_read(struct bufferevent *bev);
ev_ssize_t bufferevent_get_max_single_write(struct bufferevent *bev);

两个set函数就是分别替换读和写的最大值。如果是0或是超过了EV_SSIZE_MAX，就设置为最大值。

两个get函数就是获得当前loop的读写最大值。

Bufferevent和速率限制

有些程序在使用单个bufferevent或者一组bufferevent的时候需要限制总的带宽

速率限制模式

libevent的速率限制模式使用令牌桶的方式来决定一次读写多少字节。每一个速率限制对象，在任何时候，都有一个读桶和写桶。这两个桶分别决定可以读或是写多少字节。每个桶都有一个注满的速率，一个最大的溢出大小和一个时间单元或是钟摆。当时间单元用完的时候，桶会根据注满速率成比例的注满，但是如果超过了溢出大小，超过的数据将会丢失。

因此，注满速率决定了对象发送或接收字节时候的最大平均速率；溢出大小决定了一个桶最大可发送或是接收的数据大小。时间单元决定了传输的平滑性。

为bufferevent设置速率限制

接口

#define EV_RATE_LIMIT_MAX EV_SSIZE_MAX
struct ev_token_bucket_cfg;
struct ev_token_bucket_cfg *ev_token_bucket_cfg_new(
        size_t read_rate, size_t read_burst,
        size_t write_rate, size_t write_burst,
        const struct timeval *tick_len);
void ev_token_bucket_cfg_free(struct ev_token_bucket_cfg *cfg);
int bufferevent_set_rate_limit(struct bufferevent *bev,
    struct ev_token_bucket_cfg *cfg);

ev_token_bucket_cfg结构体中设置了一对令牌桶在一个bufferevent或是一组bufferevent上读写速率限制的配置信息。可以调用ev_token_bucket_cfg_new函数创建一个，需要提供最大平均读速率，最大读溢出，最大写速率，最大写溢出和钟摆长度。如果tick_len是NULL，那么钟摆长度默认是一秒。出错返回NULL。

注意，read_rate和write_rate是根据一个钟摆的字节数来衡量的。比如，钟摆是十分之一秒，read_rate是300，那么最大平均读速率就是3000字节每秒。比率和溢出数据在EV_RATE_LIMIT_MAX参数上不支持。

如果想限制bufferevent的传输速率，调用bufferevent_set_rate_limit()函数并附带ev_token_bucket_cfg参数。你可以给多个bufferevent传递同一个ev_token_bucket_cfg参数。调用bufferevent_set_rate_limit()，传递cfg为NULL可以去除速率限制。

调用ev_token_bucket_cfg_free()释放ev_token_bucket_cfg。如果有bufferevent在使用ev_token_bucket_cfg，是不建议做这个操作，这样是不安全的。

为一组bufferevent设置速率限制

如果你想限制总的带宽使用，你可以把一组bufferevent关联到一个限制速率群内。

接口

struct bufferevent_rate_limit_group;

struct bufferevent_rate_limit_group *bufferevent_rate_limit_group_new(
        struct event_base *base,
        const struct ev_token_bucket_cfg *cfg);
int bufferevent_rate_limit_group_set_cfg(
        struct bufferevent_rate_limit_group *group,
        const struct ev_token_bucket_cfg *cfg);
void bufferevent_rate_limit_group_free(struct bufferevent_rate_limit_group *);
int bufferevent_add_to_rate_limit_group(struct bufferevent *bev,
    struct bufferevent_rate_limit_group *g);
int bufferevent_remove_from_rate_limit_group(struct bufferevent *bev);

可以调用bufferevent_rate_limit_group()，附带一个event_base参数，一个初始化的ev_token_bucket_cfg，可以创建一个速率限制群组。可以调用bufferevent_add_to_rate_limit_group()和bufferevent_remove_from_rate_limit_group()分别把一组bufferevent添加到速率限制群或是从速率限制群删除。

一个bufferevent可以同一时间称为一个速率限制群的成员。一个bufferevent可以同时有一个个人的速率限制和群体的速率限制。当两个都限制了，更低的那一个起作用。

可以通过bufferevent_rate_limit_group_set_cfg()修改一个已经存在的速率限制群的速率。可以调用bufferevent_rate_limit_group_free()释放一个速率限制群，并且删除所有的成员。

查看当前的速率值

接口

ev_ssize_t bufferevent_get_read_limit(struct bufferevent *bev);
ev_ssize_t bufferevent_get_write_limit(struct bufferevent *bev);
ev_ssize_t bufferevent_rate_limit_group_get_read_limit(
        struct bufferevent_rate_limit_group *);
ev_ssize_t bufferevent_rate_limit_group_get_write_limit(
        struct bufferevent_rate_limit_group *);

上面的函数返回当前bufferevent或是群组中读写令牌桶的字节大小。注意，如果bufferevnet被强制的赋予超过申请空间的数据，可能返回值是负数。

接口

ev_ssize_t bufferevent_get_max_to_read(struct bufferevent *bev);
ev_ssize_t bufferevent_get_max_to_write(struct bufferevent *bev);

这个函数返回当前bufferevent可能读写的字节数，考虑所有的bufferevent的速率限制，如果有，还包括群限制，最大可读，最大写。

接口

void bufferevent_rate_limit_group_get_totals(
    struct bufferevent_rate_limit_group *grp,
    ev_uint64_t *total_read_out, ev_uint64_t *total_written_out);
void bufferevent_rate_limit_group_reset_totals(
    struct bufferevent_rate_limit_group *grp);

每个bufferevent_rate_limit_group都记录所有通过它传输的数据。你可以使用他来跟踪在一个组中的bufferevent中总的使用量。调用bufferevent_rate_limit_group_get_totals()可以为群组设置total_read_out和total_written_out，分别用来记录总的读写数据。这个总数在创建的时候是0，调用bufferevent_rate_limit_group_reset_totals()的时候也会设置为0.

手动调整速率限制

有时候，你需要调整当前的令牌桶数值，比如你的程序不是从bufferevent获取数据。

接口

int bufferevent_decrement_read_limit(struct bufferevent *bev, ev_ssize_t decr);
int bufferevent_decrement_write_limit(struct bufferevent *bev, ev_ssize_t decr);
int bufferevent_rate_limit_group_decrement_read(
        struct bufferevent_rate_limit_group *grp, ev_ssize_t decr);
int bufferevent_rate_limit_group_decrement_write(
        struct bufferevent_rate_limit_group *grp, ev_ssize_t decr);

这些函数用来设置bufferevent或是速率限制群组的读或是写的桶。注意缩减量是有符号的：如果想增加桶，可以传入负值。

在速率限制群组中设置最小共享

经常你不想对于同一个速率限制群组中的每一个bufferevent的每一个时钟平分总的限制字节。比如，你又10000个活动的bufferevent，这个速率限制群组是10000字节，你不想让每一个bufferevent都是一个时钟传输一个字节。

为了解决这个问题，每一个速率限制群组都有一个最小共享。用这个方案，必须要每个bufferevent每个时钟写一个字节，10000/SHARE的bufferevent将会允许写入共享的字节，其余的什么也不写入。每个时钟都允许随机的选择一个写入的bufferevent。

接口

int bufferevent_rate_limit_group_set_min_share(
        struct bufferevent_rate_limit_group *group, size_t min_share);

设置min_share是0可以禁止这个最小共享代码。

速率限制的局限性

在libevent 2.0的版本中，有一些速率限制的局限性：

• 并不是每一个bufferevent都支持速率限制

• bufferevent的速率限制群组不能嵌套，一个bufferevent只能在同一时间内有一个速率限制群组

• 这个速率限制只会计算TCP包的数据部分，并不计算TCP的头

• 速率限制的实现依赖于TCP堆的提示，也就是程序只能在当前速率下消耗数据，当缓存满的时候，把数据推送给TCP连接的另一端。

• 一些bufferevent的实现，比如windows的IOCP，可能会过度提交

• 桶从一个满的时钟传输开始。也就是bufferevent可以立即开始读写，不用等待一个完整的时钟经过。还表示，bufferevent为N.1的速率限制，可能会传输到N+1的时钟。

• 时钟不能小于1毫秒，所有的毫秒的分数都会被忽略

Bufferevent和SSL

Bufferevent可以使用OpenSSL来实现SSL/TLS加密传输。因为很多程序不需要或是不想连接OpenSSL，这个函数被独立的拆分到libevent_openssl库。将来的版本可能增加其他的SSL/TLS安全库，比如NSS或是GnuTLS，但是当前只有OpenSSL。

所有的这些函数都定义在"event2/bufferevent_ssl.h"

设置并使用基于OpenSSL的bufferevent

接口

enum bufferevent_ssl_state {
        BUFFEREVENT_SSL_OPEN = 0,
        BUFFEREVENT_SSL_CONNECTING = 1,
        BUFFEREVENT_SSL_ACCEPTING = 2
};

struct bufferevent *
bufferevent_openssl_filter_new(struct event_base *base,
    struct bufferevent *underlying,
    SSL *ssl,
    enum bufferevent_ssl_state state,
    int options);

struct bufferevent *
bufferevent_openssl_socket_new(struct event_base *base,
    evutil_socket_t fd,
    SSL *ssl,
    enum bufferevent_ssl_state state,
    int options);

你可以创建两种SSL的bufferevent：一种是基于过滤器的，通过另一个底层bufferevent来通信；另一种是基于socket的，通过网络底层来通信。不管怎么样，你都必须提供SSL的对象和SSL状态的描述。这个状态可能是BUFFEREVENT_SSL_CONNECTING，如果SSL当前被当做一个连接；如果是BUFFEREVENT_SSL_ACCEPTING，表示SSL当前被当做一个服务端；或是BUFFEREVENT_SSL_OPEN，SSL握手已经完成。

常见的设置是被接收的：BEV_OPT_CLOSE_ON_FREE使SSL对象和底层fd或者bufferevent在openssl bufferevent自己关闭的时候同时关闭。

当握手完成后，新的bufferevent事件会通过回调函数传递一个BEV_EVENT_CONNECTED标识回调。

如果你创建了基于socket的bufferevent并且SSL对象已经有socket了，你可以不提供socket，传入-1，稍后通过bufferevent_setfd()传入fd。

注意，BEV_OPT_CLOSE_ON_FREE是设置在SSL的bufferevent上，不需要专门关闭SSL的连接。这里有两个问题：一个是连接可能会被另一端破坏，因为没有明确的关闭，所以另一端不能告诉你这个连接是关闭了还是被攻击了还是被第三方关闭了。第二个是，OpenSSL将会认为这个session是坏的，并且从session的cache中移除。这可能会导致SSL程序性能下降。

当前，可做的就是手动延迟SLL关闭。但是这样做破坏了TLS RFC，需要确保在关闭的时候session在cache中。下面的代码实现了这个步骤：

示例

SSL *ctx = bufferevent_openssl_get_ssl(bev);

/*
 * SSL_RECEIVED_SHUTDOWN tells SSL_shutdown to act as if we had already
 * received a close notify from the other end.  SSL_shutdown will then
 * send the final close notify in reply.  The other end will receive the
 * close notify and send theirs.  By this time, we will have already
 * closed the socket and the other end's real close notify will never be
 * received.  In effect, both sides will think that they have completed a
 * clean shutdown and keep their sessions valid.  This strategy will fail
 * if the socket is not ready for writing, in which case this hack will
 * lead to an unclean shutdown and lost session on the other end.
 */
SSL_set_shutdown(ctx, SSL_RECEIVED_SHUTDOWN);
SSL_shutdown(ctx);
bufferevent_free(bev);

接口

SSL *bufferevent_openssl_get_ssl(struct bufferevent *bev);

这个将会返回OpenSSL bufferevent使用的SSL对象，如果没有设置bev，就返回NULL。

接口

unsigned long bufferevent_get_openssl_error(struct bufferevent *bev);

这个函数返回第一个关起的OpenSSL的错误，从bufferevent的操作中；如果没有错误，就返回0.错误在返回时经过OpenSSL类库中的ERR_get_error()做了格式化。

接口

int bufferevent_ssl_renegotiate(struct bufferevent *bev);

调用这个函数会告诉SSL重新协商，并且bufferevent调用对应的回调函数。这是一个高级用法，除非你知道如何做，必然的话不要使用这个方法，并且很多SSL的版本有重新协商的bug。

接口

int bufferevent_openssl_get_allow_dirty_shutdown(struct bufferevent *bev);
void bufferevent_openssl_set_allow_dirty_shutdown(struct bufferevent *bev,
    int allow_dirty_shutdown);

所有优秀的SLL协议版本，包括SSLv3和TLS版本，都支持认证的关闭操作，这样可以避免非法意外关闭或是恶意的底层中止。默认情况下，我们把除了关闭的所有错误都认为是连接上的。如果allow_dirty_shutdown标识设置位1，我们也会把关闭作为一个BEV_EVENT_EOF错误。

示例：一个简单的基于SSL的echo服务端

/* Simple echo server using OpenSSL bufferevents */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#include <openssl/ssl.h>
#include <openssl/err.h>
#include <openssl/rand.h>

#include <event.h>
#include <event2/listener.h>
#include <event2/bufferevent_ssl.h>

static void
ssl_readcb(struct bufferevent * bev, void * arg)
{
    struct evbuffer *in = bufferevent_get_input(bev);

    printf("Received %zu bytes
", evbuffer_get_length(in));
    printf("----- data ----
");
    printf("%.*s
", (int)evbuffer_get_length(in), evbuffer_pullup(in, -1));

    bufferevent_write_buffer(bev, in);
}

static void
ssl_acceptcb(struct evconnlistener *serv, int sock, struct sockaddr *sa,
             int sa_len, void *arg)
{
    struct event_base *evbase;
    struct bufferevent *bev;
    SSL_CTX *server_ctx;
    SSL *client_ctx;

    server_ctx = (SSL_CTX *)arg;
    client_ctx = SSL_new(server_ctx);
    evbase = evconnlistener_get_base(serv);

    bev = bufferevent_openssl_socket_new(evbase, sock, client_ctx,
                                         BUFFEREVENT_SSL_ACCEPTING,
                                         BEV_OPT_CLOSE_ON_FREE);

    bufferevent_enable(bev, EV_READ);
    bufferevent_setcb(bev, ssl_readcb, NULL, NULL, NULL);
}

static SSL_CTX *
evssl_init(void)
{
    SSL_CTX  *server_ctx;

    /* Initialize the OpenSSL library */
    SSL_load_error_strings();
    SSL_library_init();
    /* We MUST have entropy, or else there's no point to crypto. */
    if (!RAND_poll())
        return NULL;

    server_ctx = SSL_CTX_new(SSLv23_server_method());

    if (! SSL_CTX_use_certificate_chain_file(server_ctx, "cert") ||
        ! SSL_CTX_use_PrivateKey_file(server_ctx, "pkey", SSL_FILETYPE_PEM)) {
        puts("Couldn't read 'pkey' or 'cert' file.  To generate a key
"
           "and self-signed certificate, run:
"
           "  openssl genrsa -out pkey 2048
"
           "  openssl req -new -key pkey -out cert.req
"
           "  openssl x509 -req -days 365 -in cert.req -signkey pkey -out cert");
        return NULL;
    }
    SSL_CTX_set_options(server_ctx, SSL_OP_NO_SSLv2);

    return server_ctx;
}

int
main(int argc, char **argv)
{
    SSL_CTX *ctx;
    struct evconnlistener *listener;
    struct event_base *evbase;
    struct sockaddr_in sin;

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

    ctx = evssl_init();
    if (ctx == NULL)
        return 1;
    evbase = event_base_new();
    listener = evconnlistener_new_bind(
                         evbase, ssl_acceptcb, (void *)ctx,
                         LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE, 1024,
                         (struct sockaddr *)&sin, sizeof(sin));

    event_base_loop(evbase, 0);

    evconnlistener_free(listener);
    SSL_CTX_free(ctx);

    return 0;
}

libevent在多线程模式下使用OpenSSL并没有加锁。因为OpenSSL使用了大量全局变量，你必须确保OpenSSL是线程安全的。

示例：一个非常简单的线程安全OpenSSL

/*
 * Please refer to OpenSSL documentation to verify you are doing this correctly,
 * Libevent does not guarantee this code is the complete picture, but to be used
 * only as an example.
 */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <pthread.h>
#include <openssl/ssl.h>
#include <openssl/crypto.h>

pthread_mutex_t * ssl_locks;
int ssl_num_locks;

/* Implements a thread-ID function as requied by openssl */
static unsigned long
get_thread_id_cb(void)
{
    return (unsigned long)pthread_self();
}

static void
thread_lock_cb(int mode, int which, const char * f, int l)
{
    if (which < ssl_num_locks) {
        if (mode & CRYPTO_LOCK) {
            pthread_mutex_lock(&(ssl_locks[which]));
        } else {
            pthread_mutex_unlock(&(ssl_locks[which]));
        }
    }
}

int
init_ssl_locking(void)
{
    int i;

    ssl_num_locks = CRYPTO_num_locks();
    ssl_locks = malloc(ssl_num_locks * sizeof(pthread_mutex_t));
    if (ssl_locks == NULL)
        return -1;

    for (i = 0; i < ssl_num_locks; i++) {
        pthread_mutex_init(&(ssl_locks[i]), NULL);
    }

    CRYPTO_set_id_callback(get_thread_id_cb);
    CRYPTO_set_locking_callback(thread_lock_cb);

    return 0;
}