Reorganize code and change behavior of ast_taskprocessor_execute() when taskprocessor is shutting down.

Moved code around to be easier to follow.

ast_taskprocessor_execute() will now return 0 if the taskprocessor is being shut down.



git-svn-id: https://origsvn.digium.com/svn/asterisk/team/mmichelson/threadpool@376499 65c4cc65-6c06-0410-ace0-fbb531ad65f3

1 files changed, 34 insertions, 32 deletions

diff --git a/main/taskprocessor.c b/main/taskprocessor.c
index 78118edb5..69bec7ca0 100644
--- a/main/taskprocessor.c
+++ b/main/taskprocessor.c
@@ -76,6 +76,8 @@ struct ast_taskprocessor {
 	/*! \brief Taskprocessor singleton list entry */
 	AST_LIST_ENTRY(ast_taskprocessor) list;
 	struct ast_taskprocessor_listener *listener;
+	/*! Indicates if the taskprocessor is in the process of shuting down */
+	unsigned int shutting_down:1;
 };
 #define TPS_MAX_BUCKETS 7
 /*! \brief tps_singletons is the astobj2 container for taskprocessor singletons */
@@ -123,6 +125,7 @@ struct default_taskprocessor_listener_pvt {
 	int dead;
 };
 
+
 static void default_tps_wake_up(struct default_taskprocessor_listener_pvt *pvt, int should_die)
 {
 	SCOPED_MUTEX(lock, &pvt->lock); 
@@ -131,20 +134,6 @@ static void default_tps_wake_up(struct default_taskprocessor_listener_pvt *pvt,
 	ast_cond_signal(&pvt->cond);
 }
 
-static void listener_destroy(void *obj)
-{
-	struct ast_taskprocessor_listener *listener = obj;
-
-	listener->callbacks->destroy(listener->private_data);
-}
-
-static void listener_shutdown(struct ast_taskprocessor_listener *listener)
-{
-	listener->callbacks->shutdown(listener);
-	ao2_ref(listener->tps, -1);
-	listener->tps = NULL;
-}
-
 static int default_tps_idle(struct default_taskprocessor_listener_pvt *pvt)
 {
 	SCOPED_MUTEX(lock, &pvt->lock);
@@ -188,6 +177,20 @@ static void *default_listener_alloc(struct ast_taskprocessor_listener *listener)
 	return pvt;
 }
 
+static void default_task_pushed(struct ast_taskprocessor_listener *listener, int was_empty)
+{
+	struct default_taskprocessor_listener_pvt *pvt = listener->private_data;
+
+	if (was_empty) {
+		default_tps_wake_up(pvt, 0);
+	}
+}
+
+static void default_emptied(struct ast_taskprocessor_listener *listener)
+{
+	/* No-op */
+}
+
 static void default_listener_shutdown(struct ast_taskprocessor_listener *listener)
 {
 	struct default_taskprocessor_listener_pvt *pvt = listener->private_data;
@@ -204,20 +207,6 @@ static void default_listener_destroy(void *obj)
 	ast_free(pvt);
 }
 
-static void default_task_pushed(struct ast_taskprocessor_listener *listener, int was_empty)
-{
-	struct default_taskprocessor_listener_pvt *pvt = listener->private_data;
-
-	if (was_empty) {
-		default_tps_wake_up(pvt, 0);
-	}
-}
-
-static void default_emptied(struct ast_taskprocessor_listener *listener)
-{
-	/* No-op */
-}
-
 static const struct ast_taskprocessor_listener_callbacks default_listener_callbacks = {
 	.alloc = default_listener_alloc,
 	.task_pushed = default_task_pushed,
@@ -438,16 +427,15 @@ static void tps_taskprocessor_destroy(void *tps)
 static struct tps_task *tps_taskprocessor_pop(struct ast_taskprocessor *tps)
 {
 	struct tps_task *task;
+	SCOPED_AO2LOCK(lock, tps);
 
-	if (!tps) {
-		ast_log(LOG_ERROR, "missing taskprocessor\n");
+	if (tps->shutting_down) {
 		return NULL;
 	}
-	ao2_lock(tps);
+
 	if ((task = AST_LIST_REMOVE_HEAD(&tps->tps_queue, list))) {
 		tps->tps_queue_size--;
 	}
-	ao2_unlock(tps);
 	return task;
 }
 
@@ -466,6 +454,20 @@ const char *ast_taskprocessor_name(struct ast_taskprocessor *tps)
 	return tps->name;
 }
 
+static void listener_destroy(void *obj)
+{
+	struct ast_taskprocessor_listener *listener = obj;
+
+	listener->callbacks->destroy(listener->private_data);
+}
+
+static void listener_shutdown(struct ast_taskprocessor_listener *listener)
+{
+	listener->callbacks->shutdown(listener);
+	ao2_ref(listener->tps, -1);
+	listener->tps = NULL;
+}
+
 struct ast_taskprocessor_listener *ast_taskprocessor_listener_alloc(const struct ast_taskprocessor_listener_callbacks *callbacks)
 {
 	RAII_VAR(struct ast_taskprocessor_listener *, listener,