Simplify code by using a taskprocessor for dispatching events in the Asterisk core.

git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@115324 65c4cc65-6c06-0410-ace0-fbb531ad65f3

main/event.c

@@ -34,9 +34,9 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "asterisk/lock.h"
 #include "asterisk/utils.h"
 #include "asterisk/unaligned.h"
+#include "asterisk/taskprocessor.h"
 
-/* Only use one thread for now to ensure ordered delivery */
+struct ast_taskprocessor *event_dispatcher;
-#define NUM_EVENT_THREADS 1
 
 /*!
  * \brief An event information element
@@ -84,15 +84,6 @@ struct ast_event_iterator {
 	struct ast_event_ie *ie;
 };
 
-/*! \brief data shared between event dispatching threads */
-static struct {
-	ast_cond_t cond;
-	ast_mutex_t lock;
-	AST_LIST_HEAD_NOLOCK(, ast_event_ref) event_q;
-} event_thread = {
-	.lock = AST_MUTEX_INIT_VALUE,
-};
-
 struct ast_event_ie_val {
 	AST_LIST_ENTRY(ast_event_ie_val) entry;
 	enum ast_event_ie_type ie_type;
@@ -717,52 +708,12 @@ int ast_event_queue_and_cache(struct ast_event *event, ...)
 	return (ast_event_queue(event) || res) ? -1 : 0;
 }
 
-int ast_event_queue(struct ast_event *event)
+static int handle_event(void *data)
 {
-	struct ast_event_ref *event_ref;
+	struct ast_event_ref *event_ref = data;
-	uint16_t host_event_type;
-
-	host_event_type = ntohs(event->type);
-
-	/* Invalid type */
-	if (host_event_type >= AST_EVENT_TOTAL) {
-		ast_log(LOG_WARNING, "Someone tried to queue an event of invalid "
-			"type '%d'!\n", host_event_type);
-		return -1;
-	}
-
-	/* If nobody has subscribed to this event type, throw it away now */
-	if (ast_event_check_subscriber(host_event_type, AST_EVENT_IE_END) 
-		== AST_EVENT_SUB_NONE) {
-		ast_event_destroy(event);
-		return 0;
-	}
-
-	if (!(event_ref = ast_calloc(1, sizeof(*event_ref))))
-		return -1;
-
-	event_ref->event = event;
-
-	ast_mutex_lock(&event_thread.lock);
-	AST_LIST_INSERT_TAIL(&event_thread.event_q, event_ref, entry);
-	ast_cond_signal(&event_thread.cond);
-	ast_mutex_unlock(&event_thread.lock);
-
-	return 0;
-}
-
-static void *ast_event_dispatcher(void *unused)
-{
-	for (;;) {
-		struct ast_event_ref *event_ref;
 	struct ast_event_sub *sub;
 	uint16_t host_event_type;
 
-		ast_mutex_lock(&event_thread.lock);
-		while (!(event_ref = AST_LIST_REMOVE_HEAD(&event_thread.event_q, entry)))
-			ast_cond_wait(&event_thread.cond, &event_thread.lock);
-		ast_mutex_unlock(&event_thread.lock);
-
 	host_event_type = ntohs(event_ref->event->type);
 
 	/* Subscribers to this specific event first */
@@ -797,9 +748,37 @@ static void *ast_event_dispatcher(void *unused)
 	AST_RWDLLIST_UNLOCK(&ast_event_subs[AST_EVENT_ALL]);
 
 	ast_event_ref_destroy(event_ref);
+
+	return 0;
+}
+
+int ast_event_queue(struct ast_event *event)
+{
+	struct ast_event_ref *event_ref;
+	uint16_t host_event_type;
+
+	host_event_type = ntohs(event->type);
+
+	/* Invalid type */
+	if (host_event_type >= AST_EVENT_TOTAL) {
+		ast_log(LOG_WARNING, "Someone tried to queue an event of invalid "
+			"type '%d'!\n", host_event_type);
+		return -1;
 	}
 
-	return NULL;
+	/* If nobody has subscribed to this event type, throw it away now */
+	if (ast_event_check_subscriber(host_event_type, AST_EVENT_IE_END) 
+		== AST_EVENT_SUB_NONE) {
+		ast_event_destroy(event);
+		return 0;
+	}
+
+	if (!(event_ref = ast_calloc(1, sizeof(*event_ref))))
+		return -1;
+
+	event_ref->event = event;
+
+	return ast_taskprocessor_push(event_dispatcher, handle_event, event_ref);
 }
 
 void ast_event_init(void)
@@ -812,10 +791,5 @@ void ast_event_init(void)
 	for (i = 0; i < AST_EVENT_TOTAL; i++)
 		AST_RWLIST_HEAD_INIT(&ast_event_cache[i]);
 
-	ast_cond_init(&event_thread.cond, NULL);
+	event_dispatcher = ast_taskprocessor_get("core_event_dispatcher", 0);
-
-	for (i = 0; i < NUM_EVENT_THREADS; i++) {
-		pthread_t dont_care;
-		ast_pthread_create_background(&dont_care, NULL, ast_event_dispatcher, NULL);
-	}
 }