Merge remote-tracking branch 'github/20_active_prio_inv' into patches-2.0

event-internal.h

@@ -196,6 +196,11 @@ struct event_base {
 	int event_gotterm;
 	/** Set if we should terminate the loop immediately */
 	int event_break;
+	/** Set if we should start a new instance of the loop immediately. */
+	int event_continue;
+
+	/** The currently running priority of events */
+	int event_running_priority;
 
 	/** Set if we're running the event_base_loop function, to prevent
 	 * reentrant invocation. */

event.c

@@ -1359,6 +1359,8 @@ event_process_active_single_queue(struct event_base *base,
 
 		if (base->event_break)
 			return -1;
+		if (base->event_continue)
+			break;
 	}
 	return count;
 }
@@ -1409,11 +1411,13 @@ event_process_active(struct event_base *base)
 
 	for (i = 0; i < base->nactivequeues; ++i) {
 		if (TAILQ_FIRST(&base->activequeues[i]) != NULL) {
+			base->event_running_priority = i;
 			activeq = &base->activequeues[i];
 			c = event_process_active_single_queue(base, activeq);
-			if (c < 0)
+			if (c < 0) {
+				base->event_running_priority = -1;
 				return -1;
-			else if (c > 0)
+			} else if (c > 0)
 				break; /* Processed a real event; do not
 					* consider lower-priority events */
 			/* If we get here, all of the events we processed
@@ -1422,6 +1426,7 @@ event_process_active(struct event_base *base)
 	}
 
 	event_process_deferred_callbacks(&base->defer_queue,&base->event_break);
+	base->event_running_priority = -1;
 	return c;
 }
 
@@ -1559,6 +1564,8 @@ event_base_loop(struct event_base *base, int flags)
 	base->event_gotterm = base->event_break = 0;
 
 	while (!done) {
+		base->event_continue = 0;
+
 		/* Terminate the loop if we have been asked to */
 		if (base->event_gotterm) {
 			break;
@@ -2287,6 +2294,9 @@ event_active_nolock(struct event *ev, int res, short ncalls)
 
 	ev->ev_res = res;
 
+	if (ev->ev_pri < base->event_running_priority)
+		base->event_continue = 1;
+
 	if (ev->ev_events & EV_SIGNAL) {
 #ifndef _EVENT_DISABLE_THREAD_SUPPORT
 		if (base->current_event == ev && !EVBASE_IN_THREAD(base)) {

test/regress.c

@@ -1600,6 +1600,58 @@ test_priorities(void)
 		test_priorities_impl(3);
 }
 
+/* priority-active-inversion: activate a higher-priority event, and make sure
+ * it keeps us from running a lower-priority event first. */
+static int n_pai_calls = 0;
+static struct event pai_events[3];
+
+static void
+prio_active_inversion_cb(evutil_socket_t fd, short what, void *arg)
+{
+	int *call_order = arg;
+	*call_order = n_pai_calls++;
+	if (n_pai_calls == 1) {
+		/* This should activate later, even though it shares a
+		   priority with us. */
+		event_active(&pai_events[1], EV_READ, 1);
+		/* This should activate next, since its priority is higher,
+		   even though we activated it second. */
+		event_active(&pai_events[2], EV_TIMEOUT, 1);
+	}
+}
+
+static void
+test_priority_active_inversion(void *data_)
+{
+	struct basic_test_data *data = data_;
+	struct event_base *base = data->base;
+	int call_order[3];
+	int i;
+	tt_int_op(event_base_priority_init(base, 8), ==, 0);
+
+	n_pai_calls = 0;
+	memset(call_order, 0, sizeof(call_order));
+
+	for (i=0;i<3;++i) {
+		event_assign(&pai_events[i], data->base, -1, 0,
+		    prio_active_inversion_cb, &call_order[i]);
+	}
+
+	event_priority_set(&pai_events[0], 4);
+	event_priority_set(&pai_events[1], 4);
+	event_priority_set(&pai_events[2], 0);
+
+	event_active(&pai_events[0], EV_WRITE, 1);
+
+	event_base_dispatch(base);
+	tt_int_op(n_pai_calls, ==, 3);
+	tt_int_op(call_order[0], ==, 0);
+	tt_int_op(call_order[1], ==, 2);
+	tt_int_op(call_order[2], ==, 1);
+end:
+	;
+}
+
 
 static void
 test_multiple_cb(evutil_socket_t fd, short event, void *arg)
@@ -2364,6 +2416,7 @@ struct testcase_t main_testcases[] = {
 	{ "persistent_active_timeout", test_persistent_active_timeout,
 	  TT_FORK|TT_NEED_BASE, &basic_setup, NULL },
 	LEGACY(priorities, TT_FORK|TT_NEED_BASE),
+	BASIC(priority_active_inversion, TT_FORK|TT_NEED_BASE),
 	{ "common_timeout", test_common_timeout, TT_FORK|TT_NEED_BASE,
 	  &basic_setup, NULL },