change tqueue

src/dnode/inc/dnodeWrite.h

@@ -23,7 +23,7 @@ extern "C" {
 int32_t dnodeInitWrite();
 void    dnodeCleanupWrite();
 void    dnodeWrite(void *pMsg);
-void *  dnodeAllocateWriteWorker();
+void *  dnodeAllocateWriteWorker(void *pVnode);
 void    dnodeFreeWriteWorker(void *worker);
 
 #ifdef __cplusplus

src/dnode/src/dnodeMgmt.c

@@ -152,8 +152,8 @@ static int32_t dnodeOpenVnode(int32_t vgId) {
   vnodeObj.status   = TSDB_VN_STATUS_NOT_READY;
   vnodeObj.refCount = 1;
   vnodeObj.version  = 0;
-  vnodeObj.wworker  = dnodeAllocateWriteWorker();
-  vnodeObj.rworker  = dnodeAllocateReadWorker();
+  vnodeObj.wworker  = dnodeAllocateWriteWorker(&vnodeObj);
+  vnodeObj.rworker  = dnodeAllocateReadWorker(&vnodeObj);
   vnodeObj.wal      = NULL;
   vnodeObj.tsdb     = pTsdb;
   vnodeObj.replica  = NULL;
@@ -217,8 +217,8 @@ static int32_t dnodeCreateVnode(SMDCreateVnodeMsg *pVnodeCfg) {
   vnodeObj.status   = TSDB_VN_STATUS_NOT_READY;
   vnodeObj.refCount = 1;
   vnodeObj.version  = 0;
-  vnodeObj.wworker  = dnodeAllocateWriteWorker();
-  vnodeObj.rworker  = dnodeAllocateReadWorker();
+  vnodeObj.wworker  = dnodeAllocateWriteWorker(&vnodeObj);
+  vnodeObj.rworker  = dnodeAllocateReadWorker(&vnodeObj);
   vnodeObj.wal      = NULL;
   vnodeObj.tsdb     = pTsdb;
   vnodeObj.replica  = NULL;

src/dnode/src/dnodeRead.c

@@ -33,16 +33,15 @@ typedef struct {
   void        *pCont;
   int32_t      contLen;
   SRpcMsg      rpcMsg;
-  void        *pVnode;
   SRpcContext *pRpcContext;  // RPC message context
 } SReadMsg;
 
 static void *dnodeProcessReadQueue(void *param);
-static void  dnodeProcessReadResult(SReadMsg *pRead);
+static void  dnodeProcessReadResult(void *pVnode, SReadMsg *pRead);
 static void  dnodeHandleIdleReadWorker();
-static void  dnodeProcessQueryMsg(SReadMsg *pMsg);
-static void  dnodeProcessRetrieveMsg(SReadMsg *pMsg);
-static void(*dnodeProcessReadMsgFp[TSDB_MSG_TYPE_MAX])(SReadMsg *pNode);
+static void  dnodeProcessQueryMsg(void *pVnode, SReadMsg *pMsg);
+static void  dnodeProcessRetrieveMsg(void *pVnode, SReadMsg *pMsg);
+static void(*dnodeProcessReadMsgFp[TSDB_MSG_TYPE_MAX])(void *pVnode, SReadMsg *pNode);
 
 // module global variable
 static taos_qset readQset;
@@ -93,15 +92,14 @@ void dnodeRead(void *rpcMsg) {
     }
 
     // put message into queue
-    SReadMsg readMsg;
-    readMsg.rpcMsg      = *pMsg;
-    readMsg.pCont       = pCont;
-    readMsg.contLen     = contLen;
-    readMsg.pRpcContext = pRpcContext;
-    readMsg.pVnode      = pVnode;
+    SReadMsg *pReadMsg = taosAllocateQitem(sizeof(SReadMsg));
+    pReadMsg->rpcMsg      = *pMsg;
+    pReadMsg->pCont       = pCont;
+    pReadMsg->contLen     = contLen;
+    pReadMsg->pRpcContext = pRpcContext;
 
     taos_queue queue = dnodeGetVnodeRworker(pVnode);
-    taosWriteQitem(queue, &readMsg);
+    taosWriteQitem(queue, 0, pReadMsg);
 
     // next vnode
     leftLen -= contLen;
@@ -111,11 +109,11 @@ void dnodeRead(void *rpcMsg) {
   }
 }
 
-void *dnodeAllocateReadWorker() {
+void *dnodeAllocateReadWorker(void *pVnode) {
   taos_queue *queue = taosOpenQueue(sizeof(SReadMsg));
   if (queue == NULL) return NULL;
 
-  taosAddIntoQset(readQset, queue);
+  taosAddIntoQset(readQset, queue, pVnode);
 
   // spawn a thread to process queue
   if (threads < maxThreads) {
@@ -140,22 +138,25 @@ void dnodeFreeReadWorker(void *rqueue) {
 
 static void *dnodeProcessReadQueue(void *param) {
   taos_qset  qset = (taos_qset)param;
-  SReadMsg   readMsg;
+  SReadMsg  *pReadMsg;
+  int        type;
+  void      *pVnode;
 
   while (1) {
-    if (taosReadQitemFromQset(qset, &readMsg) <= 0) {
+    if (taosReadQitemFromQset(qset, &type, &pReadMsg, &pVnode) == 0) {
       dnodeHandleIdleReadWorker();
       continue;
     }
 
     terrno = 0;
-    if (dnodeProcessReadMsgFp[readMsg.rpcMsg.msgType]) {
-      (*dnodeProcessReadMsgFp[readMsg.rpcMsg.msgType]) (&readMsg);
+    if (dnodeProcessReadMsgFp[pReadMsg->rpcMsg.msgType]) {
+      (*dnodeProcessReadMsgFp[pReadMsg->rpcMsg.msgType]) (pVnode, pReadMsg);
     } else {
       terrno = TSDB_CODE_MSG_NOT_PROCESSED;
     }
 
-    dnodeProcessReadResult(&readMsg);
+    dnodeProcessReadResult(pVnode, pReadMsg);
+    taosFreeQitem(pReadMsg);
   }
 
   return NULL;
@@ -173,11 +174,11 @@ static void dnodeHandleIdleReadWorker() {
   }
 }
 
-static void dnodeProcessReadResult(SReadMsg *pRead) {
+static void dnodeProcessReadResult(void *pVnode, SReadMsg *pRead) {
   SRpcContext *pRpcContext = pRead->pRpcContext;
   int32_t      code = 0;
 
-  dnodeReleaseVnode(pRead->pVnode);
+  dnodeReleaseVnode(pVnode);
 
   if (pRpcContext) {
     if (terrno) {
@@ -204,10 +205,10 @@ static void dnodeProcessReadResult(SReadMsg *pRead) {
   rpcFreeCont(pRead->rpcMsg.pCont);  // free the received message
 }
 
-static void dnodeProcessQueryMsg(SReadMsg *pMsg) {
+static void dnodeProcessQueryMsg(void *pVnode, SReadMsg *pMsg) {
 
 }
 
-static void dnodeProcessRetrieveMsg(SReadMsg *pMsg) {
+static void dnodeProcessRetrieveMsg(void *pVnode, SReadMsg *pMsg) {
 
 }

src/dnode/src/dnodeWrite.c

@@ -33,7 +33,6 @@ typedef struct _write {
   void        *pCont;
   int32_t      contLen;
   SRpcMsg      rpcMsg;
-  void        *pVnode;      // pointer to vnode
   SRpcContext *pRpcContext; // RPC message context
 } SWriteMsg;
 
@@ -49,20 +48,20 @@ typedef struct _thread_obj {
   SWriteWorker  *writeWorker;
 } SWriteWorkerPool;
 
-static void (*dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MAX])(SWriteMsg *);
+static void (*dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MAX])(void *, SWriteMsg *);
 static void  *dnodeProcessWriteQueue(void *param);
 static void   dnodeHandleIdleWorker(SWriteWorker *pWorker);
-static void   dnodeProcessWriteResult(SWriteMsg *pWrite);
-static void   dnodeProcessSubmitMsg(SWriteMsg *pMsg);
-static void   dnodeProcessCreateTableMsg(SWriteMsg *pMsg);
-static void   dnodeProcessDropTableMsg(SWriteMsg *pMsg);
+static void   dnodeProcessWriteResult(void *pVnode, SWriteMsg *pWrite);
+static void   dnodeProcessSubmitMsg(void *pVnode, SWriteMsg *pMsg);
+static void   dnodeProcessCreateTableMsg(void *pVnode, SWriteMsg *pMsg);
+static void   dnodeProcessDropTableMsg(void *pVnode, SWriteMsg *pMsg);
 
 SWriteWorkerPool wWorkerPool;
 
 int32_t dnodeInitWrite() {
-  dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_SUBMIT]             = dnodeProcessSubmitMsg;
+  dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_SUBMIT]          = dnodeProcessSubmitMsg;
   dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MD_CREATE_TABLE] = dnodeProcessCreateTableMsg;
-  dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MD_DROP_TABLE] = dnodeProcessDropTableMsg;
+  dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MD_DROP_TABLE]   = dnodeProcessDropTableMsg;
 
   wWorkerPool.max = tsNumOfCores;
   wWorkerPool.writeWorker = (SWriteWorker *)calloc(sizeof(SWriteWorker), wWorkerPool.max);
@@ -107,15 +106,14 @@ void dnodeWrite(void *rpcMsg) {
     }
    
     // put message into queue
-    SWriteMsg writeMsg;
-    writeMsg.rpcMsg      = *pMsg;
-    writeMsg.pCont       = pCont;
-    writeMsg.contLen     = contLen;
-    writeMsg.pRpcContext = pRpcContext;
-    writeMsg.pVnode      = pVnode;  // pVnode shall be saved for usage later
+    SWriteMsg *pWriteMsg = taosAllocateQitem(sizeof(SWriteMsg));
+    pWriteMsg->rpcMsg      = *pMsg;
+    pWriteMsg->pCont       = pCont;
+    pWriteMsg->contLen     = contLen;
+    pWriteMsg->pRpcContext = pRpcContext;
  
     taos_queue queue = dnodeGetVnodeWworker(pVnode);
-    taosWriteQitem(queue, &writeMsg);
+    taosWriteQitem(queue, 0, pWriteMsg);
  
     // next vnode 
     leftLen -= contLen;
@@ -123,7 +121,7 @@ void dnodeWrite(void *rpcMsg) {
   }
 }
 
-void *dnodeAllocateWriteWorker() {
+void *dnodeAllocateWriteWorker(void *pVnode) {
   SWriteWorker *pWorker = wWorkerPool.writeWorker + wWorkerPool.nextId;
 
   if (pWorker->qset == NULL) {
@@ -140,9 +138,9 @@ void *dnodeAllocateWriteWorker() {
     }
   }
 
-  taos_queue *queue = taosOpenQueue(sizeof(SWriteMsg));
+  taos_queue *queue = taosOpenQueue();
   if (queue) {
-    taosAddIntoQset(pWorker->qset, queue);
+    taosAddIntoQset(pWorker->qset, queue, pVnode);
     wWorkerPool.nextId = (wWorkerPool.nextId + 1) % wWorkerPool.max;
   }
  
@@ -158,11 +156,13 @@ void dnodeFreeWriteWorker(void *wqueue) {
 static void *dnodeProcessWriteQueue(void *param) {
   SWriteWorker *pWorker = (SWriteWorker *)param;
   taos_qall     qall;
-  SWriteMsg     writeMsg;
+  SWriteMsg    *pWriteMsg;
   int32_t       numOfMsgs;
+  int           type;
+  void         *pVnode;
 
   while (1) {
-    numOfMsgs = taosReadAllQitemsFromQset(pWorker->qset, &qall);
+    numOfMsgs = taosReadAllQitemsFromQset(pWorker->qset, &qall, &pVnode);
     if (numOfMsgs <=0) { 
       dnodeHandleIdleWorker(pWorker);  // thread exit if no queues anymore
       continue;
@@ -170,7 +170,7 @@ static void *dnodeProcessWriteQueue(void *param) {
 
     for (int32_t i=0; i<numOfMsgs; ++i) {
       // retrieve all items, and write them into WAL
-      taosGetQitem(qall, &writeMsg);
+      taosGetQitem(qall, &type, &pWriteMsg);
 
       // walWrite(pVnode->whandle, writeMsg.rpcMsg.msgType, writeMsg.pCont, writeMsg.contLen);
     }
@@ -181,16 +181,16 @@ static void *dnodeProcessWriteQueue(void *param) {
     // browse all items, and process them one by one
     taosResetQitems(qall);
     for (int32_t i = 0; i < numOfMsgs; ++i) {
-      taosGetQitem(qall, &writeMsg);
+      taosGetQitem(qall, &type, &pWriteMsg);
 
       terrno = 0;
-      if (dnodeProcessWriteMsgFp[writeMsg.rpcMsg.msgType]) {
-        (*dnodeProcessWriteMsgFp[writeMsg.rpcMsg.msgType]) (&writeMsg);
+      if (dnodeProcessWriteMsgFp[pWriteMsg->rpcMsg.msgType]) {
+        (*dnodeProcessWriteMsgFp[pWriteMsg->rpcMsg.msgType]) (pVnode, pWriteMsg);
       } else {
         terrno = TSDB_CODE_MSG_NOT_PROCESSED;  
       }
      
-      dnodeProcessWriteResult(&writeMsg);
+      dnodeProcessWriteResult(pVnode, pWriteMsg);
     }
 
     // free the Qitems;
@@ -200,11 +200,11 @@ static void *dnodeProcessWriteQueue(void *param) {
   return NULL;
 }
 
-static void dnodeProcessWriteResult(SWriteMsg *pWrite) {
+static void dnodeProcessWriteResult(void *pVnode, SWriteMsg *pWrite) {
   SRpcContext *pRpcContext = pWrite->pRpcContext;
   int32_t      code = 0;
 
-  dnodeReleaseVnode(pWrite->pVnode);
+  dnodeReleaseVnode(pVnode);
 
   if (pRpcContext) {
     if (terrno) {
@@ -244,14 +244,14 @@ static void dnodeHandleIdleWorker(SWriteWorker *pWorker) {
   }
 }
 
-static void dnodeProcessSubmitMsg(SWriteMsg *pMsg) {
+static void dnodeProcessSubmitMsg(void *param, SWriteMsg *pMsg) {
 
 }
 
-static void dnodeProcessCreateTableMsg(SWriteMsg *pMsg) {
+static void dnodeProcessCreateTableMsg(void *param, SWriteMsg *pMsg) {
 
 }
 
-static void dnodeProcessDropTableMsg(SWriteMsg *pMsg) {
+static void dnodeProcessDropTableMsg(void *param, SWriteMsg *pMsg) {
 
 }

src/util/inc/tlog.h

@@ -78,7 +78,7 @@ void taosResetLogFile();
 // utility log function
 #define pError(...)                          \
   if (uDebugFlag & DEBUG_ERROR) {            \
-    tprintf("ERROR UTL ", 255, __VA_ARGS__); \
+    tprintf("ERROR UTL ", uDebugFlag, __VA_ARGS__); \
   }
 #define pWarn(...)                                  \
   if (uDebugFlag & DEBUG_WARN) {                    \

src/util/inc/tqueue.h

@@ -24,24 +24,26 @@ typedef void* taos_queue;
 typedef void* taos_qset;
 typedef void* taos_qall;
 
-taos_queue taosOpenQueue(int itemSize);
+taos_queue taosOpenQueue();
 void       taosCloseQueue(taos_queue);
-int        taosWriteQitem(taos_queue, void *item);
-int        taosReadQitem(taos_queue, void *item);
+void      *taosAllocateQitem(int size);
+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 *);
-int        taosGetQitem(taos_qall, void *item);
+int        taosGetQitem(taos_qall, int *type, void **pitem);
 void       taosResetQitems(taos_qall);
 void       taosFreeQitems(taos_qall);
 
 taos_qset  taosOpenQset();
 void       taosCloseQset();
-int        taosAddIntoQset(taos_qset, taos_queue);
+int        taosAddIntoQset(taos_qset, taos_queue, void *ahandle);
 void       taosRemoveFromQset(taos_qset, taos_queue);
 int        taosGetQueueNumber(taos_qset);
 
-int        taosReadQitemFromQset(taos_qset, void *item);
-int        taosReadAllQitemsFromQset(taos_qset, taos_qall *);
+int        taosReadQitemFromQset(taos_qset, int *type, void **pitem, 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

@@ -19,6 +19,7 @@
 #include "tqueue.h"
 
 typedef struct _taos_qnode {
+  int                 type;
   struct _taos_qnode *next;
   char                item[];
 } STaosQnode;
@@ -30,6 +31,7 @@ typedef struct _taos_q {
   struct _taos_qnode *tail;
   struct _taos_q     *next;    // for queue set
   struct _taos_qset  *qset;    // for queue set
+  void               *ahandle; // for queue set
   pthread_mutex_t     mutex;  
 } STaosQueue;
 
@@ -48,7 +50,7 @@ typedef struct _taos_qall {
   int32_t       numOfItems;
 } STaosQall; 
   
-taos_queue taosOpenQueue(int itemSize) {
+taos_queue taosOpenQueue() {
   
   STaosQueue *queue = (STaosQueue *) calloc(sizeof(STaosQueue), 1);
   if (queue == NULL) {
@@ -57,8 +59,6 @@ taos_queue taosOpenQueue(int itemSize) {
   }
 
   pthread_mutex_init(&queue->mutex, NULL);
-  queue->itemSize = (int32_t)itemSize;
-
   return queue;
 }
 
@@ -83,16 +83,24 @@ void taosCloseQueue(taos_queue param) {
   free(queue);
 }
 
-int taosWriteQitem(taos_queue param, void *item) {
-  STaosQueue *queue = (STaosQueue *)param;
+void *taosAllocateQitem(int size) {
+  STaosQnode *pNode = (STaosQnode *)calloc(sizeof(STaosQnode) + size, 1);
+  if (pNode == NULL) return NULL;
+  return (void *)pNode->item;
+}
 
-  STaosQnode *pNode = (STaosQnode *)calloc(sizeof(STaosQnode) + queue->itemSize, 1);
-  if ( pNode == NULL ) {
-    terrno = TSDB_CODE_NO_RESOURCE;
-    return -1;
-  }
+void taosFreeQitem(void *param) {
+  if (param == NULL) return;
+
+  char *temp = (char *)param;
+  temp -= sizeof(STaosQnode);
+  free(temp);
+}
 
-  memcpy(pNode->item, item, queue->itemSize);
+int taosWriteQitem(taos_queue param, int type, void *item) {
+  STaosQueue *queue = (STaosQueue *)param;
+  STaosQnode *pNode = (STaosQnode *)((char *)item - sizeof(STaosQnode));
+  pNode->type = type;
 
   pthread_mutex_lock(&queue->mutex);
 
@@ -112,7 +120,7 @@ int taosWriteQitem(taos_queue param, void *item) {
   return 0;
 }
 
-int taosReadQitem(taos_queue param, void *item) {
+int taosReadQitem(taos_queue param, int *type, void **pitem) {
   STaosQueue *queue = (STaosQueue *)param;
   STaosQnode *pNode = NULL;
   int         code = 0;
@@ -121,11 +129,11 @@ int taosReadQitem(taos_queue param, void *item) {
 
   if (queue->head) {
       pNode = queue->head;
-      memcpy(item, pNode->item, queue->itemSize);
+      *pitem = pNode->item;
+      *type = pNode->type;
       queue->head = pNode->next;
       if (queue->head == NULL) 
         queue->tail = NULL;
-      free(pNode);
       queue->numOfItems--;
       if (queue->qset) atomic_sub_fetch_32(&queue->qset->numOfItems, 1);
       code = 1;
@@ -168,7 +176,7 @@ int taosReadAllQitems(taos_queue param, taos_qall *res) {
   return code; 
 }
 
-int taosGetQitem(taos_qall param, void *item) {
+int taosGetQitem(taos_qall param, int *type, void **pitem) {
   STaosQall  *qall = (STaosQall *)param;
   STaosQnode *pNode;
   int         num = 0;
@@ -178,7 +186,7 @@ int taosGetQitem(taos_qall param, void *item) {
     qall->current = pNode->next;
  
   if (pNode) {
-    memcpy(item, pNode->item, qall->itemSize);
+    *pitem = pNode->item;
     num = 1;
   }
 
@@ -221,7 +229,7 @@ void taosCloseQset(taos_qset param) {
   free(qset);
 }
 
-int taosAddIntoQset(taos_qset p1, taos_queue p2) {
+int taosAddIntoQset(taos_qset p1, taos_queue p2, void *ahandle) {
   STaosQueue *queue = (STaosQueue *)p2;
   STaosQset  *qset = (STaosQset *)p1;
 
@@ -230,6 +238,7 @@ int taosAddIntoQset(taos_qset p1, taos_queue p2) {
   pthread_mutex_lock(&qset->mutex);
 
   queue->next = qset->head;
+  queue->ahandle = ahandle;
   qset->head = queue;
   qset->numOfQueues++;
 
@@ -283,7 +292,7 @@ int taosGetQueueNumber(taos_qset param) {
   return ((STaosQset *)param)->numOfQueues;
 }
 
-int taosReadQitemFromQset(taos_qset param, void *item) {
+int taosReadQitemFromQset(taos_qset param, int *type, void **pitem, void **phandle) {
   STaosQset  *qset = (STaosQset *)param;
   STaosQnode *pNode = NULL;
   int         code = 0;
@@ -301,11 +310,12 @@ int taosReadQitemFromQset(taos_qset param, void *item) {
 
     if (queue->head) {
         pNode = queue->head;
-        memcpy(item, pNode->item, queue->itemSize);
+        *pitem = pNode->item;
+        *type = pNode->type;
+        *phandle = queue->ahandle;
         queue->head = pNode->next;
         if (queue->head == NULL) 
           queue->tail = NULL;
-        free(pNode);
         queue->numOfItems--;
         atomic_sub_fetch_32(&qset->numOfItems, 1);
         code = 1;
@@ -318,7 +328,7 @@ int taosReadQitemFromQset(taos_qset param, void *item) {
   return code; 
 }
 
-int taosReadAllQitemsFromQset(taos_qset param, taos_qall *res) {
+int taosReadAllQitemsFromQset(taos_qset param, taos_qall *res, void **phandle) {
   STaosQset  *qset = (STaosQset *)param;
   STaosQueue *queue;
   STaosQall  *qall = NULL;
@@ -346,6 +356,7 @@ int taosReadAllQitemsFromQset(taos_qset param, taos_qall *res) {
         qall->numOfItems = queue->numOfItems;
         qall->itemSize = queue->itemSize;
         code = qall->numOfItems;
+        *phandle = queue->ahandle;
           
         queue->head = NULL;
         queue->tail = NULL;

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", destIp, destPort);
+  pTrace("open tcp client socket:%s:%d, local Ip:%s", destIp, destPort, clientIp);
 
   sockFd = (int)socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);