/** Internal: If callbacks are deferred and we have a read callback, schedule
* a readcb. Otherwise just run the readcb. */
-void bufferevent_run_readcb_(struct bufferevent *bufev);
+void bufferevent_run_readcb_(struct bufferevent *bufev, int options);
/** Internal: If callbacks are deferred and we have a write callback, schedule
* a writecb. Otherwise just run the writecb. */
-void bufferevent_run_writecb_(struct bufferevent *bufev);
+void bufferevent_run_writecb_(struct bufferevent *bufev, int options);
/** Internal: If callbacks are deferred and we have an eventcb, schedule
* it to run with events "what". Otherwise just run the eventcb. */
void bufferevent_run_eventcb_(struct bufferevent *bufev, short what);
+/** Internal: Run or schedule (if deferred or options contain
+ * BEV_TRIG_DEFER_CALLBACKS) I/O callbacks specified in iotype.
+ * Must already hold the bufev lock. */
+void bufferevent_trigger_nolock_(struct bufferevent *bufev, short iotype, int options);
+
+
/** Internal: Add the event 'ev' with timeout tv, unless tv is set to 0, in
* which case add ev with no timeout. */
int bufferevent_add_event_(struct event *ev, const struct timeval *tv);
void
-bufferevent_run_readcb_(struct bufferevent *bufev)
+bufferevent_run_readcb_(struct bufferevent *bufev, int options)
{
/* Requires that we hold the lock and a reference */
struct bufferevent_private *p =
EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
if (bufev->readcb == NULL)
return;
- if (p->options & BEV_OPT_DEFER_CALLBACKS) {
+ if ((p->options & BEV_OPT_DEFER_CALLBACKS) ||
+ (options & BEV_TRIG_DEFER_CALLBACKS)) {
p->readcb_pending = 1;
SCHEDULE_DEFERRED(p);
} else {
}
void
-bufferevent_run_writecb_(struct bufferevent *bufev)
+bufferevent_run_writecb_(struct bufferevent *bufev, int options)
{
/* 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) {
+ if ((p->options & BEV_OPT_DEFER_CALLBACKS) ||
+ (options & BEV_TRIG_DEFER_CALLBACKS)) {
p->writecb_pending = 1;
SCHEDULE_DEFERRED(p);
} else {
}
}
+void
+bufferevent_trigger_nolock_(struct bufferevent *bufev, short iotype, int options)
+{
+ if ((iotype & EV_READ) && ((options & BEV_TRIG_IGNORE_WATERMARKS) ||
+ evbuffer_get_length(bufev->input) >= bufev->wm_read.low))
+ bufferevent_run_readcb_(bufev, options);
+ if ((iotype & EV_WRITE) && ((options & BEV_TRIG_IGNORE_WATERMARKS) ||
+ evbuffer_get_length(bufev->output) <= bufev->wm_write.low))
+ bufferevent_run_writecb_(bufev, options);
+}
+
+void
+bufferevent_trigger(struct bufferevent *bufev, short iotype, int options)
+{
+ bufferevent_incref_and_lock_(bufev);
+ bufferevent_trigger_nolock_(bufev, iotype, options);
+ bufferevent_decref_and_unlock_(bufev);
+}
+
void
bufferevent_run_eventcb_(struct bufferevent *bufev, short what)
{
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,
- event_base_get_npriorities(base) / 2,
- bufferevent_run_deferred_callbacks_unlocked,
- bufev_private);
- else
- event_deferred_cb_init_(
- &bufev_private->deferred,
- event_base_get_npriorities(base) / 2,
- bufferevent_run_deferred_callbacks_locked,
- bufev_private);
- }
+ if (options & BEV_OPT_UNLOCK_CALLBACKS)
+ event_deferred_cb_init_(
+ &bufev_private->deferred,
+ event_base_get_npriorities(base) / 2,
+ bufferevent_run_deferred_callbacks_unlocked,
+ bufev_private);
+ else
+ event_deferred_cb_init_(
+ &bufev_private->deferred,
+ event_base_get_npriorities(base) / 2,
+ bufferevent_run_deferred_callbacks_locked,
+ bufev_private);
bufev_private->options = options;
if (bev_a->ok) {
if (ok && nbytes) {
BEV_RESET_GENERIC_READ_TIMEOUT(bev);
- if (evbuffer_get_length(bev->input) >= bev->wm_read.low)
- bufferevent_run_readcb_(bev);
+ bufferevent_trigger_nolock_(bev, EV_READ, 0);
bev_async_consider_reading(bev_a);
} else if (!ok) {
what |= BEV_EVENT_ERROR;
if (bev_a->ok) {
if (ok && nbytes) {
BEV_RESET_GENERIC_WRITE_TIMEOUT(bev);
- if (evbuffer_get_length(bev->output) <=
- bev->wm_write.low)
- bufferevent_run_writecb_(bev);
+ bufferevent_trigger_nolock_(bev, EV_WRITE, 0);
bev_async_consider_writing(bev_a);
} else if (!ok) {
what |= BEV_EVENT_ERROR;
/* Or if we have filled the underlying output buffer. */
!be_underlying_writebuf_full(bevf,state));
- if (processed &&
- evbuffer_get_length(bufev->output) <= bufev->wm_write.low) {
+ if (processed) {
/* call the write callback.*/
- bufferevent_run_writecb_(bufev);
+ bufferevent_trigger_nolock_(bufev, EV_WRITE, 0);
if (res == BEV_OK &&
(bufev->enabled & EV_WRITE) &&
/* XXX This should be in process_input, not here. There are
* other places that can call process-input, and they should
* force readcb calls as needed. */
- if (processed_any &&
- evbuffer_get_length(bufev->input) >= bufev->wm_read.low)
- bufferevent_run_readcb_(bufev);
+ if (processed_any)
+ bufferevent_trigger_nolock_(bufev, EV_READ, 0);
bufferevent_decref_and_unlock_(bufev);
}
if (bev_ssl->underlying)
BEV_RESET_GENERIC_WRITE_TIMEOUT(bev);
- if (evbuffer_get_length(output) <= bev->wm_write.low)
- bufferevent_run_writecb_(bev);
+ bufferevent_trigger_nolock_(bev, EV_WRITE, 0);
}
return result;
}
if (all_result_flags & OP_MADE_PROGRESS) {
struct bufferevent *bev = &bev_ssl->bev.bev;
- struct evbuffer *input = bev->input;
- if (evbuffer_get_length(input) >= bev->wm_read.low) {
- bufferevent_run_readcb_(bev);
- }
+ bufferevent_trigger_nolock_(bev, EV_READ, 0);
}
if (!bev_ssl->underlying) {
r = do_read(bev_ssl, 1024); /* XXXX 1024 is a hack */
if (r & OP_MADE_PROGRESS) {
struct bufferevent *bev = &bev_ssl->bev.bev;
- struct evbuffer *input = bev->input;
- if (evbuffer_get_length(input) >= bev->wm_read.low) {
- bufferevent_run_readcb_(bev);
- }
+ bufferevent_trigger_nolock_(bev, EV_READ, 0);
}
if (r & (OP_ERR|OP_BLOCKED))
break;
be_pair_transfer(struct bufferevent *src, struct bufferevent *dst,
int ignore_wm)
{
- size_t src_size, dst_size;
+ size_t dst_size;
size_t n;
evbuffer_unfreeze(src->output, 1);
BEV_DEL_GENERIC_WRITE_TIMEOUT(dst);
}
- src_size = evbuffer_get_length(src->output);
- dst_size = evbuffer_get_length(dst->input);
-
- if (dst_size >= dst->wm_read.low) {
- bufferevent_run_readcb_(dst);
- }
- if (src_size <= src->wm_write.low) {
- bufferevent_run_writecb_(src);
- }
+ bufferevent_trigger_nolock_(dst, EV_READ, 0);
+ bufferevent_trigger_nolock_(src, EV_WRITE, 0);
done:
evbuffer_freeze(src->output, 1);
evbuffer_freeze(dst->input, 0);
bufferevent_decrement_read_buckets_(bufev_p, res);
/* Invoke the user callback - must always be called last */
- if (evbuffer_get_length(input) >= bufev->wm_read.low)
- bufferevent_run_readcb_(bufev);
+ bufferevent_trigger_nolock_(bufev, EV_READ, 0);
goto done;
* 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);
+ if (res || !connected) {
+ bufferevent_trigger_nolock_(bufev, EV_WRITE, 0);
}
goto done;
short iotype,
enum bufferevent_flush_mode mode);
+/**
+ Flags for bufferevent_trigger(_event) that modify when and how to trigger
+ the callback.
+*/
+enum bufferevent_trigger_options {
+ /** trigger the callback regardless of the watermarks */
+ BEV_TRIG_IGNORE_WATERMARKS = (1<<0),
+
+ /** defer even if the callbacks are not */
+ BEV_TRIG_DEFER_CALLBACKS = (1<<1),
+};
+
+/**
+ Triggers bufferevent data callbacks.
+
+ The function will honor watermarks unless options contain
+ BEV_TRIG_IGNORE_WATERMARKS. If the options contain BEV_OPT_DEFER_CALLBACKS,
+ the callbacks are deferred.
+
+ @param bufev the bufferevent object
+ @param iotype either EV_READ or EV_WRITE or both.
+ @param options
+ */
+void bufferevent_trigger(struct bufferevent *bufev, short iotype,
+ int options);
+
/**
@name Filtering support
}
static int bufferevent_connect_test_flags = 0;
+static int bufferevent_trigger_test_flags = 0;
static int n_strings_read = 0;
static int n_reads_invoked = 0;
bufferevent_free(bev2);
}
+static void
+trigger_failure_cb(evutil_socket_t fd, short what, void *ctx)
+{
+ TT_FAIL(("The triggered callback did not fire or the machine is really slow (try increasing timeout)."));
+}
+
+static void
+trigger_readcb_triggered(struct bufferevent *bev, void *ctx)
+{
+ struct event_base *base = ctx;
+
+ TT_BLATHER(("Read successfully triggered."));
+ n_reads_invoked++;
+ event_base_loopexit(base, NULL);
+}
+
+static void
+trigger_readcb(struct bufferevent *bev, void *ctx)
+{
+ struct timeval timeout = { 30, 0 };
+ struct event_base *base = ctx;
+ size_t low, high, len;
+ int expected_reads;
+
+ TT_BLATHER(("Read invoked on %d.", (int)bufferevent_getfd(bev)));
+ expected_reads = ++n_reads_invoked;
+
+ bufferevent_setcb(bev, trigger_readcb_triggered, NULL, reader_eventcb, ctx);
+
+ bufferevent_getwatermark(bev, EV_READ, &low, &high);
+ len = evbuffer_get_length(bufferevent_get_input(bev));
+
+ bufferevent_setwatermark(bev, EV_READ, len + 1, 0);
+ bufferevent_trigger(bev, EV_READ, bufferevent_trigger_test_flags);
+ /* no callback expected */
+ tt_int_op(n_reads_invoked, ==, expected_reads);
+
+ if ((bufferevent_trigger_test_flags & BEV_TRIG_DEFER_CALLBACKS) ||
+ (bufferevent_connect_test_flags & BEV_OPT_DEFER_CALLBACKS)) {
+ /* will be deferred */
+ } else {
+ expected_reads++;
+ }
+
+ event_base_once(base, -1, EV_TIMEOUT, trigger_failure_cb, NULL, &timeout);
+
+ bufferevent_trigger(bev, EV_READ,
+ bufferevent_trigger_test_flags | BEV_TRIG_IGNORE_WATERMARKS);
+ tt_int_op(n_reads_invoked, ==, expected_reads);
+
+ bufferevent_setwatermark(bev, EV_READ, low, high);
+end:
+ ;
+}
+
+static void
+test_bufferevent_trigger(void *arg)
+{
+ struct basic_test_data *data = arg;
+ struct evconnlistener *lev=NULL;
+ struct bufferevent *bev=NULL;
+ struct sockaddr_in localhost;
+ struct sockaddr_storage ss;
+ struct sockaddr *sa;
+ ev_socklen_t slen;
+
+ int be_flags=BEV_OPT_CLOSE_ON_FREE;
+ int trig_flags=0;
+
+ if (strstr((char*)data->setup_data, "defer")) {
+ be_flags |= BEV_OPT_DEFER_CALLBACKS;
+ }
+ bufferevent_connect_test_flags = be_flags;
+
+ if (strstr((char*)data->setup_data, "postpone")) {
+ trig_flags |= BEV_TRIG_DEFER_CALLBACKS;
+ }
+ bufferevent_trigger_test_flags = trig_flags;
+
+ memset(&localhost, 0, sizeof(localhost));
+
+ localhost.sin_port = 0; /* pick-a-port */
+ localhost.sin_addr.s_addr = htonl(0x7f000001L);
+ localhost.sin_family = AF_INET;
+ sa = (struct sockaddr *)&localhost;
+ lev = evconnlistener_new_bind(data->base, listen_cb, data->base,
+ LEV_OPT_CLOSE_ON_FREE|LEV_OPT_REUSEABLE,
+ 16, sa, sizeof(localhost));
+ tt_assert(lev);
+
+ sa = (struct sockaddr *)&ss;
+ slen = sizeof(ss);
+ if (regress_get_listener_addr(lev, sa, &slen) < 0) {
+ tt_abort_perror("getsockname");
+ }
+
+ tt_assert(!evconnlistener_enable(lev));
+ bev = bufferevent_socket_new(data->base, -1, be_flags);
+ tt_assert(bev);
+ bufferevent_setcb(bev, trigger_readcb, NULL, reader_eventcb, data->base);
+
+ bufferevent_enable(bev, EV_READ);
+
+ tt_want(!bufferevent_socket_connect(bev, sa, sizeof(localhost)));
+
+ event_base_dispatch(data->base);
+
+ tt_int_op(n_reads_invoked, ==, 2);
+end:
+ if (lev)
+ evconnlistener_free(lev);
+
+ if (bev)
+ bufferevent_free(bev);
+}
+
struct testcase_t bufferevent_testcases[] = {
LEGACY(bufferevent, TT_ISOLATED),
TT_FORK|TT_NEED_BASE, &basic_setup, (void*)"filter" },
{ "bufferevent_timeout_filter_pair", test_bufferevent_timeouts,
TT_FORK|TT_NEED_BASE, &basic_setup, (void*)"filter pair" },
+ { "bufferevent_trigger", test_bufferevent_trigger, TT_FORK|TT_NEED_BASE,
+ &basic_setup, (void*)"" },
+ { "bufferevent_trigger_defer", test_bufferevent_trigger,
+ TT_FORK|TT_NEED_BASE, &basic_setup, (void*)"defer" },
+ { "bufferevent_trigger_postpone", test_bufferevent_trigger,
+ TT_FORK|TT_NEED_BASE|TT_NEED_THREADS, &basic_setup,
+ (void*)"postpone" },
+ { "bufferevent_trigger_defer_postpone", test_bufferevent_trigger,
+ TT_FORK|TT_NEED_BASE|TT_NEED_THREADS, &basic_setup,
+ (void*)"defer postpone" },
#ifdef EVENT__HAVE_LIBZ
LEGACY(bufferevent_zlib, TT_ISOLATED),
#else
projects / libevent / commitdiff