Libevent源码分析（六）--- bufferevent

上一节说过，libevent提供六种bufferevent类型，后面会详细分析其中的两个：bufferevent_sock和bufferevent_async.下面是bufferevent的详细定义：

struct bufferevent {
/** Event base for which this bufferevent was created. */
struct event_base *ev_base;
/** Pointer to a table of function pointers to set up how this
bufferevent behaves. */
const struct bufferevent_ops *be_ops;

/** A read event that triggers when a timeout has happened or a socket
is ready to read data.  Only used by some subtypes of
bufferevent. */
struct event ev_read;
/** A write event that triggers when a timeout has happened or a socket
is ready to write data.  Only used by some subtypes of
bufferevent. */
struct event ev_write;

/** An input buffer. Only the bufferevent is allowed to add data to
this buffer, though the user is allowed to drain it. */
struct evbuffer *input;

/** An input buffer. Only the bufferevent is allowed to drain data
from this buffer, though the user is allowed to add it. */
struct evbuffer *output;

struct event_watermark wm_read;
struct event_watermark wm_write;

bufferevent_data_cb readcb;
bufferevent_data_cb writecb;
/* This should be called 'eventcb', but renaming it would break
* backward compatibility */
bufferevent_event_cb errorcb;
void *cbarg;

struct timeval timeout_read;
struct timeval timeout_write;

/** Events that are currently enabled: currently EV_READ and EV_WRITE
are supported. */
short enabled;
};

其中ev_base指向bufferevent所属的event_base ,ev_read,ev_write是读写事件，timeout_read和timeout_write是读写超时时间，readcb，writecb和errorcb分别是读回调，写回调和事件回调，因为版本兼容问题使用errorcb命名。input和output是evbuffer类型的读写缓存。enabled表示支持的事件，目前只有EV_READ 和EV_WRITE。wm_read和wm_write代表读写缓存的伐值，be_ops是bufferevent_ops类型变量，它的定义如下：

struct bufferevent_ops {
/** The name of the bufferevent's type. */
const char *type;
/** At what offset into the implementation type will we find a
bufferevent structure?

Example: if the type is implemented as
struct bufferevent_x {
int extra_data;
struct bufferevent bev;
}
then mem_offset should be offsetof(struct bufferevent_x, bev)
*/
off_t mem_offset;

/** Enables one or more of EV_READ|EV_WRITE on a bufferevent.  Does
not need to adjust the 'enabled' field.  Returns 0 on success, -1
on failure.
*/
int (*enable)(struct bufferevent *, short);

/** Disables one or more of EV_READ|EV_WRITE on a bufferevent.  Does
not need to adjust the 'enabled' field.  Returns 0 on success, -1
on failure.
*/
int (*disable)(struct bufferevent *, short);

/** Free any storage and deallocate any extra data or structures used
in this implementation.
*/
void (*destruct)(struct bufferevent *);

/** Called when the timeouts on the bufferevent have changed.*/
int (*adj_timeouts)(struct bufferevent *);

/** Called to flush data. */
int (*flush)(struct bufferevent *, short, enum bufferevent_flush_mode);

/** Called to access miscellaneous fields. */
int (*ctrl)(struct bufferevent *, enum bufferevent_ctrl_op, union bufferevent_ctrl_data *);

};

bufferevent_ops 的实现和eventop比较类似，都是定义了一组指针，不同的bufferevent实现可以自定义这些具体操作。比如bufferevent_sock的对应定义：

const struct bufferevent_ops bufferevent_ops_socket = {
"socket",
evutil_offsetof(struct bufferevent_private, bev),
be_socket_enable,
be_socket_disable,
be_socket_destruct,
be_socket_adj_timeouts,
be_socket_flush,
be_socket_ctrl,
};

接下来分析一下buffevent_sock的实现方式。

struct bufferevent *
bufferevent_new(evutil_socket_t fd,
bufferevent_data_cb readcb, bufferevent_data_cb writecb,
bufferevent_event_cb eventcb, void *cbarg)
{
struct bufferevent *bufev;

if (!(bufev = bufferevent_socket_new(NULL, fd, 0)))
return NULL;

bufferevent_setcb(bufev, readcb, writecb, eventcb, cbarg);

return bufev;
}

struct bufferevent *
bufferevent_socket_new(struct event_base *base, evutil_socket_t fd,
int options)
{
struct bufferevent_private *bufev_p;
struct bufferevent *bufev;

#ifdef WIN32
if (base && event_base_get_iocp(base))
return bufferevent_async_new(base, fd, options);
#endif

if ((bufev_p = mm_calloc(1, sizeof(struct bufferevent_private)))== NULL)
return NULL;

if (bufferevent_init_common(bufev_p, base, &bufferevent_ops_socket,
options) < 0) {
mm_free(bufev_p);
return NULL;
}
bufev = &bufev_p->bev;
evbuffer_set_flags(bufev->output, EVBUFFER_FLAG_DRAINS_TO_FD);

event_assign(&bufev->ev_read, bufev->ev_base, fd,
EV_READ|EV_PERSIST, bufferevent_readcb, bufev);
event_assign(&bufev->ev_write, bufev->ev_base, fd,
EV_WRITE|EV_PERSIST, bufferevent_writecb, bufev);

evbuffer_add_cb(bufev->output, bufferevent_socket_outbuf_cb, bufev);

evbuffer_freeze(bufev->input, 0);
evbuffer_freeze(bufev->output, 1);

return bufev;
}

bufferevent_socket_new方法用于创建一个bufferevent_socket类型的bufferevent。如果是在win32环境下则会默认调用iocp的对应实现，即bufferevent_async_new。该方法还会创建一个bufferevent_private变量，下面是他的定义：

struct bufferevent_private {
/** The underlying bufferevent structure. */
struct bufferevent bev;

/** Evbuffer callback to enforce watermarks on input. */
struct evbuffer_cb_entry *read_watermarks_cb;

/** If set, we should free the lock when we free the bufferevent. */
unsigned own_lock : 1;
/** Flag: set if we have deferred callbacks and a read callback is
* pending. */
unsigned readcb_pending : 1;
/** Flag: set if we have deferred callbacks and a write callback is
* pending. */
unsigned writecb_pending : 1;
/** Flag: set if we are currently busy connecting. */
unsigned connecting : 1;
/** Flag: set if a connect failed prematurely; this is a hack for
* getting around the bufferevent abstraction. */
unsigned connection_refused : 1;
/** Set to the events pending if we have deferred callbacks and
* an events callback is pending. */
short eventcb_pending;

/** If set, read is suspended until one or more conditions are over.
* The actual value here is a bitfield of those conditions; see the
* BEV_SUSPEND_* flags above. */
bufferevent_suspend_flags read_suspended;

/** If set, writing is suspended until one or more conditions are over.
* The actual value here is a bitfield of those conditions; see the
* BEV_SUSPEND_* flags above. */
bufferevent_suspend_flags write_suspended;

/** Set to the current socket errno if we have deferred callbacks and
* an events callback is pending. */
int errno_pending;

/** The DNS error code for bufferevent_socket_connect_hostname */
int dns_error;

/** Used to implement deferred callbacks */
struct deferred_cb deferred;

/** The options this bufferevent was constructed with */
enum bufferevent_options options;

/** Current reference count for this bufferevent. */
int refcnt;

/** Lock for this bufferevent.  Shared by the inbuf and the outbuf.
* If NULL, locking is disabled. */
void *lock;

/** Rate-limiting information for this bufferevent */
struct bufferevent_rate_limit *rate_limiting;
};

bufferevent_private结构的第一个变量是bufferevent类型的，其实每一个bufferevent都对应一个bufferevent_private变量，只是对用户来说是透明的，用户只需要了解bufferevent即可。bufferevent_private的变量基本都是用于记录状态。

继续分析bufferevent_socket_new函数，创建bufferevent_private之后调用bufferevent_init_common函数，这个函数定义在bufferevent.c文件中，该文件中的函数都是各个bufferevent类型通用的函数。

int bufferevent_init_common(struct bufferevent_private *bufev_private,
struct event_base *base,
const struct bufferevent_ops *ops,
enum bufferevent_options options)
{
struct bufferevent *bufev = &bufev_private->bev;

if (!bufev->input) {
if ((bufev->input = evbuffer_new()) == NULL)
return -1;
}

if (!bufev->output) {
if ((bufev->output = evbuffer_new()) == NULL) {
evbuffer_free(bufev->input);
return -1;
}
}

bufev_private->refcnt = 1;
bufev->ev_base = base;

/* Disable timeouts. */
evutil_timerclear(&bufev->timeout_read);
evutil_timerclear(&bufev->timeout_write);

bufev->be_ops = ops;

/*
* Set to EV_WRITE so that using bufferevent_write is going to
* trigger a callback.  Reading needs to be explicitly enabled
* because otherwise no data will be available.
*/
bufev->enabled = EV_WRITE;

#ifndef _EVENT_DISABLE_THREAD_SUPPORT
if (options & BEV_OPT_THREADSAFE) {
if (bufferevent_enable_locking(bufev, NULL) < 0) {
/* cleanup */
evbuffer_free(bufev->input);
evbuffer_free(bufev->output);
bufev->input = NULL;
bufev->output = NULL;
return -1;
}
}
#endif
if ((options & (BEV_OPT_DEFER_CALLBACKS|BEV_OPT_UNLOCK_CALLBACKS))
== BEV_OPT_UNLOCK_CALLBACKS) {
event_warnx("UNLOCK_CALLBACKS requires DEFER_CALLBACKS");
return -1;
}
if (options & BEV_OPT_DEFER_CALLBACKS) {
if (options & BEV_OPT_UNLOCK_CALLBACKS)
event_deferred_cb_init(&bufev_private->deferred,
bufferevent_run_deferred_callbacks_unlocked,
bufev_private);
else
event_deferred_cb_init(&bufev_private->deferred,
bufferevent_run_deferred_callbacks_locked,
bufev_private);
}

bufev_private->options = options;

evbuffer_set_parent(bufev->input, bufev);
evbuffer_set_parent(bufev->output, bufev);

return 0;
}

bufferevent_init_common比较简单，需要注意 bufferEvent默认开启EV_WRITE，EV_READ需要手动打开，如果需要支持多线程则为bufferevent创建锁，另外在目前的版本中，UNLOCK_CALLBACKS需要DEFER_CALLBACKS作为前置条件。

继续分析bufferevent_socket_new函数：

evbuffer_set_flags(bufev->output, EVBUFFER_FLAG_DRAINS_TO_FD);
event_assign(&bufev->ev_read, bufev->ev_base, fd,
EV_READ|EV_PERSIST, bufferevent_readcb, bufev);
event_assign(&bufev->ev_write, bufev->ev_base, fd,
EV_WRITE|EV_PERSIST, bufferevent_writecb, bufev);

evbuffer_add_cb(bufev->output, bufferevent_socket_outbuf_cb, bufev);

调用bufferevent_init_common之后会将bufev->output添加EVBUFFER_FLAG_DRAINS_TO_FD标记，这样output就支持从文件读取数据了，接下来设置读写事件回掉，最后添加bufferevent_socket_outbuf_cb作为output的call，当output中有数据时调用bufferevent_socket_outbuf_cb，这样就可以不必一直监听写事件，bufferevent_writecb在数据全部写出之后可以移除写事件， 然后在bufferevent_socket_outbuf_cb中判断是否需要添加事件。

初始化bufferevent之后需要调用be_socket_setfd设置网络套接字来监听读写事件：

static void
be_socket_setfd(struct bufferevent *bufev, evutil_socket_t fd)
{
BEV_LOCK(bufev);
EVUTIL_ASSERT(bufev->be_ops == &bufferevent_ops_socket);

event_del(&bufev->ev_read);
event_del(&bufev->ev_write);

event_assign(&bufev->ev_read, bufev->ev_base, fd,
EV_READ|EV_PERSIST, bufferevent_readcb, bufev);
event_assign(&bufev->ev_write, bufev->ev_base, fd,
EV_WRITE|EV_PERSIST, bufferevent_writecb, bufev);

if (fd >= 0)
bufferevent_enable(bufev, bufev->enabled);

BEV_UNLOCK(bufev);
}
int bufferevent_enable(struct bufferevent *bufev, short event)
{
struct bufferevent_private *bufev_private =
EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
short impl_events = event;
int r = 0;

_bufferevent_incref_and_lock(bufev);
if (bufev_private->read_suspended)
impl_events &= ~EV_READ;
if (bufev_private->write_suspended)
impl_events &= ~EV_WRITE;

bufev->enabled |= event;

if (impl_events && bufev->be_ops->enable(bufev, impl_events) < 0)
r = -1;

_bufferevent_decref_and_unlock(bufev);
return r;
}

be_socket_setfd中设置的套接字必须是已经建立好连接的socket。如果是client端，bufferevent_sock提供了bufferevent_socket_connect用于连接套接字，但是如果是server端，libevent没有提供listen的相关函数，不过libevent提供了listener.c文件可以用于监听连接。

int
bufferevent_socket_connect(struct bufferevent *bev,
struct sockaddr *sa, int socklen)
{
struct bufferevent_private *bufev_p =
EVUTIL_UPCAST(bev, struct bufferevent_private, bev);

evutil_socket_t fd;
int r = 0;
int result=-1;
int ownfd = 0;

_bufferevent_incref_and_lock(bev);

if (!bufev_p)
goto done;

fd = bufferevent_getfd(bev);
if (fd < 0) {
if (!sa)
goto done;
fd = socket(sa->sa_family, SOCK_STREAM, 0);
if (fd < 0)
goto done;
if (evutil_make_socket_nonblocking(fd)<0)
goto done;
ownfd = 1;
}
if (sa) {
#ifdef WIN32
if (bufferevent_async_can_connect(bev)) {
bufferevent_setfd(bev, fd);
r = bufferevent_async_connect(bev, fd, sa, socklen);
if (r < 0)
goto freesock;
bufev_p->connecting = 1;
result = 0;
goto done;
} else
#endif
r = evutil_socket_connect(&fd, sa, socklen);
if (r < 0)
goto freesock;
}
#ifdef WIN32
/* ConnectEx() isn't always around, even when IOCP is enabled.
* Here, we borrow the socket object's write handler to fall back
* on a non-blocking connect() when ConnectEx() is unavailable. */
if (BEV_IS_ASYNC(bev)) {
event_assign(&bev->ev_write, bev->ev_base, fd,
EV_WRITE|EV_PERSIST, bufferevent_writecb, bev);
}
#endif
bufferevent_setfd(bev, fd);
if (r == 0) {
if (! be_socket_enable(bev, EV_WRITE)) {
bufev_p->connecting = 1;
result = 0;
goto done;
}
} else if (r == 1) {
/* The connect succeeded already. How very BSD of it. */
result = 0;
bufev_p->connecting = 1;
event_active(&bev->ev_write, EV_WRITE, 1);
} else {
/* The connect failed already.  How very BSD of it. */
bufev_p->connection_refused = 1;
bufev_p->connecting = 1;
result = 0;
event_active(&bev->ev_write, EV_WRITE, 1);
}

goto done;

freesock:
_bufferevent_run_eventcb(bev, BEV_EVENT_ERROR);
if (ownfd)
evutil_closesocket(fd);
/* do something about the error? */
done:
_bufferevent_decref_and_unlock(bev);
return result;
}

win32的connect使用iocp实现，将在下一节iocp章节详细分析，这里只看其他平台的实现方式。evutil_socket_connect是libevent提供的工具函数，它的返回值有三种情况：0代表正在连接，此时需要调用be_socket_enable激活写事件。当连接成功时会调用bufferevent_writecb回掉。1代表已经连接成功，此时调用event_active直接手动触发bufferevent_writecb事件处理连接。-1代表连接被拒绝，此时同样调用event_active处理连接失败情况。

下面是bufferevent_writecb的具体实现：

static void
bufferevent_writecb(evutil_socket_t fd, short event, void *arg)
{
struct bufferevent *bufev = arg;
struct bufferevent_private *bufev_p =
EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
int res = 0;
short what = BEV_EVENT_WRITING;
int connected = 0;
ev_ssize_t atmost = -1;

_bufferevent_incref_and_lock(bufev);

if (event == EV_TIMEOUT) {
/* Note that we only check for event==EV_TIMEOUT. If
* event==EV_TIMEOUT|EV_WRITE, we can safely ignore the
* timeout, since a read has occurred */
what |= BEV_EVENT_TIMEOUT;
goto error;
}
if (bufev_p->connecting) {
int c = evutil_socket_finished_connecting(fd);
/* we need to fake the error if the connection was refused
* immediately - usually connection to localhost on BSD */
if (bufev_p->connection_refused) {
bufev_p->connection_refused = 0;
c = -1;
}

if (c == 0)
goto done;

bufev_p->connecting = 0;
if (c < 0) {
event_del(&bufev->ev_write);
event_del(&bufev->ev_read);
_bufferevent_run_eventcb(bufev, BEV_EVENT_ERROR);
goto done;
} else {
connected = 1;
#ifdef WIN32
if (BEV_IS_ASYNC(bufev)) {
event_del(&bufev->ev_write);
bufferevent_async_set_connected(bufev);
_bufferevent_run_eventcb(bufev,
BEV_EVENT_CONNECTED);
goto done;
}
#endif
_bufferevent_run_eventcb(bufev,
BEV_EVENT_CONNECTED);
if (!(bufev->enabled & EV_WRITE) ||
bufev_p->write_suspended) {
event_del(&bufev->ev_write);
goto done;
}
}
}

atmost = _bufferevent_get_write_max(bufev_p);

if (bufev_p->write_suspended)
goto done;

if (evbuffer_get_length(bufev->output)) {
evbuffer_unfreeze(bufev->output, 1);
res = evbuffer_write_atmost(bufev->output, fd, atmost);
evbuffer_freeze(bufev->output, 1);
if (res == -1) {
int err = evutil_socket_geterror(fd);
if (EVUTIL_ERR_RW_RETRIABLE(err))
goto reschedule;
what |= BEV_EVENT_ERROR;
} else if (res == 0) {
/* eof case
XXXX Actually, a 0 on write doesn't indicate
an EOF. An ECONNRESET might be more typical.
*/
what |= BEV_EVENT_EOF;
}
if (res <= 0)
goto error;

_bufferevent_decrement_write_buckets(bufev_p, res);
}

if (evbuffer_get_length(bufev->output) == 0) {
event_del(&bufev->ev_write);
}

/*
* Invoke the user callback if our buffer is drained or below the
* low watermark.
*/
if ((res || !connected) &&
evbuffer_get_length(bufev->output) <= bufev->wm_write.low) {
_bufferevent_run_writecb(bufev);
}

goto done;

reschedule:
if (evbuffer_get_length(bufev->output) == 0) {
event_del(&bufev->ev_write);
}
goto done;

error:
bufferevent_disable(bufev, EV_WRITE);
_bufferevent_run_eventcb(bufev, what);

done:
_bufferevent_decref_and_unlock(bufev);
}

bufferevent_writecb需要处理两件事，一个是连接逻辑，一个是发送数据逻辑，注意当处理连接逻辑时，如果当前没有激活EV_WRITE或者处于write_suspended状态，需要删除写监听，因为当前的写监听事件只用于处理连接。另外当处理数据逻辑时如果output的数据长度为0，同样需要删除写事件，因为output的callback在有数据之后会重新添加写事件：

static void
bufferevent_socket_outbuf_cb(struct evbuffer *buf,
const struct evbuffer_cb_info *cbinfo,
void *arg)
{
struct bufferevent *bufev = arg;
struct bufferevent_private *bufev_p =
EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);

if (cbinfo->n_added &&
(bufev->enabled & EV_WRITE) &&
!event_pending(&bufev->ev_write, EV_WRITE, NULL) &&
!bufev_p->write_suspended) {
/* Somebody added data to the buffer, and we would like to
* write, and we were not writing.  So, start writing. */
if (be_socket_add(&bufev->ev_write, &bufev->timeout_write) == -1) {
/* Should we log this? */
}
}
}

bufferevent_writecb和bufferevent_readcb还需要调用用户自定义的读写事件：

void
_bufferevent_run_writecb(struct bufferevent *bufev)
{
/* Requires that we hold the lock and a reference */
struct bufferevent_private *p =
EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
if (bufev->writecb == NULL)
return;
if (p->options & BEV_OPT_DEFER_CALLBACKS) {
p->writecb_pending = 1;
if (!p->deferred.queued)
SCHEDULE_DEFERRED(p);
} else {
bufev->writecb(bufev, bufev->cbarg);
}
}
#define SCHEDULE_DEFERRED(bevp)                     \
do {                                \
bufferevent_incref(&(bevp)->bev);           \
event_deferred_cb_schedule(             \
event_base_get_deferred_cb_queue((bevp)->bev.ev_base), \
&(bevp)->deferred);             \
} while (0)

如果bufferevent的选项包含BEV_OPT_DEFER_CALLBACKS则需要使用延迟调用方式，延迟调用是在event_base的loop循环统一处理的。p->deferred.queued如果没有被设置过，则加入到ev_base的延迟调用队列。bufferevent_init_common中设置过defered对应的回掉：

if (options & BEV_OPT_DEFER_CALLBACKS) {
if (options & BEV_OPT_UNLOCK_CALLBACKS)
event_deferred_cb_init(&bufev_private->deferred,
bufferevent_run_deferred_callbacks_unlocked,
bufev_private);
else
event_deferred_cb_init(&bufev_private->deferred,
bufferevent_run_deferred_callbacks_locked,
bufev_private);
}

这两个回掉只是使用锁的方式不同：

static void
bufferevent_run_deferred_callbacks_locked(struct deferred_cb *_, void *arg)
{
struct bufferevent_private *bufev_private = arg;
struct bufferevent *bufev = &bufev_private->bev;

BEV_LOCK(bufev);
if ((bufev_private->eventcb_pending & BEV_EVENT_CONNECTED) &&
bufev->errorcb) {
/* The "connected" happened before any reads or writes, so
send it first. */
bufev_private->eventcb_pending &= ~BEV_EVENT_CONNECTED;
bufev->errorcb(bufev, BEV_EVENT_CONNECTED, bufev->cbarg);
}
if (bufev_private->readcb_pending && bufev->readcb) {
bufev_private->readcb_pending = 0;
bufev->readcb(bufev, bufev->cbarg);
}
if (bufev_private->writecb_pending && bufev->writecb) {
bufev_private->writecb_pending = 0;
bufev->writecb(bufev, bufev->cbarg);
}
if (bufev_private->eventcb_pending && bufev->errorcb) {
short what = bufev_private->eventcb_pending;
int err = bufev_private->errno_pending;
bufev_private->eventcb_pending = 0;
bufev_private->errno_pending = 0;
EVUTIL_SET_SOCKET_ERROR(err);
bufev->errorcb(bufev, what, bufev->cbarg);
}
_bufferevent_decref_and_unlock(bufev);
}

static void
bufferevent_run_deferred_callbacks_unlocked(struct deferred_cb *_, void *arg)
{
struct bufferevent_private *bufev_private = arg;
struct bufferevent *bufev = &bufev_private->bev;

BEV_LOCK(bufev);
#define UNLOCKED(stmt) \
do { BEV_UNLOCK(bufev); stmt; BEV_LOCK(bufev); } while(0)

if ((bufev_private->eventcb_pending & BEV_EVENT_CONNECTED) &&
bufev->errorcb) {
/* The "connected" happened before any reads or writes, so
send it first. */
bufferevent_event_cb errorcb = bufev->errorcb;
void *cbarg = bufev->cbarg;
bufev_private->eventcb_pending &= ~BEV_EVENT_CONNECTED;
UNLOCKED(errorcb(bufev, BEV_EVENT_CONNECTED, cbarg));
}
if (bufev_private->readcb_pending && bufev->readcb) {
bufferevent_data_cb readcb = bufev->readcb;
void *cbarg = bufev->cbarg;
bufev_private->readcb_pending = 0;
UNLOCKED(readcb(bufev, cbarg));
}
if (bufev_private->writecb_pending && bufev->writecb) {
bufferevent_data_cb writecb = bufev->writecb;
void *cbarg = bufev->cbarg;
bufev_private->writecb_pending = 0;
UNLOCKED(writecb(bufev, cbarg));
}
if (bufev_private->eventcb_pending && bufev->errorcb) {
bufferevent_event_cb errorcb = bufev->errorcb;
void *cbarg = bufev->cbarg;
short what = bufev_private->eventcb_pending;
int err = bufev_private->errno_pending;
bufev_private->eventcb_pending = 0;
bufev_private->errno_pending = 0;
EVUTIL_SET_SOCKET_ERROR(err);
UNLOCKED(errorcb(bufev,what,cbarg));
}
_bufferevent_decref_and_unlock(bufev);
#undef UNLOCKED
}

以上就是bufferevent_sock的实现方式，下一节我们将分析IOCP的实现方式。