Initial support for a lightweight 'deferred callbacks'.

A 'deferred callback' is just a function that we've queued in the
event base.  This ability is needed for some mt stuff, and for complex
callback chains.  For internal use only.

svn:r1150

defer-internal.h

@@ -0,0 +1,56 @@
+/*
+ * Copyright (c) 2009 Niels Provos and Nick Mathewson
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in the
+ *    documentation and/or other materials provided with the distribution.
+ * 3. The name of the author may not be used to endorse or promote products
+ *    derived from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
+ * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
+ * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
+ * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
+ * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
+ * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
+ * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+#ifndef _DEFER_INTERNAL_H_
+#define _DEFER_INTERNAL_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "event-config.h"
+#include <sys/queue.h>
+
+struct deferred_cb;
+
+typedef void (*deferred_cb_fn)(struct deferred_cb *, void *);
+
+struct deferred_cb {
+	TAILQ_ENTRY (deferred_cb) (cb_next);
+	unsigned queued : 1;
+	deferred_cb_fn cb;
+	void *arg;
+};
+
+void event_deferred_cb_init(struct deferred_cb *, deferred_cb_fn, void *);
+void event_deferred_cb_cancel(struct event_base *, struct deferred_cb *);
+void event_deferred_cb_schedule(struct event_base *, struct deferred_cb *);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* _EVENT_INTERNAL_H_ */
+

event-internal.h

@@ -101,12 +101,16 @@ struct event_base {
 	struct event_list **activequeues;
 	int nactivequeues;
 
+	/* deferred callback management */
+	TAILQ_HEAD (deferred_cb_list, deferred_cb) deferred_cb_list;
+
 	/* for mapping io activity to events */
 	struct event_io_map io;
 
 	/* for mapping signal activity to events */
 	struct event_signal_map sigmap;
 
+
 	struct event_list eventqueue;
 	struct timeval event_tv;
 

event.c

@@ -65,6 +65,7 @@
 #include "event2/event_struct.h"
 #include "event2/event_compat.h"
 #include "event-internal.h"
+#include "defer-internal.h"
 #include "evthread-internal.h"
 #include "event2/thread.h"
 #include "event2/util.h"
@@ -250,6 +251,7 @@ event_base_new_with_config(struct event_config *cfg)
 
 	min_heap_ctor(&base->timeheap);
 	TAILQ_INIT(&base->eventqueue);
+	TAILQ_INIT(&base->deferred_cb_list);
 	base->sig.ev_signal_pair[0] = -1;
 	base->sig.ev_signal_pair[1] = -1;
 
@@ -649,6 +651,28 @@ event_process_active_single_queue(struct event_base *base,
 	return count;
 }
 
+static int
+event_process_deferred_callbacks(struct event_base *base)
+{
+	int count = 0;
+	struct deferred_cb *cb;
+
+	while ((cb = TAILQ_FIRST(&base->deferred_cb_list))) {
+		cb->queued = 0;
+		TAILQ_REMOVE(&base->deferred_cb_list, cb, cb_next);
+		--base->event_count_active;
+		EVBASE_RELEASE_LOCK(base, EVTHREAD_WRITE, th_base_lock);
+
+		cb->cb(cb, cb->arg);
+		++count;
+		if (event_gotsig || base->event_break)
+			return -1;
+
+		EVBASE_ACQUIRE_LOCK(base, EVTHREAD_WRITE, th_base_lock);
+	}
+	return count;
+}
+
 /*
  * Active events are stored in priority queues.  Lower priorities are always
  * process before higher priorities.  Low priority events can starve high
@@ -677,6 +701,8 @@ event_process_active(struct event_base *base)
 		}
 	}
 
+	event_process_deferred_callbacks(base);
+
 	EVBASE_RELEASE_LOCK(base, EVTHREAD_WRITE, th_base_lock);
 }
 
@@ -1307,6 +1333,44 @@ event_active_internal(struct event *ev, int res, short ncalls)
 	event_queue_insert(base, ev, EVLIST_ACTIVE);
 }
 
+void
+event_deferred_cb_init(struct deferred_cb *cb, deferred_cb_fn fn, void *arg)
+{
+	memset(cb, 0, sizeof(struct deferred_cb));
+	cb->cb = fn;
+	cb->arg = arg;
+}
+
+void
+event_deferred_cb_cancel(struct event_base *base, struct deferred_cb *cb)
+{
+	if (!base)
+		base = current_base;
+
+	EVBASE_ACQUIRE_LOCK(base, EVTHREAD_WRITE, th_base_lock);
+	if (cb->queued) {
+		TAILQ_REMOVE(&base->deferred_cb_list, cb, cb_next);
+		--base->event_count_active;
+		cb->queued = 0;
+	}
+	EVBASE_RELEASE_LOCK(base, EVTHREAD_WRITE, th_base_lock);
+}
+
+void
+event_deferred_cb_schedule(struct event_base *base, struct deferred_cb *cb)
+{
+	assert(!cb->queued);
+	if (!base)
+		base = current_base;
+	EVBASE_ACQUIRE_LOCK(base, EVTHREAD_WRITE, th_base_lock);
+	cb->queued = 1;
+	TAILQ_INSERT_TAIL(&base->deferred_cb_list, cb, cb_next);
+	++base->event_count_active;
+	if (!EVBASE_IN_THREAD(base))
+		evthread_notify_base(base);
+	EVBASE_RELEASE_LOCK(base, EVTHREAD_WRITE, th_base_lock);
+}
+
 static int
 timeout_next(struct event_base *base, struct timeval **tv_p)
 {