cuberite/libevent
1
0
Fork 0
mirror of https://github.com/cuberite/libevent.git synced 2025-09-10 04:50:37 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add an option to trigger bufferevent event callbacks

Browse Source
This commit is contained in:
Ondřej Kuzník 2013-12-03 23:01:54 +00:00
parent 61ee18b8b1
commit a7384c7824
9 changed files with 64 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
bufferevent-internal.h
View File

@ -346,8 +346,9 @@ void bufferevent_run_readcb_(struct bufferevent *bufev, int options);
* a writecb. Otherwise just run the writecb. */ * a writecb. Otherwise just run the writecb. */
void bufferevent_run_writecb_(struct bufferevent *bufev, int options); void bufferevent_run_writecb_(struct bufferevent *bufev, int options);
/** Internal: If callbacks are deferred and we have an eventcb, schedule /** Internal: If callbacks are deferred and we have an eventcb, schedule
* it to run with events "what". Otherwise just run the eventcb. */ * it to run with events "what". Otherwise just run the eventcb.
void bufferevent_run_eventcb_(struct bufferevent *bufev, short what); * See bufferevent_trigger_event for meaning of "options". */
void bufferevent_run_eventcb_(struct bufferevent *bufev, short what, int options);
/** Internal: Run or schedule (if deferred or options contain /** Internal: Run or schedule (if deferred or options contain
* BEV_TRIG_DEFER_CALLBACKS) I/O callbacks specified in iotype. * BEV_TRIG_DEFER_CALLBACKS) I/O callbacks specified in iotype.

17
bufferevent.c
View File

@ -272,14 +272,15 @@ bufferevent_trigger(struct bufferevent *bufev, short iotype, int options)
} }
void void
bufferevent_run_eventcb_(struct bufferevent *bufev, short what) bufferevent_run_eventcb_(struct bufferevent *bufev, short what, int options)
{ {
/* Requires that we hold the lock and a reference */ /* Requires that we hold the lock and a reference */
struct bufferevent_private *p = struct bufferevent_private *p =
EVUTIL_UPCAST(bufev, struct bufferevent_private, bev); EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
if (bufev->errorcb == NULL) if (bufev->errorcb == NULL)
return; return;
if (p->options & BEV_OPT_DEFER_CALLBACKS) { if ((p->options & BEV_OPT_DEFER_CALLBACKS) ||
(options & BEV_TRIG_DEFER_CALLBACKS)) {
p->eventcb_pending |= what; p->eventcb_pending |= what;
p->errno_pending = EVUTIL_SOCKET_ERROR(); p->errno_pending = EVUTIL_SOCKET_ERROR();
SCHEDULE_DEFERRED(p); SCHEDULE_DEFERRED(p);
@ -288,6 +289,14 @@ bufferevent_run_eventcb_(struct bufferevent *bufev, short what)
} }
} }
void
bufferevent_trigger_event(struct bufferevent *bufev, short what, int options)
{
bufferevent_incref_and_lock_(bufev);
bufferevent_run_eventcb_(bufev, what, options);
bufferevent_decref_and_unlock_(bufev);
}
int int
bufferevent_init_common_(struct bufferevent_private *bufev_private, bufferevent_init_common_(struct bufferevent_private *bufev_private,
struct event_base *base, struct event_base *base,
@ -914,7 +923,7 @@ bufferevent_generic_read_timeout_cb(evutil_socket_t fd, short event, void *ctx)
struct bufferevent *bev = ctx; struct bufferevent *bev = ctx;
bufferevent_incref_and_lock_(bev); bufferevent_incref_and_lock_(bev);
bufferevent_disable(bev, EV_READ); bufferevent_disable(bev, EV_READ);
bufferevent_run_eventcb_(bev, BEV_EVENT_TIMEOUT|BEV_EVENT_READING); bufferevent_run_eventcb_(bev, BEV_EVENT_TIMEOUT|BEV_EVENT_READING, 0);
bufferevent_decref_and_unlock_(bev); bufferevent_decref_and_unlock_(bev);
} }
static void static void
@ -923,7 +932,7 @@ bufferevent_generic_write_timeout_cb(evutil_socket_t fd, short event, void *ctx)
struct bufferevent *bev = ctx; struct bufferevent *bev = ctx;
bufferevent_incref_and_lock_(bev); bufferevent_incref_and_lock_(bev);
bufferevent_disable(bev, EV_WRITE); bufferevent_disable(bev, EV_WRITE);
bufferevent_run_eventcb_(bev, BEV_EVENT_TIMEOUT|BEV_EVENT_WRITING); bufferevent_run_eventcb_(bev, BEV_EVENT_TIMEOUT|BEV_EVENT_WRITING, 0);
bufferevent_decref_and_unlock_(bev); bufferevent_decref_and_unlock_(bev);
} }

14
bufferevent_async.c
View File

@ -217,7 +217,7 @@ bev_async_consider_writing(struct bufferevent_async *beva)
&beva->write_overlapped)) { &beva->write_overlapped)) {
bufferevent_decref_(bev); bufferevent_decref_(bev);
beva->ok = 0; beva->ok = 0;
bufferevent_run_eventcb_(bev, BEV_EVENT_ERROR); bufferevent_run_eventcb_(bev, BEV_EVENT_ERROR, 0);
} else { } else {
beva->write_in_progress = at_most; beva->write_in_progress = at_most;
bufferevent_decrement_write_buckets_(&beva->bev, at_most); bufferevent_decrement_write_buckets_(&beva->bev, at_most);
@ -270,7 +270,7 @@ bev_async_consider_reading(struct bufferevent_async *beva)
bufferevent_incref_(bev); bufferevent_incref_(bev);
if (evbuffer_launch_read_(bev->input, at_most, &beva->read_overlapped)) { if (evbuffer_launch_read_(bev->input, at_most, &beva->read_overlapped)) {
beva->ok = 0; beva->ok = 0;
bufferevent_run_eventcb_(bev, BEV_EVENT_ERROR); bufferevent_run_eventcb_(bev, BEV_EVENT_ERROR, 0);
bufferevent_decref_(bev); bufferevent_decref_(bev);
} else { } else {
beva->read_in_progress = at_most; beva->read_in_progress = at_most;
@ -428,7 +428,7 @@ connect_complete(struct event_overlapped *eo, ev_uintptr_t key,
bev_async_set_wsa_error(bev, eo); bev_async_set_wsa_error(bev, eo);
bufferevent_run_eventcb_(bev, bufferevent_run_eventcb_(bev,
ok? BEV_EVENT_CONNECTED : BEV_EVENT_ERROR); ok? BEV_EVENT_CONNECTED : BEV_EVENT_ERROR, 0);
event_base_del_virtual_(bev->ev_base); event_base_del_virtual_(bev->ev_base);
@ -463,11 +463,11 @@ read_complete(struct event_overlapped *eo, ev_uintptr_t key,
} else if (!ok) { } else if (!ok) {
what |= BEV_EVENT_ERROR; what |= BEV_EVENT_ERROR;
bev_a->ok = 0; bev_a->ok = 0;
bufferevent_run_eventcb_(bev, what); bufferevent_run_eventcb_(bev, what, 0);
} else if (!nbytes) { } else if (!nbytes) {
what |= BEV_EVENT_EOF; what |= BEV_EVENT_EOF;
bev_a->ok = 0; bev_a->ok = 0;
bufferevent_run_eventcb_(bev, what); bufferevent_run_eventcb_(bev, what, 0);
} }
} }
@ -506,11 +506,11 @@ write_complete(struct event_overlapped *eo, ev_uintptr_t key,
} else if (!ok) { } else if (!ok) {
what |= BEV_EVENT_ERROR; what |= BEV_EVENT_ERROR;
bev_a->ok = 0; bev_a->ok = 0;
bufferevent_run_eventcb_(bev, what); bufferevent_run_eventcb_(bev, what, 0);
} else if (!nbytes) { } else if (!nbytes) {
what |= BEV_EVENT_EOF; what |= BEV_EVENT_EOF;
bev_a->ok = 0; bev_a->ok = 0;
bufferevent_run_eventcb_(bev, what); bufferevent_run_eventcb_(bev, what, 0);
} }
} }

2
bufferevent_filter.c
View File

@ -470,7 +470,7 @@ be_filter_eventcb(struct bufferevent *underlying, short what, void *me_)
bufferevent_incref_and_lock_(bev); bufferevent_incref_and_lock_(bev);
/* All we can really to is tell our own eventcb. */ /* All we can really to is tell our own eventcb. */
bufferevent_run_eventcb_(bev, what); bufferevent_run_eventcb_(bev, what, 0);
bufferevent_decref_and_unlock_(bev); bufferevent_decref_and_unlock_(bev);
} }

12
bufferevent_openssl.c
View File

@ -533,7 +533,7 @@ conn_closed(struct bufferevent_openssl *bev_ssl, int when, int errcode, int ret)
/* when is BEV_EVENT_{READING|WRITING} */ /* when is BEV_EVENT_{READING|WRITING} */
event = when | event; event = when | event;
bufferevent_run_eventcb_(&bev_ssl->bev.bev, event); bufferevent_run_eventcb_(&bev_ssl->bev.bev, event, 0);
} }
static void static void
@ -921,7 +921,7 @@ be_openssl_eventcb(struct bufferevent *bev_base, short what, void *ctx)
eat it. */ eat it. */
} }
if (event) if (event)
bufferevent_run_eventcb_(&bev_ssl->bev.bev, event); bufferevent_run_eventcb_(&bev_ssl->bev.bev, event, 0);
} }
static void static void
@ -931,7 +931,7 @@ be_openssl_readeventcb(evutil_socket_t fd, short what, void *ptr)
bufferevent_incref_and_lock_(&bev_ssl->bev.bev); bufferevent_incref_and_lock_(&bev_ssl->bev.bev);
if (what == EV_TIMEOUT) { if (what == EV_TIMEOUT) {
bufferevent_run_eventcb_(&bev_ssl->bev.bev, bufferevent_run_eventcb_(&bev_ssl->bev.bev,
BEV_EVENT_TIMEOUT|BEV_EVENT_READING); BEV_EVENT_TIMEOUT|BEV_EVENT_READING, 0);
} else { } else {
consider_reading(bev_ssl); consider_reading(bev_ssl);
} }
@ -945,7 +945,7 @@ be_openssl_writeeventcb(evutil_socket_t fd, short what, void *ptr)
bufferevent_incref_and_lock_(&bev_ssl->bev.bev); bufferevent_incref_and_lock_(&bev_ssl->bev.bev);
if (what == EV_TIMEOUT) { if (what == EV_TIMEOUT) {
bufferevent_run_eventcb_(&bev_ssl->bev.bev, bufferevent_run_eventcb_(&bev_ssl->bev.bev,
BEV_EVENT_TIMEOUT|BEV_EVENT_WRITING); BEV_EVENT_TIMEOUT|BEV_EVENT_WRITING, 0);
} else { } else {
consider_writing(bev_ssl); consider_writing(bev_ssl);
} }
@ -1012,7 +1012,7 @@ do_handshake(struct bufferevent_openssl *bev_ssl)
/* Call do_read and do_write as needed */ /* Call do_read and do_write as needed */
bufferevent_enable(&bev_ssl->bev.bev, bev_ssl->bev.bev.enabled); bufferevent_enable(&bev_ssl->bev.bev, bev_ssl->bev.bev.enabled);
bufferevent_run_eventcb_(&bev_ssl->bev.bev, bufferevent_run_eventcb_(&bev_ssl->bev.bev,
BEV_EVENT_CONNECTED); BEV_EVENT_CONNECTED, 0);
return 1; return 1;
} else { } else {
int err = SSL_get_error(bev_ssl->ssl, r); int err = SSL_get_error(bev_ssl->ssl, r);
@ -1051,7 +1051,7 @@ be_openssl_handshakeeventcb(evutil_socket_t fd, short what, void *ptr)
bufferevent_incref_and_lock_(&bev_ssl->bev.bev); bufferevent_incref_and_lock_(&bev_ssl->bev.bev);
if (what & EV_TIMEOUT) { if (what & EV_TIMEOUT) {
bufferevent_run_eventcb_(&bev_ssl->bev.bev, BEV_EVENT_TIMEOUT); bufferevent_run_eventcb_(&bev_ssl->bev.bev, BEV_EVENT_TIMEOUT, 0);
} else } else
do_handshake(bev_ssl);/* XXX handle failure */ do_handshake(bev_ssl);/* XXX handle failure */
bufferevent_decref_and_unlock_(&bev_ssl->bev.bev); bufferevent_decref_and_unlock_(&bev_ssl->bev.bev);

2
bufferevent_pair.c
View File

@ -292,7 +292,7 @@ be_pair_flush(struct bufferevent *bev, short iotype,
be_pair_transfer(bev, partner, 1); be_pair_transfer(bev, partner, 1);
if (mode == BEV_FINISHED) { if (mode == BEV_FINISHED) {
bufferevent_run_eventcb_(partner, iotype|BEV_EVENT_EOF); bufferevent_run_eventcb_(partner, iotype|BEV_EVENT_EOF, 0);
} }
decref_and_unlock(bev); decref_and_unlock(bev);
return 0; return 0;

14
bufferevent_sock.c
View File

@ -193,7 +193,7 @@ bufferevent_readcb(evutil_socket_t fd, short event, void *arg)
error: error:
bufferevent_disable(bufev, EV_READ); bufferevent_disable(bufev, EV_READ);
bufferevent_run_eventcb_(bufev, what); bufferevent_run_eventcb_(bufev, what, 0);
done: done:
bufferevent_decref_and_unlock_(bufev); bufferevent_decref_and_unlock_(bufev);
@ -235,7 +235,7 @@ bufferevent_writecb(evutil_socket_t fd, short event, void *arg)
if (c < 0) { if (c < 0) {
event_del(&bufev->ev_write); event_del(&bufev->ev_write);
event_del(&bufev->ev_read); event_del(&bufev->ev_read);
bufferevent_run_eventcb_(bufev, BEV_EVENT_ERROR); bufferevent_run_eventcb_(bufev, BEV_EVENT_ERROR, 0);
goto done; goto done;
} else { } else {
connected = 1; connected = 1;
@ -244,12 +244,12 @@ bufferevent_writecb(evutil_socket_t fd, short event, void *arg)
event_del(&bufev->ev_write); event_del(&bufev->ev_write);
bufferevent_async_set_connected_(bufev); bufferevent_async_set_connected_(bufev);
bufferevent_run_eventcb_(bufev, bufferevent_run_eventcb_(bufev,
BEV_EVENT_CONNECTED); BEV_EVENT_CONNECTED, 0);
goto done; goto done;
} }
#endif #endif
bufferevent_run_eventcb_(bufev, bufferevent_run_eventcb_(bufev,
BEV_EVENT_CONNECTED); BEV_EVENT_CONNECTED, 0);
if (!(bufev->enabled & EV_WRITE) || if (!(bufev->enabled & EV_WRITE) ||
bufev_p->write_suspended) { bufev_p->write_suspended) {
event_del(&bufev->ev_write); event_del(&bufev->ev_write);
@ -307,7 +307,7 @@ bufferevent_writecb(evutil_socket_t fd, short event, void *arg)
error: error:
bufferevent_disable(bufev, EV_WRITE); bufferevent_disable(bufev, EV_WRITE);
bufferevent_run_eventcb_(bufev, what); bufferevent_run_eventcb_(bufev, what, 0);
done: done:
bufferevent_decref_and_unlock_(bufev); bufferevent_decref_and_unlock_(bufev);
@ -424,7 +424,7 @@ bufferevent_socket_connect(struct bufferevent *bev,
goto done; goto done;
freesock: freesock:
bufferevent_run_eventcb_(bev, BEV_EVENT_ERROR); bufferevent_run_eventcb_(bev, BEV_EVENT_ERROR, 0);
if (ownfd) if (ownfd)
evutil_closesocket(fd); evutil_closesocket(fd);
/* do something about the error? */ /* do something about the error? */
@ -448,7 +448,7 @@ bufferevent_connect_getaddrinfo_cb(int result, struct evutil_addrinfo *ai,
if (result != 0) { if (result != 0) {
bev_p->dns_error = result; bev_p->dns_error = result;
bufferevent_run_eventcb_(bev, BEV_EVENT_ERROR); bufferevent_run_eventcb_(bev, BEV_EVENT_ERROR, 0);
bufferevent_decref_and_unlock_(bev); bufferevent_decref_and_unlock_(bev);
if (ai) if (ai)
evutil_freeaddrinfo(ai); evutil_freeaddrinfo(ai);

12
include/event2/bufferevent.h
View File

@ -584,6 +584,18 @@ enum bufferevent_trigger_options {
void bufferevent_trigger(struct bufferevent *bufev, short iotype, void bufferevent_trigger(struct bufferevent *bufev, short iotype,
int options); int options);
/**
Triggers the bufferevent event callback.
If the options contain BEV_OPT_DEFER_CALLBACKS, the callbacks are deferred.
@param bufev the bufferevent object
@param what the flags to pass onto the event callback
@param options
*/
void bufferevent_trigger_event(struct bufferevent *bufev, short what,
int options);
/** /**
@name Filtering support @name Filtering support

18
test/regress_bufferevent.c
View File

@ -820,13 +820,23 @@ trigger_failure_cb(evutil_socket_t fd, short what, void *ctx)
} }
static void static void
trigger_readcb_triggered(struct bufferevent *bev, void *ctx) trigger_eventcb(struct bufferevent *bev, short what, void *ctx)
{ {
struct event_base *base = ctx; struct event_base *base = ctx;
if (what == ~0) {
TT_BLATHER(("Event successfully triggered."));
event_base_loopexit(base, NULL);
return;
}
reader_eventcb(bev, what, ctx);
}
static void
trigger_readcb_triggered(struct bufferevent *bev, void *ctx)
{
TT_BLATHER(("Read successfully triggered.")); TT_BLATHER(("Read successfully triggered."));
n_reads_invoked++; n_reads_invoked++;
event_base_loopexit(base, NULL); bufferevent_trigger_event(bev, ~0, bufferevent_trigger_test_flags);
} }
static void static void
@ -840,7 +850,7 @@ trigger_readcb(struct bufferevent *bev, void *ctx)
TT_BLATHER(("Read invoked on %d.", (int)bufferevent_getfd(bev))); TT_BLATHER(("Read invoked on %d.", (int)bufferevent_getfd(bev)));
expected_reads = ++n_reads_invoked; expected_reads = ++n_reads_invoked;
bufferevent_setcb(bev, trigger_readcb_triggered, NULL, reader_eventcb, ctx); bufferevent_setcb(bev, trigger_readcb_triggered, NULL, trigger_eventcb, ctx);
bufferevent_getwatermark(bev, EV_READ, &low, &high); bufferevent_getwatermark(bev, EV_READ, &low, &high);
len = evbuffer_get_length(bufferevent_get_input(bev)); len = evbuffer_get_length(bufferevent_get_input(bev));
@ -912,7 +922,7 @@ test_bufferevent_trigger(void *arg)
tt_assert(!evconnlistener_enable(lev)); tt_assert(!evconnlistener_enable(lev));
bev = bufferevent_socket_new(data->base, -1, be_flags); bev = bufferevent_socket_new(data->base, -1, be_flags);
tt_assert(bev); tt_assert(bev);
bufferevent_setcb(bev, trigger_readcb, NULL, reader_eventcb, data->base); bufferevent_setcb(bev, trigger_readcb, NULL, trigger_eventcb, data->base);
bufferevent_enable(bev, EV_READ); bufferevent_enable(bev, EV_READ);