From 7c5f95db015921f3353f7c0bd36eb53bba267956 Mon Sep 17 00:00:00 2001
From: Victor Julien <victor@inliniac.net>
Date: Thu, 3 Sep 2009 23:11:09 +0200
Subject: [PATCH] Small threading update.

---
 src/tm-queues.c  | 15 ++++++++++++---
 src/tm-threads.c |  6 ++++--
 2 files changed, 16 insertions(+), 5 deletions(-)

diff --git a/src/tm-queues.c b/src/tm-queues.c
index f92a6b6232..dbddd1be70 100644
--- a/src/tm-queues.c
+++ b/src/tm-queues.c
@@ -24,6 +24,9 @@ Tmq* TmqCreateQueue(char *name) {
     if (tmq_id >= TMQ_MAX_QUEUES)
         goto error;
 
+#ifdef DEBUG
+    printf("TmqCreateQueue: created queue \'%s\'...\n", name);
+#endif
     Tmq *q = &tmqs[tmq_id];
     q->name = name;
     q->id = tmq_id++;
@@ -66,19 +69,25 @@ void TmqResetQueues(void) {
 void TmValidateQueueState(void)
 {
     int i = 0;
+    char err = FALSE;
 
     for (i = 0; i < tmq_id; i++) {
+        mutex_lock(&trans_q[tmqs[i].id].mutex_q);
         if (tmqs[i].reader_cnt == 0) {
             printf("Error: Queue \"%s\" doesn't have a reader\n", tmqs[i].name);
-            goto error;
+            err = TRUE;
         } else if (tmqs[i].writer_cnt == 0) {
             printf("Error: Queue \"%s\" doesn't have a writer\n", tmqs[i].name);
-            goto error;
+            err = TRUE;
         }
+        mutex_unlock(&trans_q[tmqs[i].id].mutex_q);
+
+        if (err == TRUE)
+            goto error;
     }
 
     return;
 
- error:
+error:
     exit(EXIT_FAILURE);
 }
diff --git a/src/tm-threads.c b/src/tm-threads.c
index ea2d2d63ca..1c9d3b80af 100644
--- a/src/tm-threads.c
+++ b/src/tm-threads.c
@@ -576,7 +576,7 @@ ThreadVars *TmThreadCreate(char *name, char *inq_name, char *inqh_name,
     tv->aof = THV_RESTART_THREAD;
 
     /* set the incoming queue */
-    if (inq_name != NULL) {
+    if (inq_name != NULL && strcmp(inq_name,"packetpool") != 0) {
         tmq = TmqGetQueueByName(inq_name);
         if (tmq == NULL) {
             tmq = TmqCreateQueue(inq_name);
@@ -603,9 +603,10 @@ ThreadVars *TmThreadCreate(char *name, char *inq_name, char *inqh_name,
         tv->tmqh_out = tmqh->OutHandler;
         //printf("TmThreadCreate: tv->tmqh_out %p\n", tv->tmqh_out);
 
-        if (outq_name != NULL) {
+        if (outq_name != NULL && strcmp(outq_name,"packetpool") != 0) {
             if (tmqh->OutHandlerCtxSetup != NULL) {
                 tv->outctx = tmqh->OutHandlerCtxSetup(outq_name);
+                tv->outq = NULL;
             } else {
                 tmq = TmqGetQueueByName(outq_name);
                 if (tmq == NULL) {
@@ -614,6 +615,7 @@ ThreadVars *TmThreadCreate(char *name, char *inq_name, char *inqh_name,
                 }
 
                 tv->outq = tmq;
+                tv->outctx = NULL;
                 tv->outq->writer_cnt++;
             }
             //printf("TmThreadCreate: tv->outq->id %" PRIu32 "\n", tv->outq->id);