libevent(九)bufferevent
接上文: libevent(八)bufferevent
在用戶的回調函數中,通過bufferevent_read從輸入緩沖input中讀數據,相應地,通過bufferevent_write向輸出緩沖output中寫數據。
關於寫事件,這裏要多說幾句。
對於一個fd,只要它的寫緩沖區沒有滿,就會觸發寫事件。一般情況下,如果不向這個fd發送大量的數據,它的寫緩沖區是不會滿的。
所以如果一開始就監聽寫事件,那麽寫事件會一直被觸發。
libevent的做法是:當我們確實要寫入數據時,才監聽寫事件。
下面是bufferevent_write的實現:
int bufferevent_write(struct bufferevent *bufev, const void *data, size_t size) { if (evbuffer_add(bufev->output, data, size) == -1) return (-1); return 0; }
/* Adds data to an event buffer */ int evbuffer_add(struct evbuffer *buf, const void *data_in, size_t datlen) { struct evbuffer_chain *chain, *tmp;const unsigned char *data = data_in; size_t remain, to_alloc; int result = -1; EVBUFFER_LOCK(buf); if (buf->freeze_end) { goto done; } /* Prevent buf->total_len overflow */ if (datlen > EV_SIZE_MAX - buf->total_len) { goto done; } chain= buf->last; /* If there are no chains allocated for this buffer, allocate one * big enough to hold all the data. */ if (chain == NULL) { chain = evbuffer_chain_new(datlen); if (!chain) goto done; evbuffer_chain_insert(buf, chain); } if ((chain->flags & EVBUFFER_IMMUTABLE) == 0) { /* Always true for mutable buffers */ EVUTIL_ASSERT(chain->misalign >= 0 && (ev_uint64_t)chain->misalign <= EVBUFFER_CHAIN_MAX); remain = chain->buffer_len - (size_t)chain->misalign - chain->off; if (remain >= datlen) { /* there‘s enough space to hold all the data in the * current last chain */ memcpy(chain->buffer + chain->misalign + chain->off, data, datlen); chain->off += datlen; buf->total_len += datlen; buf->n_add_for_cb += datlen; goto out; } else if (!CHAIN_PINNED(chain) && evbuffer_chain_should_realign(chain, datlen)) { /* we can fit the data into the misalignment */ evbuffer_chain_align(chain); memcpy(chain->buffer + chain->off, data, datlen); chain->off += datlen; buf->total_len += datlen; buf->n_add_for_cb += datlen; goto out; } } else { /* we cannot write any data to the last chain */ remain = 0; } /* we need to add another chain */ to_alloc = chain->buffer_len; if (to_alloc <= EVBUFFER_CHAIN_MAX_AUTO_SIZE/2) to_alloc <<= 1; if (datlen > to_alloc) to_alloc = datlen; tmp = evbuffer_chain_new(to_alloc); if (tmp == NULL) goto done; if (remain) { memcpy(chain->buffer + chain->misalign + chain->off, data, remain); chain->off += remain; buf->total_len += remain; buf->n_add_for_cb += remain; } data += remain; datlen -= remain; memcpy(tmp->buffer, data, datlen); tmp->off = datlen; evbuffer_chain_insert(buf, tmp); buf->n_add_for_cb += datlen; out: evbuffer_invoke_callbacks(buf); result = 0; done: EVBUFFER_UNLOCK(buf); return result; }
有一點要註意的是:
evbuffer_add將用戶數據寫入輸出緩沖output之後,還要通過evbuffer_invoke_callbacks調用回調函數。
在上文libevent(八)bufferevent中,我們通過bufferevent_socket_new創建bufferevent的時候,設置了輸出緩沖output的回調函數為bufferevent_socket_outbuf_cb,此時就派上用場了。
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_socket_outbuf_cb中,我們將寫事件添加到了event_base中。
總結一下bufferevent_write:
1. 將用戶數據寫入輸出緩沖區output
2. 添加寫事件到事件循環
由於此時fd的寫緩沖區沒有滿,所以寫事件被觸發,開始執行寫事件的回調函數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); }
三個註意點:
1. 調用evbuffer_write_atmost寫數據
evbuffer_write_atmost通過write系統調用將數據寫入fd寫緩沖區。
2. 如果數據寫完,刪除寫事件
3. 最後會調用用戶定義的回調函數
整個寫事件流程如下:
1. 用戶主動調用
libevent(九)bufferevent