[asterisk-commits] dlee: branch dlee/performance r399645 - /team/dlee/performance/main/

[SVN commits to the Asterisk project]

Author: dlee
Date: Mon Sep 23 14:34:17 2013
New Revision: 399645

URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=399645
Log:
Reverting taskprocessor changes; something is seriously amiss

Modified:
    team/dlee/performance/main/taskprocessor.c

Modified: team/dlee/performance/main/taskprocessor.c
URL: http://svnview.digium.com/svn/asterisk/team/dlee/performance/main/taskprocessor.c?view=diff&rev=399645&r1=399644&r2=399645
==============================================================================
--- team/dlee/performance/main/taskprocessor.c (original)
+++ team/dlee/performance/main/taskprocessor.c Mon Sep 23 14:34:17 2013
@@ -179,12 +179,6 @@
 			dead = default_tps_idle(pvt);
 		}
 	}
-
-	/* Finish processing the queue */
-	while (ast_taskprocessor_execute(tps)) {
-		/* No-op */
-	}
-
 	return NULL;
 }
 
@@ -366,8 +360,8 @@
 	char name[256];
 	int tcount;
 	unsigned long qsize;
-	unsigned long maxqsize = 0;
-	unsigned long processed = 0;
+	unsigned long maxqsize;
+	unsigned long processed;
 	struct ast_taskprocessor *p;
 	struct ao2_iterator i;
 
@@ -388,8 +382,6 @@
 	ast_cli(a->fd, "\n\t+----- Processor -----+--- Processed ---+- In Queue -+- Max Depth -+");
 	i = ao2_iterator_init(tps_singletons, 0);
 	while ((p = ao2_iterator_next(&i))) {
-		ast_assert(p->stats != NULL);
-
 		ast_copy_string(name, p->name, sizeof(name));
 		qsize = p->tps_queue_size;
 		maxqsize = p->stats->max_qsize;
@@ -425,6 +417,7 @@
 static void tps_taskprocessor_destroy(void *tps)
 {
 	struct ast_taskprocessor *t = tps;
+	struct tps_task *task;
 
 	if (!tps) {
 		ast_log(LOG_ERROR, "missing taskprocessor\n");
@@ -432,9 +425,10 @@
 	}
 	ast_debug(1, "destroying taskprocessor '%s'\n", t->name);
 	/* free it */
-	ast_free(t->stats);
-	t->stats = NULL;
-
+	if (t->stats) {
+		ast_free(t->stats);
+		t->stats = NULL;
+	}
 	ast_free((char *) t->name);
 	if (t->listener) {
 		/* This code should not be reached since the listener
@@ -444,13 +438,16 @@
 		ao2_ref(t->listener, -1);
 		t->listener = NULL;
 	}
-	ast_assert(AST_LIST_EMPTY(&t->tps_queue));
+	while ((task = AST_LIST_REMOVE_HEAD(&t->tps_queue, list))) {
+		tps_task_free(task);
+	}
 }
 
 /* pop the front task and return it */
 static struct tps_task *tps_taskprocessor_pop(struct ast_taskprocessor *tps)
 {
 	struct tps_task *task;
+	SCOPED_AO2LOCK(lock, tps);
 
 	if ((task = AST_LIST_REMOVE_HEAD(&tps->tps_queue, list))) {
 		tps->tps_queue_size--;
@@ -533,7 +530,6 @@
 		ast_log(LOG_WARNING, "failed to create taskprocessor stats for '%s'\n", name);
 		return NULL;
 	}
-
 	if (!(p->name = ast_strdup(name))) {
 		ao2_ref(p, -1);
 		return NULL;
@@ -626,17 +622,13 @@
 		return NULL;
 	}
 
-	ao2_lock(tps);
-	if (ao2_ref(tps, -1) > 3 + tps->tps_queue_size) {
-		ao2_unlock(tps);
-		return NULL;
-	}
-	ao2_unlock(tps);
-	/* If we're down to 3+n references, then those must be:
+	if (ao2_ref(tps, -1) > 3) {
+		return NULL;
+	}
+	/* If we're down to 3 references, then those must be:
 	 * 1. The reference we just got rid of
 	 * 2. The container
 	 * 3. The listener
-	 * 4..n. The tasks currently in flight
 	 */
 	ao2_unlink(tps_singletons, tps);
 	listener_shutdown(tps->listener);
@@ -659,7 +651,6 @@
 		return -1;
 	}
 	ao2_lock(tps);
-	ao2_ref(tps, +1); /* Let's say the queued task has a reference */
 	AST_LIST_INSERT_TAIL(&tps->tps_queue, t, list);
 	previous_size = tps->tps_queue_size++;
 	/* The currently executing task counts as still in queue */
@@ -674,23 +665,16 @@
 	struct tps_task *t;
 	int size;
 
-	ast_assert(tps->stats != NULL);
-
 	ao2_lock(tps);
-	t = tps_taskprocessor_pop(tps);
-
-	/* Empty queue; return false */
-	if (!t) {
-		ao2_unlock(tps);
-
-		/* Since there was no task, no need to decrement the refcount */
-		return 0;
-	}
 	tps->executing = 1;
 	ao2_unlock(tps);
 
-	t->execute(t->datap);
-	tps_task_free(t);
+	t = tps_taskprocessor_pop(tps);
+
+	if (t) {
+		t->execute(t->datap);
+		tps_task_free(t);
+	}
 
 	ao2_lock(tps);
 	/* We need to check size in the same critical section where we reset the
@@ -700,17 +684,17 @@
 	tps->executing = 0;
 	size = tps_taskprocessor_depth(tps);
 	/* If we executed a task, bump the stats */
-	tps->stats->_tasks_processed_count++;
-	if (size > tps->stats->max_qsize) {
-		tps->stats->max_qsize = size;
+	if (t && tps->stats) {
+		tps->stats->_tasks_processed_count++;
+		if (size > tps->stats->max_qsize) {
+			tps->stats->max_qsize = size;
+		}
 	}
 	ao2_unlock(tps);
 
 	/* If we executed a task, check for the transition to empty */
-	if (size == 0 && tps->listener->callbacks->emptied) {
+	if (t && size == 0 && tps->listener->callbacks->emptied) {
 		tps->listener->callbacks->emptied(tps->listener);
 	}
-
-	ao2_ref(tps, -1); /* task no longer has a reference */
 	return size > 0;
 }