change the code to accomodate the changes by tqueue

src/dnode/src/dnodeRead.c

@@ -90,7 +90,7 @@ void dnodeRead(SRpcMsg *pMsg) {
 
   while (leftLen > 0) {
     SMsgHead *pHead = (SMsgHead *) pCont;
-    pHead->vgId    = 1;//htonl(pHead->vgId);
+    pHead->vgId    = 1;   //htonl(pHead->vgId);
     pHead->contLen = pMsg->contLen; //htonl(pHead->contLen);
 
     void *pVnode = dnodeGetVnode(pHead->vgId);
@@ -101,22 +101,19 @@ void dnodeRead(SRpcMsg *pMsg) {
     }
 
     // put message into queue
-    SReadMsg readMsg;
-    readMsg.rpcMsg      = *pMsg;
-    readMsg.pCont       = pCont;
-    readMsg.contLen     = pHead->contLen;
-    readMsg.pRpcContext = pRpcContext;
-    readMsg.pVnode      = pVnode;
+    SReadMsg *pRead = (SReadMsg *)taosAllocateQitem(sizeof(SReadMsg));
+    pRead->rpcMsg      = *pMsg;
+    pRead->pCont       = pCont;
+    pRead->contLen     = pHead->contLen;
+    pRead->pRpcContext = pRpcContext;
 
     taos_queue queue = dnodeGetVnodeRworker(pVnode);
-    taosWriteQitem(queue, 0, pReadMsg);
+    taosWriteQitem(queue, TAOS_QTYPE_RPC, pRead);
 
     // next vnode
     leftLen -= pHead->contLen;
     pCont -= pHead->contLen;
     queuedMsgNum++;
-
-    dnodeReleaseVnode(pVnode);
   }
 
   if (queuedMsgNum == 0) {
@@ -179,6 +176,8 @@ static void *dnodeProcessReadQueue(void *param) {
 
     dnodeProcessReadResult(pVnode, pReadMsg);
     taosFreeQitem(pReadMsg);
+
+    dnodeReleaseVnode(pVnode);  
   }
 
   return NULL;

src/dnode/src/dnodeWrite.c

@@ -115,15 +115,14 @@ void dnodeWrite(SRpcMsg *pMsg) {
     }
    
     // put message into queue
-    SWriteMsg writeMsg;
-    writeMsg.rpcMsg      = *pMsg;
-    writeMsg.pCont       = pCont;
-    writeMsg.contLen     = pHead->contLen;
-    writeMsg.pRpcContext = pRpcContext;
-    writeMsg.pVnode      = pVnode;  // pVnode shall be saved for usage later
+    SWriteMsg *pWrite = (SWriteMsg *)taosAllocateQitem(sizeof(SWriteMsg));
+    pWrite->rpcMsg      = *pMsg;
+    pWrite->pCont       = pCont;
+    pWrite->contLen     = pHead->contLen;
+    pWrite->pRpcContext = pRpcContext;
  
     taos_queue queue = dnodeGetVnodeWworker(pVnode);
-    taosWriteQitem(queue, &writeMsg);
+    taosWriteQitem(queue, TAOS_QTYPE_RPC, pWrite);
 
     // next vnode 
     leftLen -= pHead->contLen;
@@ -145,14 +144,14 @@ void dnodeWrite(SRpcMsg *pMsg) {
 
 void *dnodeAllocateWriteWorker(void *pVnode) {
   SWriteWorker *pWorker = wWorkerPool.writeWorker + wWorkerPool.nextId;
-  taos_queue *queue = taosOpenQueue(sizeof(SWriteMsg));
+  taos_queue *queue = taosOpenQueue();
   if (queue == NULL) return NULL;
 
   if (pWorker->qset == NULL) {
     pWorker->qset = taosOpenQset();
     if (pWorker->qset == NULL) return NULL;
 
-    taosAddIntoQset(pWorker->qset, queue);
+    taosAddIntoQset(pWorker->qset, queue, pVnode);
     wWorkerPool.nextId = (wWorkerPool.nextId + 1) % wWorkerPool.max;
 
     pthread_attr_t thAttr;
@@ -164,7 +163,7 @@ void *dnodeAllocateWriteWorker(void *pVnode) {
       taosCloseQset(pWorker->qset);
     }
   } else {
-    taosAddIntoQset(pWorker->qset, queue);
+    taosAddIntoQset(pWorker->qset, queue, pVnode);
     wWorkerPool.nextId = (wWorkerPool.nextId + 1) % wWorkerPool.max;
   }
 
@@ -185,8 +184,10 @@ static void *dnodeProcessWriteQueue(void *param) {
   int           type;
   void         *pVnode;
 
+  qall = taosAllocateQall();
+
   while (1) {
-    numOfMsgs = taosReadAllQitemsFromQset(pWorker->qset, &qall, &pVnode);
+    numOfMsgs = taosReadAllQitemsFromQset(pWorker->qset, qall, &pVnode);
     if (numOfMsgs <=0) { 
       dnodeHandleIdleWorker(pWorker);  // thread exit if no queues anymore
       continue;
@@ -215,12 +216,13 @@ static void *dnodeProcessWriteQueue(void *param) {
       }
      
       dnodeProcessWriteResult(pVnode, pWriteMsg);
+      taosFreeQitem(pWriteMsg);
     }
 
-    // free the Qitems;
-    taosFreeQitems(qall);
   }
 
+  taosFreeQall(qall);
+
   return NULL;
 }
 

src/rpc/test/rserver.c

@@ -28,10 +28,13 @@ void *qhandle = NULL;
 void processShellMsg() {
   static int num = 0;
   taos_qall  qall;
-  SRpcMsg    rpcMsg;
+  SRpcMsg   *pRpcMsg, rpcMsg;
+  int        type;
+ 
+  qall = taosAllocateQall();
 
   while (1) {
-    int numOfMsgs = taosReadAllQitems(qhandle, &qall);
+    int numOfMsgs = taosReadAllQitems(qhandle, qall);
     if (numOfMsgs <= 0) {
       usleep(1000);
       continue;
@@ -40,10 +43,10 @@ void processShellMsg() {
     tTrace("%d shell msgs are received", numOfMsgs);
 
     for (int i=0; i<numOfMsgs; ++i) {
-      taosGetQitem(qall, &rpcMsg);
+      taosGetQitem(qall, &type, (void **)&pRpcMsg);
  
       if (dataFd >=0) {
-        if ( write(dataFd, rpcMsg.pCont, rpcMsg.contLen) <0 ) {
+        if ( write(dataFd, pRpcMsg->pCont, pRpcMsg->contLen) <0 ) {
           tPrint("failed to write data file, reason:%s", strerror(errno));
         }
       }
@@ -62,19 +65,22 @@ void processShellMsg() {
   
     taosResetQitems(qall);
     for (int i=0; i<numOfMsgs; ++i) {
-      taosGetQitem(qall, &rpcMsg);
 
-      rpcFreeCont(rpcMsg.pCont);
+      taosGetQitem(qall, &type, (void **)&pRpcMsg);
+      rpcFreeCont(pRpcMsg->pCont);
+
       rpcMsg.pCont = rpcMallocCont(msgSize);
       rpcMsg.contLen = msgSize;
-      rpcMsg.handle = rpcMsg.handle;
+      rpcMsg.handle = pRpcMsg->handle;
       rpcMsg.code = 1;
       rpcSendResponse(&rpcMsg);
+
+      taosFreeQitem(pRpcMsg);
     }
 
-    taosFreeQitems(qall);
   }
 
+  taosFreeQall(qall);
 /*
   SRpcIpSet ipSet;
   ipSet.numOfIps = 1;
@@ -108,8 +114,13 @@ int retrieveAuthInfo(char *meterId, char *spi, char *encrypt, char *secret, char
 }
 
 void processRequestMsg(SRpcMsg *pMsg) {
-  tTrace("request is received, type:%d, contLen:%d", pMsg->msgType, pMsg->contLen);
-  taosWriteQitem(qhandle, pMsg); 
+  SRpcMsg *pTemp;
+
+  pTemp = taosAllocateQitem(sizeof(SRpcMsg));
+  memcpy(pTemp, pMsg, sizeof(SRpcMsg));
+
+  tTrace("request is received, type:%d, contLen:%d, item:%p", pMsg->msgType, pMsg->contLen, pTemp);
+  taosWriteQitem(qhandle, TAOS_QTYPE_RPC, pTemp); 
 }
 
 int main(int argc, char *argv[]) {
@@ -143,6 +154,7 @@ int main(int argc, char *argv[]) {
       commit = atoi(argv[++i]);
     } else if (strcmp(argv[i], "-d")==0 && i < argc-1) {
       rpcDebugFlag = atoi(argv[++i]);
+      ddebugFlag = rpcDebugFlag;
       uDebugFlag = rpcDebugFlag;
     } else {
       printf("\nusage: %s [options] \n", argv[0]);

src/util/inc/tqueue.h

@@ -20,6 +20,10 @@
 extern "C" {
 #endif
 
+#define TAOS_QTYPE_RPC      0
+#define TAOS_QTYPE_FWD      1
+#define TAOS_QTYPE_WAL      2 
+
 typedef void* taos_queue;
 typedef void* taos_qset;
 typedef void* taos_qall;
@@ -31,10 +35,11 @@ void       taosFreeQitem(void *item);
 int        taosWriteQitem(taos_queue, int type, void *item);
 int        taosReadQitem(taos_queue, int *type, void **pitem);
 
-int        taosReadAllQitems(taos_queue, taos_qall *);
+taos_qall  taosAllocateQall();
+void       taosFreeQall(taos_qall);
+int        taosReadAllQitems(taos_queue, taos_qall);
 int        taosGetQitem(taos_qall, int *type, void **pitem);
 void       taosResetQitems(taos_qall);
-void       taosFreeQitems(taos_qall);
 
 taos_qset  taosOpenQset();
 void       taosCloseQset();
@@ -43,7 +48,7 @@ void       taosRemoveFromQset(taos_qset, taos_queue);
 int        taosGetQueueNumber(taos_qset);
 
 int        taosReadQitemFromQset(taos_qset, int *type, void **pitem, void **handle);
-int        taosReadAllQitemsFromQset(taos_qset, taos_qall *, void **handle);
+int        taosReadAllQitemsFromQset(taos_qset, taos_qall, void **handle);
 
 int        taosGetQueueItemsNumber(taos_queue param);
 int        taosGetQsetItemsNumber(taos_qset param);

src/util/src/tqueue.c

@@ -92,6 +92,8 @@ void *taosAllocateQitem(int size) {
 void taosFreeQitem(void *param) {
   if (param == NULL) return;
 
+  //pTrace("item:%p is freed", param);
+
   char *temp = (char *)param;
   temp -= sizeof(STaosQnode);
   free(temp);
@@ -115,6 +117,8 @@ int taosWriteQitem(taos_queue param, int type, void *item) {
   queue->numOfItems++;
   if (queue->qset) atomic_add_fetch_32(&queue->qset->numOfItems, 1);
 
+  //pTrace("item:%p is put into queue, items:%d", item, queue->numOfItems);
+
   pthread_mutex_unlock(&queue->mutex);
 
   return 0;
@@ -137,6 +141,7 @@ int taosReadQitem(taos_queue param, int *type, void **pitem) {
       queue->numOfItems--;
       if (queue->qset) atomic_sub_fetch_32(&queue->qset->numOfItems, 1);
       code = 1;
+      //pTrace("item:%p is read out from queue, items:%d", *pitem, queue->numOfItems);
   } 
 
   pthread_mutex_unlock(&queue->mutex);
@@ -144,35 +149,38 @@ int taosReadQitem(taos_queue param, int *type, void **pitem) {
   return code;
 }
 
-int taosReadAllQitems(taos_queue param, taos_qall *res) {
+void *taosAllocateQall() {
+  void *p = malloc(sizeof(STaosQall));
+  return p;
+}
+
+void taosFreeQall(void *param) {
+  free(param);
+}
+
+int taosReadAllQitems(taos_queue param, taos_qall p2) {
   STaosQueue *queue = (STaosQueue *)param;
-  STaosQall  *qall = NULL;
+  STaosQall  *qall = (STaosQall *)p2;
   int         code = 0;
 
   pthread_mutex_lock(&queue->mutex);
 
   if (queue->head) {
-    qall = (STaosQall *) calloc(sizeof(STaosQall), 1);
-    if ( qall == NULL ) {
-      terrno = TSDB_CODE_NO_RESOURCE;
-      code = -1;
-    } else {
-      qall->current = queue->head;
-      qall->start = queue->head;
-      qall->numOfItems = queue->numOfItems;
-      qall->itemSize = queue->itemSize;
-      code = qall->numOfItems;
-
-      queue->head = NULL;
-      queue->tail = NULL;
-      queue->numOfItems = 0;
-      if (queue->qset) atomic_sub_fetch_32(&queue->qset->numOfItems, qall->numOfItems);
-    }
+    memset(qall, 0, sizeof(STaosQall));
+    qall->current = queue->head;
+    qall->start = queue->head;
+    qall->numOfItems = queue->numOfItems;
+    qall->itemSize = queue->itemSize;
+    code = qall->numOfItems;
+
+    queue->head = NULL;
+    queue->tail = NULL;
+    queue->numOfItems = 0;
+    if (queue->qset) atomic_sub_fetch_32(&queue->qset->numOfItems, qall->numOfItems);
   } 
 
   pthread_mutex_unlock(&queue->mutex);
   
-  *res = qall;
   return code; 
 }
 
@@ -189,6 +197,7 @@ int taosGetQitem(taos_qall param, int *type, void **pitem) {
     *pitem = pNode->item;
     *type = pNode->type;
     num = 1;
+    //pTrace("item:%p is fetched", *pitem);
   }
 
   return num;
@@ -199,19 +208,6 @@ void taosResetQitems(taos_qall param) {
   qall->current = qall->start;
 }
 
-void taosFreeQitems(taos_qall param) {
-  STaosQall  *qall = (STaosQall *)param;
-  STaosQnode *pNode;
-
-  while (qall->current) {
-    pNode = qall->current;
-    qall->current = pNode->next;
-    free(pNode);
-  }
-
-  free(qall);
-}
-
 taos_qset taosOpenQset() {
 
   STaosQset *qset = (STaosQset *) calloc(sizeof(STaosQset), 1);
@@ -329,10 +325,10 @@ int taosReadQitemFromQset(taos_qset param, int *type, void **pitem, void **phand
   return code; 
 }
 
-int taosReadAllQitemsFromQset(taos_qset param, taos_qall *res, void **phandle) {
+int taosReadAllQitemsFromQset(taos_qset param, taos_qall p2, void **phandle) {
   STaosQset  *qset = (STaosQset *)param;
   STaosQueue *queue;
-  STaosQall  *qall = NULL;
+  STaosQall  *qall = (STaosQall *)p2;
   int         code = 0;
 
   for(int i=0; i<qset->numOfQueues; ++i) {
@@ -347,23 +343,17 @@ int taosReadAllQitemsFromQset(taos_qset param, taos_qall *res, void **phandle) {
     pthread_mutex_lock(&queue->mutex);
 
     if (queue->head) {
-      qall = (STaosQall *) calloc(sizeof(STaosQall), 1);
-      if (qall == NULL) {
-        terrno = TSDB_CODE_NO_RESOURCE;
-        code = -1;
-      } else {
-        qall->current = queue->head;
-        qall->start = queue->head;
-        qall->numOfItems = queue->numOfItems;
-        qall->itemSize = queue->itemSize;
-        code = qall->numOfItems;
-        *phandle = queue->ahandle;
+      qall->current = queue->head;
+      qall->start = queue->head;
+      qall->numOfItems = queue->numOfItems;
+      qall->itemSize = queue->itemSize;
+      code = qall->numOfItems;
+      *phandle = queue->ahandle;
           
-        queue->head = NULL;
-        queue->tail = NULL;
-        queue->numOfItems = 0;
-        atomic_sub_fetch_32(&qset->numOfItems, qall->numOfItems);
-      }
+      queue->head = NULL;
+      queue->tail = NULL;
+      queue->numOfItems = 0;
+      atomic_sub_fetch_32(&qset->numOfItems, qall->numOfItems);
     } 
 
     pthread_mutex_unlock(&queue->mutex);
@@ -371,8 +361,6 @@ int taosReadAllQitemsFromQset(taos_qset param, taos_qall *res, void **phandle) {
     if (code != 0) break;  
   }
 
-  *res = qall;
-
   return code;
 }
 

src/util/src/tsocket.c

@@ -330,7 +330,7 @@ int taosOpenTcpClientSocket(char *destIp, uint16_t destPort, char *clientIp) {
   struct sockaddr_in serverAddr, clientAddr;
   int                ret;
 
-  pTrace("open tcp client socket:%s:%d, local Ip:%s", destIp, destPort, clientIp);
+  // pTrace("open tcp client socket:%s:%d, local Ip:%s", destIp, destPort, clientIp);
 
   sockFd = (int)socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
 
@@ -362,7 +362,7 @@ int taosOpenTcpClientSocket(char *destIp, uint16_t destPort, char *clientIp) {
   ret = connect(sockFd, (struct sockaddr *)&serverAddr, sizeof(serverAddr));
 
   if (ret != 0) {
-    pError("failed to connect socket, ip:%s, port:%hu, reason: %s", destIp, destPort, strerror(errno));
+    //pError("failed to connect socket, ip:%s, port:%hu, reason: %s", destIp, destPort, strerror(errno));
     taosCloseSocket(sockFd);
     sockFd = -1;
   }