From 5eaae4f71816f83600ccc7afc9e75cc106612726 Mon Sep 17 00:00:00 2001
From: Jeff Tao <jhtao@taosdata.com>
Date: Sat, 1 Aug 2020 11:49:58 +0000
Subject: [PATCH 1/2] add more trace for tqueue

---
 src/util/src/tqueue.c | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/src/util/src/tqueue.c b/src/util/src/tqueue.c
index d9abf0d7c3..1482d49d5b 100644
--- a/src/util/src/tqueue.c
+++ b/src/util/src/tqueue.c
@@ -61,6 +61,7 @@ taos_queue taosOpenQueue() {
 
   pthread_mutex_init(&queue->mutex, NULL);
 
+  uTrace("queue:%p is openned", queue);
   return queue;
 }
 
@@ -89,6 +90,8 @@ void taosCloseQueue(taos_queue param) {
   pthread_mutex_unlock(&queue->mutex);
   pthread_mutex_destroy(&queue->mutex);
   free(queue);
+
+  uTrace("queue:%p is closed", queue);
 }
 
 void *taosAllocateQitem(int size) {
@@ -161,7 +164,7 @@ int taosReadQitem(taos_queue param, int *type, void **pitem) {
 }
 
 void *taosAllocateQall() {
-  void *p = malloc(sizeof(STaosQall));
+  void *p = calloc(sizeof(STaosQall), 1);
   return p;
 }
 
@@ -230,6 +233,7 @@ taos_qset taosOpenQset() {
   pthread_mutex_init(&qset->mutex, NULL);
   tsem_init(&qset->sem, 0, 0);
 
+  uTrace("qset:%p is openned", qset);
   return qset;
 }
 
@@ -239,6 +243,7 @@ void taosCloseQset(taos_qset param) {
   pthread_mutex_destroy(&qset->mutex);
   tsem_destroy(&qset->sem);
   free(qset);
+  uTrace("qset:%p is closed", qset);
 }
 
 // tsem_post 'qset->sem', so that reader threads waiting for it
@@ -269,6 +274,7 @@ int taosAddIntoQset(taos_qset p1, taos_queue p2, void *ahandle) {
 
   pthread_mutex_unlock(&qset->mutex);
 
+  uTrace("queue:%p is added into qset:%p", queue, qset);
   return 0;
 }
 
@@ -288,6 +294,7 @@ void taosRemoveFromQset(taos_qset p1, taos_queue p2) {
       STaosQueue *prev = qset->head;
       tqueue = qset->head->next;
       while (tqueue) {
+        assert(tqueue->qset);
         if (tqueue== queue) {
           prev->next = tqueue->next;
           break;
@@ -310,6 +317,8 @@ void taosRemoveFromQset(taos_qset p1, taos_queue p2) {
   } 
   
   pthread_mutex_unlock(&qset->mutex);
+
+  uTrace("queue:%p is removed from qset:%p", queue, qset);
 }
 
 int taosGetQueueNumber(taos_qset param) {

From 0fdffcd2af4b23a3e226ca737b510f50696828c0 Mon Sep 17 00:00:00 2001
From: Jeff Tao <jhtao@taosdata.com>
Date: Sat, 1 Aug 2020 14:59:32 +0000
Subject: [PATCH 2/2] remove all queues from qset when qset is closed

---
 src/util/src/tqueue.c | 13 +++++++++++++
 1 file changed, 13 insertions(+)

diff --git a/src/util/src/tqueue.c b/src/util/src/tqueue.c
index 1482d49d5b..8c6d6243eb 100644
--- a/src/util/src/tqueue.c
+++ b/src/util/src/tqueue.c
@@ -240,6 +240,18 @@ taos_qset taosOpenQset() {
 void taosCloseQset(taos_qset param) {
   if (param == NULL) return;
   STaosQset *qset = (STaosQset *)param;
+
+  // remove all the queues from qset
+  pthread_mutex_lock(&qset->mutex);
+  while (qset->head) {
+    STaosQueue *queue = qset->head;
+    qset->head = qset->head->next;
+
+    queue->qset = NULL;
+    queue->next = NULL;
+  }
+  pthread_mutex_unlock(&qset->mutex);
+
   pthread_mutex_destroy(&qset->mutex);
   tsem_destroy(&qset->sem);
   free(qset);
@@ -312,6 +324,7 @@ void taosRemoveFromQset(taos_qset p1, taos_queue p2) {
       pthread_mutex_lock(&queue->mutex);
       atomic_sub_fetch_32(&qset->numOfItems, queue->numOfItems);
       queue->qset = NULL;
+      queue->next = NULL;
       pthread_mutex_unlock(&queue->mutex);
     }
   }