Merge branch '3.0' into feature/tq

contrib/CMakeLists.txt

@@ -213,6 +213,7 @@ endif(${BUILD_WITH_TRAFT})
 
 # LIBUV
 if(${BUILD_WITH_UV})
+    add_compile_options(-Wno-sign-compare)
     if (${TD_WINDOWS})
         file(READ "libuv/include/uv.h" CONTENTS)
         string(REGEX REPLACE "/([\r]*)\nstruct uv_tcp_s {" "/\\1\ntypedef BOOL (PASCAL *LPFN_CONNECTEX) (SOCKET s, const struct sockaddr* name, int namelen, PVOID lpSendBuffer, DWORD dwSendDataLength,LPDWORD lpdwBytesSent, LPOVERLAPPED lpOverlapped);\\1\nstruct uv_tcp_s {" CONTENTS_NEW "${CONTENTS}")

include/common/tmsg.h

@@ -1920,13 +1920,14 @@ typedef struct {
 } SVCreateTSmaReq;
 
 typedef struct {
-  int8_t      type;                            // 0 status report, 1 update data
-  char        indexName[TSDB_INDEX_NAME_LEN];  //
-  STimeWindow windows;
+  int8_t  type;  // 0 status report, 1 update data
+  int64_t indexUid;
+  int64_t skey;  // start TS key of interval/sliding window
 } STSmaMsg;
 
 typedef struct {
   int64_t ver;  // use a general definition
+  int64_t indexUid;
   char    indexName[TSDB_INDEX_NAME_LEN];
 } SVDropTSmaReq;
 
@@ -2333,9 +2334,9 @@ struct SRpcMsg;
 struct SEpSet;
 struct SMgmtWrapper;
 typedef int32_t (*PutToQueueFp)(struct SMgmtWrapper* pWrapper, struct SRpcMsg* pReq);
-typedef int32_t (*SendReqFp)(struct SMgmtWrapper* pWrapper, struct SEpSet* epSet, struct SRpcMsg* rpcMsg);
-typedef int32_t (*SendMnodeReqFp)(struct SMgmtWrapper* pWrapper, struct SRpcMsg* rpcMsg);
-typedef void (*SendRspFp)(struct SMgmtWrapper* pWrapper, struct SRpcMsg* rpcMsg);
+typedef int32_t (*SendReqFp)(struct SMgmtWrapper* pWrapper, struct SEpSet* epSet, struct SRpcMsg* pReq);
+typedef int32_t (*SendMnodeReqFp)(struct SMgmtWrapper* pWrapper, struct SRpcMsg* pReq);
+typedef void (*SendRspFp)(struct SMgmtWrapper* pWrapper, struct SRpcMsg* pRsp);
 
 #ifdef __cplusplus
 }

include/libs/sync/sync.h

@@ -157,7 +157,8 @@ void    syncCleanUp();
 int64_t    syncStart(const SSyncInfo* pSyncInfo);
 void       syncStop(int64_t rid);
 int32_t    syncReconfig(int64_t rid, const SSyncCfg* pSyncCfg);
-int32_t    syncForwardToPeer(int64_t rid, const SRpcMsg* pMsg, bool isWeak);
+int32_t    syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak);        // use this function
+int32_t    syncForwardToPeer(int64_t rid, const SRpcMsg* pMsg, bool isWeak);  // just for compatibility
 ESyncState syncGetMyRole(int64_t rid);
 void       syncGetNodesRole(int64_t rid, SNodesRole* pNodeRole);
 

include/os/osSystem.h

@@ -28,7 +28,6 @@ extern "C" {
     #define tcgetattr TCGETATTR_FUNC_TAOS_FORBID
 #endif
 
-int32_t taosSystem(const char *cmd, char *buf, int32_t bufSize);
 void* taosLoadDll(const char* filename);
 void* taosLoadSym(void* handle, char* name);
 void  taosCloseDll(void* handle);

source/common/src/tmsg.c

@@ -2641,12 +2641,14 @@ int32_t tSerializeSVDropTSmaReq(void **buf, SVDropTSmaReq *pReq) {
   int32_t tlen = 0;
 
   tlen += taosEncodeFixedI64(buf, pReq->ver);
+  tlen += taosEncodeFixedI64(buf, pReq->indexUid);
   tlen += taosEncodeString(buf, pReq->indexName);
 
   return tlen;
 }
 void *tDeserializeSVDropTSmaReq(void *buf, SVDropTSmaReq *pReq) {
   buf = taosDecodeFixedI64(buf, &(pReq->ver));
+  buf = taosDecodeFixedI64(buf, &(pReq->indexUid));
   buf = taosDecodeStringTo(buf, pReq->indexName);
 
   return buf;

source/dnode/mgmt/vnode/inc/vmInt.h

@@ -24,6 +24,8 @@
 extern "C" {
 #endif
 
+typedef enum { VND_WRITE_QUEUE, VND_QUERY_QUEUE, VND_FETCH_QUEUE, VND_APPLY_QUEUE, VND_SYNC_QUEUE } EVndQueueType;
+
 typedef struct SVnodesMgmt {
   SHashObj     *hash;
   SRWLatch      latch;
@@ -102,7 +104,8 @@ int32_t vmAllocQueue(SVnodesMgmt *pMgmt, SVnodeObj *pVnode);
 void    vmFreeQueue(SVnodesMgmt *pMgmt, SVnodeObj *pVnode);
 
 int32_t vmPutMsgToQueryQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg);
-int32_t vmPutMsgToApplyQueue(SMgmtWrapper *pWrapper, int32_t vgId, SRpcMsg *pMsg);
+int32_t vmPutMsgToFetchQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg);
+int32_t vmPutMsgToApplyQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg);
 
 int32_t vmProcessWriteMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg);
 int32_t vmProcessSyncMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg);

source/dnode/mgmt/vnode/src/vmInt.c

@@ -296,7 +296,10 @@ static int32_t vmInit(SMgmtWrapper *pWrapper) {
 
   vnodeOpt.nthreads = tsNumOfCommitThreads;
   vnodeOpt.putToQueryQFp = vmPutMsgToQueryQueue;
+  vnodeOpt.putToFetchQFp = vmPutMsgToQueryQueue;
   vnodeOpt.sendReqFp = dndSendReqToDnode;
+  vnodeOpt.sendMnodeReqFp = dndSendReqToMnode;
+  vnodeOpt.sendRspFp = dndSendRsp;
   if (vnodeInit(&vnodeOpt) != 0) {
     dError("failed to init vnode since %s", terrstr());
     goto _OVER;

source/dnode/mgmt/vnode/src/vmWorker.c

@@ -16,46 +16,109 @@
 #define _DEFAULT_SOURCE
 #include "vmInt.h"
 
+static void vmSendRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) {
+  SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle, .code = code};
+  dndSendRsp(pWrapper, &rsp);
+}
+
+static void vmProcessMgmtQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
+  int32_t code = -1;
+  tmsg_t  msgType = pMsg->rpcMsg.msgType;
+  dTrace("msg:%p, will be processed in vnode-mgmt queue", pMsg);
+
+  switch (msgType) {
+    case TDMT_DND_CREATE_VNODE:
+      code = vmProcessCreateVnodeReq(pMgmt, pMsg);
+      break;
+    case TDMT_DND_ALTER_VNODE:
+      code = vmProcessAlterVnodeReq(pMgmt, pMsg);
+      break;
+    case TDMT_DND_DROP_VNODE:
+      code = vmProcessDropVnodeReq(pMgmt, pMsg);
+      break;
+    case TDMT_DND_SYNC_VNODE:
+      code = vmProcessSyncVnodeReq(pMgmt, pMsg);
+      break;
+    case TDMT_DND_COMPACT_VNODE:
+      code = vmProcessCompactVnodeReq(pMgmt, pMsg);
+      break;
+    default:
+      terrno = TSDB_CODE_MSG_NOT_PROCESSED;
+      dError("msg:%p, not processed in vnode-mgmt queue", pMsg);
+  }
+
+  if (msgType & 1u) {
+    if (code != 0 && terrno != 0) code = terrno;
+    vmSendRsp(pMgmt->pWrapper, pMsg, code);
+  }
+
+  dTrace("msg:%p, is freed, result:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code));
+  rpcFreeCont(pMsg->rpcMsg.pCont);
+  taosFreeQitem(pMsg);
+}
+
 static void vmProcessQueryQueue(SVnodeObj *pVnode, SNodeMsg *pMsg) {
-  dTrace("msg:%p, will be processed in vnode query queue", pMsg);
-  vnodeProcessQueryMsg(pVnode->pImpl, &pMsg->rpcMsg);
+  dTrace("msg:%p, will be processed in vnode-query queue", pMsg);
+  int32_t code = vnodeProcessQueryMsg(pVnode->pImpl, &pMsg->rpcMsg);
+  if (code != 0) {
+    vmSendRsp(pVnode->pWrapper, pMsg, code);
+  }
+
+  dTrace("msg:%p, is freed, result:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code));
+  rpcFreeCont(pMsg->rpcMsg.pCont);
+  taosFreeQitem(pMsg);
 }
 
 static void vmProcessFetchQueue(SVnodeObj *pVnode, SNodeMsg *pMsg) {
-  dTrace("msg:%p, will be processed in vnode fetch queue", pMsg);
-  vnodeProcessFetchMsg(pVnode->pImpl, &pMsg->rpcMsg);
+  dTrace("msg:%p, will be processed in vnode-fetch queue", pMsg);
+  int32_t code = vnodeProcessFetchMsg(pVnode->pImpl, &pMsg->rpcMsg);
+  if (code != 0) {
+    vmSendRsp(pVnode->pWrapper, pMsg, code);
+  }
+
+  dTrace("msg:%p, is freed, result:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code));
+  rpcFreeCont(pMsg->rpcMsg.pCont);
+  taosFreeQitem(pMsg);
 }
 
 static void vmProcessWriteQueue(SVnodeObj *pVnode, STaosQall *qall, int32_t numOfMsgs) {
   SArray *pArray = taosArrayInit(numOfMsgs, sizeof(SNodeMsg *));
+  if (pArray == NULL) {
+    dError("failed to process %d msgs in write-queue since %s", numOfMsgs, terrstr());
+    return;
+  }
 
   for (int32_t i = 0; i < numOfMsgs; ++i) {
     SNodeMsg *pMsg = NULL;
-    taosGetQitem(qall, (void **)&pMsg);
-    dTrace("msg:%p, will be processed in vnode write queue", pMsg);
-    void *ptr = taosArrayPush(pArray, &pMsg);
-    assert(ptr != NULL);
+    if (taosGetQitem(qall, (void **)&pMsg) == 0) continue;
+
+    dTrace("msg:%p, will be processed in vnode-write queue", pMsg);
+    if (taosArrayPush(pArray, &pMsg) == NULL) {
+      dTrace("msg:%p, failed to process since %s", pMsg, terrstr());
+      vmSendRsp(pVnode->pWrapper, pMsg, TSDB_CODE_OUT_OF_MEMORY);
+    }
   }
 
   vnodeProcessWMsgs(pVnode->pImpl, pArray);
 
-  for (size_t i = 0; i < numOfMsgs; i++) {
-    SRpcMsg  *pRsp = NULL;
+  numOfMsgs = taosArrayGetSize(pArray);
+  for (int32_t i = 0; i < numOfMsgs; i++) {
     SNodeMsg *pMsg = *(SNodeMsg **)taosArrayGet(pArray, i);
     SRpcMsg  *pRpc = &pMsg->rpcMsg;
-    int32_t   code = vnodeApplyWMsg(pVnode->pImpl, pRpc, &pRsp);
+    SRpcMsg  *pRsp = NULL;
+
+    int32_t code = vnodeApplyWMsg(pVnode->pImpl, pRpc, &pRsp);
     if (pRsp != NULL) {
       pRsp->ahandle = pRpc->ahandle;
       dndSendRsp(pVnode->pWrapper, pRsp);
       free(pRsp);
     } else {
-      if (code != 0) code = terrno;
-      SRpcMsg rpcRsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = code};
-      dndSendRsp(pVnode->pWrapper, &rpcRsp);
+      if (code != 0 && terrno != 0) code = terrno;
+      vmSendRsp(pVnode->pWrapper, pMsg, code);
     }
   }
 
-  for (size_t i = 0; i < numOfMsgs; i++) {
+  for (int32_t i = 0; i < numOfMsgs; i++) {
     SNodeMsg *pMsg = *(SNodeMsg **)taosArrayGet(pArray, i);
     dTrace("msg:%p, is freed", pMsg);
     rpcFreeCont(pMsg->rpcMsg.pCont);
@@ -89,93 +152,112 @@ static void vmProcessSyncQueue(SVnodeObj *pVnode, STaosQall *qall, int32_t numOf
   }
 }
 
-static SVnodeObj *vmAcquireFromMsg(SVnodesMgmt *pMgmt, SNodeMsg *pNodeMsg) {
-  SRpcMsg *pMsg = &pNodeMsg->rpcMsg;
+static int32_t vmPutNodeMsgToQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg, EVndQueueType qtype) {
+  SRpcMsg *pRpc = &pMsg->rpcMsg;
+  int32_t  code = -1;
 
-  SMsgHead *pHead = pMsg->pCont;
+  SMsgHead *pHead = pRpc->pCont;
   pHead->contLen = htonl(pHead->contLen);
   pHead->vgId = htonl(pHead->vgId);
 
   SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
   if (pVnode == NULL) {
-    dError("vgId:%d, failed to acquire vnode while process req", pHead->vgId);
+    dError("vgId:%d, failed to write msg:%p to queue since %s", pHead->vgId, pMsg, terrstr());
+    return -1;
   }
 
-  return pVnode;
-}
-
-int32_t vmProcessWriteMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
-  SVnodeObj *pVnode = vmAcquireFromMsg(pMgmt, pMsg);
-  if (pVnode == NULL) return -1;
+  switch (qtype) {
+    case VND_QUERY_QUEUE:
+      dTrace("msg:%p, will be written into vnode-query queue", pMsg);
+      code = taosWriteQitem(pVnode->pQueryQ, pMsg);
+      break;
+    case VND_FETCH_QUEUE:
+      dTrace("msg:%p, will be written into vnode-fetch queue", pMsg);
+      code = taosWriteQitem(pVnode->pFetchQ, pMsg);
+      break;
+    case VND_WRITE_QUEUE:
+      dTrace("msg:%p, will be written into vnode-write queue", pMsg);
+      code = taosWriteQitem(pVnode->pWriteQ, pMsg);
+    case VND_SYNC_QUEUE:
+      dTrace("msg:%p, will be written into vnode-sync queue", pMsg);
+      code = taosWriteQitem(pVnode->pSyncQ, pMsg);
+    default:
+      terrno = TSDB_CODE_INVALID_PARA;
+      break;
+  }
 
-  int32_t code = taosWriteQitem(pVnode->pWriteQ, pMsg);
   vmReleaseVnode(pMgmt, pVnode);
   return code;
 }
 
 int32_t vmProcessSyncMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
-  SVnodeObj *pVnode = vmAcquireFromMsg(pMgmt, pMsg);
-  if (pVnode == NULL) return -1;
+  return vmPutNodeMsgToQueue(pMgmt, pMsg, VND_SYNC_QUEUE);
+}
 
-  int32_t code = taosWriteQitem(pVnode->pSyncQ, pMsg);
-  vmReleaseVnode(pMgmt, pVnode);
-  return code;
+int32_t vmProcessWriteMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
+  return vmPutNodeMsgToQueue(pMgmt, pMsg, VND_WRITE_QUEUE);
 }
 
 int32_t vmProcessQueryMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
-  SVnodeObj *pVnode = vmAcquireFromMsg(pMgmt, pMsg);
-  if (pVnode == NULL) return -1;
-
-  int32_t code = taosWriteQitem(pVnode->pQueryQ, pMsg);
-  vmReleaseVnode(pMgmt, pVnode);
-  return code;
+  return vmPutNodeMsgToQueue(pMgmt, pMsg, VND_QUERY_QUEUE);
 }
 
 int32_t vmProcessFetchMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
-  SVnodeObj *pVnode = vmAcquireFromMsg(pMgmt, pMsg);
-  if (pVnode == NULL) return -1;
+  return vmPutNodeMsgToQueue(pMgmt, pMsg, VND_FETCH_QUEUE);
+}
 
-  int32_t code = taosWriteQitem(pVnode->pFetchQ, pMsg);
-  vmReleaseVnode(pMgmt, pVnode);
-  return code;
+int32_t vmProcessMgmtMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
+  SDnodeWorker *pWorker = &pMgmt->mgmtWorker;
+  dTrace("msg:%p, will be written to vnode-mgmt queue, worker:%s", pMsg, pWorker->name);
+  return dndWriteMsgToWorker(pWorker, pMsg);
 }
 
-int32_t vmPutMsgToQueryQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) {
+static int32_t vmPutRpcMsgToQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, EVndQueueType qtype) {
   SVnodesMgmt *pMgmt = pWrapper->pMgmt;
-
-  int32_t   code = -1;
-  SMsgHead *pHead = pRpc->pCont;
-  // pHead->vgId = htonl(pHead->vgId);
+  int32_t      code = -1;
+  SMsgHead    *pHead = pRpc->pCont;
 
   SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
   if (pVnode == NULL) return -1;
 
   SNodeMsg *pMsg = taosAllocateQitem(sizeof(SNodeMsg));
   if (pMsg != NULL) {
+    dTrace("msg:%p, is created, type:%s", pMsg, TMSG_INFO(pRpc->msgType));
     pMsg->rpcMsg = *pRpc;
-    code = taosWriteQitem(pVnode->pQueryQ, pMsg);
+    switch (qtype) {
+      case VND_QUERY_QUEUE:
+        dTrace("msg:%p, will be put into vnode-query queue", pMsg);
+        code = taosWriteQitem(pVnode->pQueryQ, pMsg);
+        break;
+      case VND_FETCH_QUEUE:
+        dTrace("msg:%p, will be put into vnode-fetch queue", pMsg);
+        code = taosWriteQitem(pVnode->pFetchQ, pMsg);
+        break;
+      case VND_APPLY_QUEUE:
+        dTrace("msg:%p, will be put into vnode-apply queue", pMsg);
+        code = taosWriteQitem(pVnode->pApplyQ, pMsg);
+        break;
+      case VND_WRITE_QUEUE:
+      case VND_SYNC_QUEUE:
+      default:
+        terrno = TSDB_CODE_INVALID_PARA;
+        break;
+    }
   }
   vmReleaseVnode(pMgmt, pVnode);
   return code;
 }
 
-int32_t vmPutMsgToApplyQueue(SMgmtWrapper *pWrapper, int32_t vgId, SRpcMsg *pRpc) {
-  SVnodesMgmt *pMgmt = pWrapper->pMgmt;
-
-  int32_t   code = -1;
-  SMsgHead *pHead = pRpc->pCont;
-  // pHead->vgId = htonl(pHead->vgId);
+int32_t vmPutMsgToQueryQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) {
+  return vmPutRpcMsgToQueue(pWrapper, pRpc, VND_QUERY_QUEUE);
+}
 
-  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
-  if (pVnode == NULL) return -1;
+int32_t vmPutMsgToFetchQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) {
+  return vmPutRpcMsgToQueue(pWrapper, pRpc, VND_FETCH_QUEUE);
+}
 
-  SNodeMsg *pMsg = taosAllocateQitem(sizeof(SNodeMsg));
-  if (pMsg != NULL) {
-    pMsg->rpcMsg = *pRpc;
-    code = taosWriteQitem(pVnode->pApplyQ, pMsg);
-  }
-  vmReleaseVnode(pMgmt, pVnode);
-  return code;
+int32_t vmPutMsgToApplyQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) {
+  return vmPutRpcMsgToQueue(pWrapper, pRpc, VND_APPLY_QUEUE);
 }
 
 int32_t vmAllocQueue(SVnodesMgmt *pMgmt, SVnodeObj *pVnode) {
@@ -191,6 +273,7 @@ int32_t vmAllocQueue(SVnodesMgmt *pMgmt, SVnodeObj *pVnode) {
     return -1;
   }
 
+  dDebug("vgId:%d, vnode queue is alloced", pVnode->vgId);
   return 0;
 }
 
@@ -205,43 +288,7 @@ void vmFreeQueue(SVnodesMgmt *pMgmt, SVnodeObj *pVnode) {
   pVnode->pSyncQ = NULL;
   pVnode->pFetchQ = NULL;
   pVnode->pQueryQ = NULL;
-}
-
-static void vmProcessMgmtQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
-  int32_t code = -1;
-  tmsg_t  msgType = pMsg->rpcMsg.msgType;
-  dTrace("msg:%p, will be processed in vnode mgmt queue", pMsg);
-
-  switch (msgType) {
-    case TDMT_DND_CREATE_VNODE:
-      code = vmProcessCreateVnodeReq(pMgmt, pMsg);
-      break;
-    case TDMT_DND_ALTER_VNODE:
-      code = vmProcessAlterVnodeReq(pMgmt, pMsg);
-      break;
-    case TDMT_DND_DROP_VNODE:
-      code = vmProcessDropVnodeReq(pMgmt, pMsg);
-      break;
-    case TDMT_DND_SYNC_VNODE:
-      code = vmProcessSyncVnodeReq(pMgmt, pMsg);
-      break;
-    case TDMT_DND_COMPACT_VNODE:
-      code = vmProcessCompactVnodeReq(pMgmt, pMsg);
-      break;
-    default:
-      terrno = TSDB_CODE_MSG_NOT_PROCESSED;
-      dError("msg:%p, not processed in mgmt queue", pMsg);
-  }
-
-  if (msgType & 1u) {
-    if (code != 0) code = terrno;
-    SRpcMsg rsp = {.code = code, .handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle};
-    dndSendRsp(pMgmt->pWrapper, &rsp);
-  }
-
-  dTrace("msg:%p, is freed, result:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code));
-  rpcFreeCont(pMsg->rpcMsg.pCont);
-  taosFreeQitem(pMsg);
+  dDebug("vgId:%d, vnode queue is freed", pVnode->vgId);
 }
 
 int32_t vmStartWorker(SVnodesMgmt *pMgmt) {
@@ -275,7 +322,7 @@ int32_t vmStartWorker(SVnodesMgmt *pMgmt) {
   if (tWWorkerInit(pWPool) != 0) return -1;
 
   if (dndInitWorker(pMgmt, &pMgmt->mgmtWorker, DND_WORKER_SINGLE, "vnode-mgmt", 1, 1, vmProcessMgmtQueue) != 0) {
-    dError("failed to start dnode mgmt worker since %s", terrstr());
+    dError("failed to start vnode-mgmt worker since %s", terrstr());
     return -1;
   }
 
@@ -291,9 +338,3 @@ void vmStopWorker(SVnodesMgmt *pMgmt) {
   tWWorkerCleanup(&pMgmt->syncPool);
   dDebug("vnode workers is closed");
 }
-
-int32_t vmProcessMgmtMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
-  SDnodeWorker *pWorker = &pMgmt->mgmtWorker;
-  dTrace("msg:%p, will be written to worker %s", pMsg, pWorker->name);
-  return dndWriteMsgToWorker(pWorker, pMsg);
-}
\ No newline at end of file

source/dnode/mnode/impl/src/mndDb.c

@@ -802,6 +802,32 @@ static int32_t mndSetDropDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *p
   return 0;
 }
 
+static int32_t mndBuildDropDbRsp(SDbObj *pDb, int32_t *pRspLen, void **ppRsp, bool useRpcMalloc) {
+  SDropDbRsp dropRsp = {0};
+  if (pDb != NULL) {
+    memcpy(dropRsp.db, pDb->name, TSDB_DB_FNAME_LEN);
+    dropRsp.uid = pDb->uid;
+  }
+
+  int32_t rspLen = tSerializeSDropDbRsp(NULL, 0, &dropRsp);
+  void   *pRsp = NULL;
+  if (useRpcMalloc) {
+    pRsp = rpcMallocCont(rspLen);
+  } else {
+    pRsp = malloc(rspLen);
+  }
+
+  if (pRsp == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+
+  tSerializeSDropDbRsp(pRsp, rspLen, &dropRsp);
+  *pRspLen = rspLen;
+  *ppRsp = pRsp;
+  return 0;
+}
+
 static int32_t mndDropDb(SMnode *pMnode, SNodeMsg *pReq, SDbObj *pDb) {
   int32_t code = -1;
   STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_DROP_DB, &pReq->rpcMsg);
@@ -814,18 +840,9 @@ static int32_t mndDropDb(SMnode *pMnode, SNodeMsg *pReq, SDbObj *pDb) {
   if (mndSetDropDbCommitLogs(pMnode, pTrans, pDb) != 0) goto DROP_DB_OVER;
   if (mndSetDropDbRedoActions(pMnode, pTrans, pDb) != 0) goto DROP_DB_OVER;
 
-  SDropDbRsp dropRsp = {0};
-  memcpy(dropRsp.db, pDb->name, TSDB_DB_FNAME_LEN);
-  dropRsp.uid = pDb->uid;
-
-  int32_t rspLen = tSerializeSDropDbRsp(NULL, 0, &dropRsp);
-  void   *pRsp = malloc(rspLen);
-  if (pRsp == NULL) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    goto DROP_DB_OVER;
-  }
-  tSerializeSDropDbRsp(pRsp, rspLen, &dropRsp);
-
+  int32_t rspLen = 0;
+  void   *pRsp = NULL;
+  if (mndBuildDropDbRsp(pDb, &rspLen, &pRsp, false) < 0) goto DROP_DB_OVER;
   mndTransSetRpcRsp(pTrans, pRsp, rspLen);
 
   if (mndTransPrepare(pMnode, pTrans) != 0) goto DROP_DB_OVER;
@@ -854,7 +871,7 @@ static int32_t mndProcessDropDbReq(SNodeMsg *pReq) {
   pDb = mndAcquireDb(pMnode, dropReq.db);
   if (pDb == NULL) {
     if (dropReq.ignoreNotExists) {
-      code = 0;
+      code = mndBuildDropDbRsp(pDb, &pReq->rspLen, &pReq->pRsp, true);
       goto DROP_DB_OVER;
     } else {
       terrno = TSDB_CODE_MND_DB_NOT_EXIST;

source/dnode/vnode/inc/vnode.h

@@ -62,6 +62,7 @@ typedef struct {
 typedef struct {
   uint16_t       nthreads;  // number of commit threads. 0 for no threads and a schedule queue should be given (TODO)
   PutToQueueFp   putToQueryQFp;
+  PutToQueueFp   putToFetchQFp;
   SendReqFp      sendReqFp;
   SendMnodeReqFp sendMnodeReqFp;
   SendRspFp      sendRspFp;
@@ -125,9 +126,8 @@ void vnodeDestroy(const char *path);
  *
  * @param pVnode The vnode object.
  * @param pMsgs The array of SRpcMsg
- * @return int 0 for success, -1 for failure
  */
-int vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs);
+void vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs);
 
 /**
  * @brief Apply a write request message.

source/dnode/vnode/src/inc/vnd.h

@@ -46,15 +46,18 @@ typedef struct SVnodeTask {
 typedef struct SVnodeMgr {
   td_mode_flag_t vnodeInitFlag;
   // For commit
-  bool            stop;
-  uint16_t        nthreads;
-  TdThread*      threads;
+  bool          stop;
+  uint16_t      nthreads;
+  TdThread*     threads;
   TdThreadMutex mutex;
   TdThreadCond  hasTask;
   TD_DLIST(SVnodeTask) queue;
   // For vnode Mgmt
-  PutToQueueFp putToQueryQFp;
-  SendReqFp    sendReqFp;
+  PutToQueueFp   putToQueryQFp;
+  PutToQueueFp   putToFetchQFp;
+  SendReqFp      sendReqFp;
+  SendMnodeReqFp sendMnodeReqFp;
+  SendRspFp      sendRspFp;
 } SVnodeMgr;
 
 extern SVnodeMgr vnodeMgr;
@@ -85,7 +88,10 @@ struct SVnode {
 int vnodeScheduleTask(SVnodeTask* task);
 
 int32_t vnodePutToVQueryQ(SVnode* pVnode, struct SRpcMsg* pReq);
-void    vnodeSendReq(SVnode* pVnode, struct SEpSet* epSet, struct SRpcMsg* pReq);
+int32_t vnodePutToVFetchQ(SVnode* pVnode, struct SRpcMsg* pReq);
+int32_t vnodeSendReq(SVnode* pVnode, struct SEpSet* epSet, struct SRpcMsg* pReq);
+int32_t vnodeSendMnodeReq(SVnode* pVnode, struct SRpcMsg* pReq);
+void    vnodeSendRsp(SVnode* pVnode, struct SEpSet* epSet, struct SRpcMsg* pRsp);
 
 #define vFatal(...)                                              \
   do {                                                           \

source/dnode/vnode/src/tsdb/tsdbFS.c

@@ -1289,7 +1289,7 @@ static int tsdbRestoreCurrent(STsdb *pRepo) {
   }
 
   if (tsdbSaveFSStatus(pRepo, pRepo->fs->cstatus) < 0) {
-    tsdbError("vgId:%d failed to restore corrent since %s", REPO_ID(pRepo), tstrerror(terrno));
+    tsdbError("vgId:%d failed to restore current since %s", REPO_ID(pRepo), tstrerror(terrno));
     return -1;
   }
 

source/dnode/vnode/src/vnd/vnodeMgr.c

@@ -26,7 +26,10 @@ int vnodeInit(const SVnodeOpt *pOption) {
 
   vnodeMgr.stop = false;
   vnodeMgr.putToQueryQFp = pOption->putToQueryQFp;
+  vnodeMgr.putToFetchQFp = pOption->putToFetchQFp;
   vnodeMgr.sendReqFp = pOption->sendReqFp;
+  vnodeMgr.sendMnodeReqFp = pOption->sendMnodeReqFp;
+  vnodeMgr.sendRspFp = pOption->sendRspFp;
 
   // Start commit handers
   if (pOption->nthreads > 0) {
@@ -90,15 +93,23 @@ int vnodeScheduleTask(SVnodeTask* pTask) {
 }
 
 int32_t vnodePutToVQueryQ(SVnode* pVnode, struct SRpcMsg* pReq) {
-  if (pVnode == NULL || pVnode->pMeta == NULL || vnodeMgr.putToQueryQFp == NULL) {
-    terrno = TSDB_CODE_VND_APP_ERROR;
-    return -1;
-  }
   return (*vnodeMgr.putToQueryQFp)(pVnode->pWrapper, pReq);
 }
 
-void vnodeSendReq(SVnode* pVnode, struct SEpSet* epSet, struct SRpcMsg* pReq) {
-  (*vnodeMgr.sendReqFp)(pVnode->pWrapper, epSet, pReq);
+int32_t vnodePutToVFetchQ(SVnode* pVnode, struct SRpcMsg* pReq) {
+  return (*vnodeMgr.putToFetchQFp)(pVnode->pWrapper, pReq);
+}
+
+int32_t vnodeSendReq(SVnode* pVnode, struct SEpSet* epSet, struct SRpcMsg* pReq) {
+  return (*vnodeMgr.sendReqFp)(pVnode->pWrapper, epSet, pReq);
+}
+
+int32_t vnodeSendMnodeReq(SVnode* pVnode, struct SRpcMsg* pReq) {
+  return (*vnodeMgr.sendMnodeReqFp)(pVnode->pWrapper, pReq);
+}
+
+void vnodeSendRsp(SVnode* pVnode, struct SEpSet* epSet, struct SRpcMsg* pRsp) {
+  (*vnodeMgr.sendRspFp)(pVnode->pWrapper, pRsp);
 }
 
 /* ------------------------ STATIC METHODS ------------------------ */

source/dnode/vnode/src/vnd/vnodeWrite.c

@@ -16,7 +16,7 @@
 #include "tq.h"
 #include "vnd.h"
 
-int vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs) {
+void vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs) {
   SNodeMsg *pMsg;
   SRpcMsg *pRpc;
 
@@ -40,7 +40,8 @@ int vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs) {
 
   // TODO: Integrate RAFT module here
 
-  return 0;
+  // No results are returned because error handling is difficult
+  // return 0;
 }
 
 int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {

source/libs/sync/inc/syncInt.h

@@ -142,7 +142,6 @@ typedef struct SSyncNode {
   SRaftId   leaderCache;
 
   // life cycle
-  int32_t refCount;
   int64_t rid;
 
   // tla+ server vars

source/libs/sync/src/syncMain.c

@@ -31,6 +31,7 @@
 #include "syncTimeout.h"
 #include "syncUtil.h"
 #include "syncVoteMgr.h"
+#include "tref.h"
 
 static int32_t tsNodeRefId = -1;
 
@@ -44,31 +45,57 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId);
 static int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg);
 static int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg);
 static int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg);
+
+// life cycle
+static void syncFreeNode(void* param);
 // ---------------------------------
 
 int32_t syncInit() {
-  int32_t ret = syncEnvStart();
+  int32_t ret;
+  tsNodeRefId = taosOpenRef(200, syncFreeNode);
+  if (tsNodeRefId < 0) {
+    sError("failed to init node ref");
+    syncCleanUp();
+    ret = -1;
+  } else {
+    ret = syncEnvStart();
+  }
+
   return ret;
 }
 
 void syncCleanUp() {
   int32_t ret = syncEnvStop();
   assert(ret == 0);
+
+  if (tsNodeRefId != -1) {
+    taosCloseRef(tsNodeRefId);
+    tsNodeRefId = -1;
+  }
 }
 
 int64_t syncStart(const SSyncInfo* pSyncInfo) {
-  int32_t    ret = 0;
   SSyncNode* pSyncNode = syncNodeOpen(pSyncInfo);
   assert(pSyncNode != NULL);
 
-  // todo : return ref id
-  return ret;
+  pSyncNode->rid = taosAddRef(tsNodeRefId, pSyncNode);
+  if (pSyncNode->rid < 0) {
+    syncFreeNode(pSyncNode);
+    return -1;
+  }
+
+  return pSyncNode->rid;
 }
 
 void syncStop(int64_t rid) {
-  // todo : get pointer from rid
-  SSyncNode* pSyncNode = NULL;
+  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
+  if (pSyncNode == NULL) {
+    return;
+  }
   syncNodeClose(pSyncNode);
+
+  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
+  taosRemoveRef(tsNodeRefId, rid);
 }
 
 int32_t syncReconfig(int64_t rid, const SSyncCfg* pSyncCfg) {
@@ -76,11 +103,16 @@ int32_t syncReconfig(int64_t rid, const SSyncCfg* pSyncCfg) {
   return ret;
 }
 
-int32_t syncForwardToPeer(int64_t rid, const SRpcMsg* pMsg, bool isWeak) {
+int32_t syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak) {
   int32_t ret = 0;
 
   // todo : get pointer from rid
-  SSyncNode* pSyncNode = NULL;
+  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
+  if (pSyncNode == NULL) {
+    return -1;
+  }
+  assert(rid == pSyncNode->rid);
+
   if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
     SyncClientRequest* pSyncMsg = syncClientRequestBuild2(pMsg, 0, isWeak);
     SRpcMsg            rpcMsg;
@@ -93,6 +125,13 @@ int32_t syncForwardToPeer(int64_t rid, const SRpcMsg* pMsg, bool isWeak) {
     sTrace("syncForwardToPeer not leader, %s", syncUtilState2String(pSyncNode->state));
     ret = -1;  // todo : need define err code !!
   }
+
+  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
+  return ret;
+}
+
+int32_t syncForwardToPeer(int64_t rid, const SRpcMsg* pMsg, bool isWeak) {
+  int32_t ret = syncPropose(rid, pMsg, isWeak);
   return ret;
 }
 
@@ -155,7 +194,7 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo) {
   pSyncNode->quorum = syncUtilQuorum(pSyncInfo->syncCfg.replicaNum);
   pSyncNode->leaderCache = EMPTY_RAFT_ID;
 
-  // init life cycle
+  // init life cycle outside
 
   // TLA+ Spec
   // InitHistoryVars == /\ elections = {}
@@ -444,6 +483,10 @@ cJSON* syncNode2Json(const SSyncNode* pSyncNode) {
     cJSON* pLaderCache = syncUtilRaftId2Json(&pSyncNode->leaderCache);
     cJSON_AddItemToObject(pRoot, "leaderCache", pLaderCache);
 
+    // life cycle
+    snprintf(u64buf, sizeof(u64buf), "%ld", pSyncNode->rid);
+    cJSON_AddStringToObject(pRoot, "rid", u64buf);
+
     // tla+ server vars
     cJSON_AddNumberToObject(pRoot, "state", pSyncNode->state);
     cJSON_AddStringToObject(pRoot, "state_str", syncUtilState2String(pSyncNode->state));
@@ -813,3 +856,10 @@ static int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg
   syncEntryDestory(pEntry);
   return ret;
 }
+
+static void syncFreeNode(void* param) {
+  SSyncNode* pNode = param;
+  syncNodePrint2((char*)"==syncFreeNode==", pNode);
+
+  free(pNode);
+}
\ No newline at end of file

source/libs/sync/test/CMakeLists.txt

@@ -31,6 +31,7 @@ add_executable(syncElectTest "")
 add_executable(syncEncodeTest "")
 add_executable(syncWriteTest "")
 add_executable(syncReplicateTest "")
+add_executable(syncRefTest "")
 
 
 target_sources(syncTest
@@ -165,6 +166,10 @@ target_sources(syncReplicateTest
     PRIVATE
     "syncReplicateTest.cpp"
 )
+target_sources(syncRefTest
+    PRIVATE
+    "syncRefTest.cpp"
+)
 
 
 target_include_directories(syncTest
@@ -337,6 +342,11 @@ target_include_directories(syncReplicateTest
     "${CMAKE_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
+target_include_directories(syncRefTest
+    PUBLIC
+    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+)
 
 
 target_link_libraries(syncTest
@@ -471,6 +481,10 @@ target_link_libraries(syncReplicateTest
     sync
     gtest_main
 )
+target_link_libraries(syncRefTest
+    sync
+    gtest_main
+)
 
 
 enable_testing()

source/libs/sync/test/syncElectTest.cpp

@@ -116,9 +116,10 @@ int main(int argc, char** argv) {
 
   //---------------------------
   while (1) {
-    sTrace("elect sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, electTimerMS:%d",
-           gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm, gSyncNode->electTimerLogicClock,
-           gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS);
+    sTrace(
+        "elect sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, electTimerMS:%d",
+        gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm,
+        gSyncNode->electTimerLogicClock, gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS);
   }
 
   return 0;

source/libs/sync/test/syncRefTest.cpp

+#include <gtest/gtest.h>
+#include <stdio.h>
+#include "syncIO.h"
+#include "syncInt.h"
+#include "syncRaftStore.h"
+#include "tref.h"
+
+void logTest() {
+  sTrace("--- sync log test: trace");
+  sDebug("--- sync log test: debug");
+  sInfo("--- sync log test: info");
+  sWarn("--- sync log test: warn");
+  sError("--- sync log test: error");
+  sFatal("--- sync log test: fatal");
+}
+
+static void syncFreeObj(void *param);
+int32_t     init();
+void        cleanup();
+int64_t     start();
+void        stop(int64_t rid);
+
+static int32_t tsNodeRefId = -1;
+int            g = 100;
+
+typedef struct SyncObj {
+  int64_t rid;
+  void *  data;
+  char    name[32];
+  int     counter;
+} SyncObj;
+
+static void syncFreeObj(void *param) {
+  SyncObj *pObj = (SyncObj *)param;
+  printf("syncFreeObj name:%s rid:%ld \n", pObj->name, pObj->rid);
+  free(pObj);
+}
+
+int32_t init() {
+  tsNodeRefId = taosOpenRef(200, syncFreeObj);
+  if (tsNodeRefId < 0) {
+    sError("failed to init node ref");
+    cleanup();
+    return -1;
+  }
+  return 0;
+}
+
+void cleanup() {
+  if (tsNodeRefId != -1) {
+    taosCloseRef(tsNodeRefId);
+    tsNodeRefId = -1;
+  }
+}
+
+int64_t start() {
+  SyncObj *pObj = (SyncObj *)malloc(sizeof(SyncObj));
+  assert(pObj != NULL);
+
+  pObj->data = &g;
+  snprintf(pObj->name, sizeof(pObj->name), "%s", "hello");
+
+  pObj->rid = taosAddRef(tsNodeRefId, pObj);
+  if (pObj->rid < 0) {
+    syncFreeObj(pObj);
+    return -1;
+  }
+
+  printf("start name:%s rid:%ld \n", pObj->name, pObj->rid);
+  return pObj->rid;
+}
+
+void stop(int64_t rid) {
+  SyncObj *pObj = (SyncObj *)taosAcquireRef(tsNodeRefId, rid);
+  if (pObj == NULL) return;
+
+  printf("stop name:%s rid:%ld \n", pObj->name, pObj->rid);
+  pObj->data = NULL;
+
+  taosReleaseRef(tsNodeRefId, pObj->rid);
+  taosRemoveRef(tsNodeRefId, rid);
+}
+
+void *func(void *param) {
+  int64_t rid = (int64_t)param;
+
+  int32_t ms = taosRand() % 10000;
+  taosMsleep(ms);
+
+  SyncObj *pObj = (SyncObj *)taosAcquireRef(tsNodeRefId, rid);
+  if (pObj != NULL) {
+    printf("taosAcquireRef sleep:%d, name:%s, rid:%ld \n", ms, pObj->name, pObj->rid);
+  } else {
+    printf("taosAcquireRef sleep:%d, NULL! \n", ms);
+  }
+
+  taosReleaseRef(tsNodeRefId, rid);
+  return NULL;
+}
+
+int main() {
+  // taosInitLog((char *)"syncTest.log", 100000, 10);
+  tsAsyncLog = 0;
+  sDebugFlag = 143 + 64;
+  logTest();
+
+  taosSeedRand(taosGetTimestampSec());
+
+  int32_t ret;
+
+  ret = init();
+  assert(ret == 0);
+
+  int64_t rid = start();
+  assert(rid > 0);
+
+  for (int i = 0; i < 20; ++i) {
+    TdThread tid;
+    taosThreadCreate(&tid, NULL, func, (void *)rid);
+  }
+
+  int32_t ms = taosRand() % 10000;
+  taosMsleep(ms);
+  printf("main sleep %d, stop and clean ", ms);
+
+  stop(rid);
+  cleanup();
+
+  while (1) {
+    taosMsleep(1000);
+    printf("sleep 1 ... \n");
+  }
+
+  return 0;
+}

source/libs/sync/test/syncReplicateTest.cpp

@@ -172,15 +172,19 @@ int main(int argc, char **argv) {
     gSyncNode->FpEqMsg(gSyncNode->queue, &rpcMsg);
 
     taosMsleep(1000);
-    sTrace("replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, electTimerMS:%d",
-           gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm, gSyncNode->electTimerLogicClock,
-           gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS);
+    sTrace(
+        "replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
+        "electTimerMS:%d",
+        gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm,
+        gSyncNode->electTimerLogicClock, gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS);
   }
 
   while (1) {
-    sTrace("replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, electTimerMS:%d",
-           gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm, gSyncNode->electTimerLogicClock,
-           gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS);
+    sTrace(
+        "replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
+        "electTimerMS:%d",
+        gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm,
+        gSyncNode->electTimerLogicClock, gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS);
     taosMsleep(1000);
   }
 

source/libs/transport/inc/transComm.h

@@ -228,8 +228,8 @@ typedef struct SConnBuffer {
 typedef void (*AsyncCB)(uv_async_t* handle);
 
 typedef struct {
-  void*           pThrd;
-  queue           qmsg;
+  void*         pThrd;
+  queue         qmsg;
   TdThreadMutex mtx;  // protect qmsg;
 } SAsyncItem;
 
@@ -273,11 +273,52 @@ void transCloseClient(void* arg);
 void transCloseServer(void* arg);
 
 void  transCtxInit(STransCtx* ctx);
-void  transCtxDestroy(STransCtx* ctx);
+void  transCtxCleanup(STransCtx* ctx);
 void  transCtxClear(STransCtx* ctx);
 void  transCtxMerge(STransCtx* dst, STransCtx* src);
 void* transCtxDumpVal(STransCtx* ctx, int32_t key);
 
+// queue sending msgs
+typedef struct {
+  SArray* q;
+  void (*free)(void* arg);
+} STransQueue;
+
+/*
+ * init queue
+ * note: queue'size is small, default 1
+ */
+void transQueueInit(STransQueue* queue, void (*free)(void* arg));
+
+/*
+ * put arg into queue
+ * if queue'size > 1, return false; else return true
+ */
+bool transQueuePush(STransQueue* queue, void* arg);
+/*
+ * pop head from queue
+ */
+
+void* transQueuePop(STransQueue* queue);
+/*
+ * get head from queue
+ */
+void* transQueueGet(STransQueue* queue);
+
+/*
+ * queue empty or not
+ */
+
+bool transQueueEmpty(STransQueue* queue);
+/*
+ * clear queue
+ */
+void transQueueClear(STransQueue* queue);
+/*
+ * destroy queue
+ */
+void transQueueDestroy(STransQueue* queue);
+
 #ifdef __cplusplus
 }
 #endif

source/libs/transport/src/transCli.c

@@ -25,13 +25,14 @@ typedef struct SCliConn {
   void*        hostThrd;
   SConnBuffer  readBuf;
   void*        data;
-  SArray*      cliMsgs;
-  queue        conn;
-  uint64_t     expireTime;
-  int          hThrdIdx;
-  bool         broken;  // link broken or not
-  STransCtx    ctx;
-
+  // SArray*      cliMsgs;
+  STransQueue cliMsgs;
+  queue       conn;
+  uint64_t    expireTime;
+  int         hThrdIdx;
+  STransCtx   ctx;
+
+  bool       broken;   // link broken or not
   ConnStatus status;   //
   int        release;  // 1: release
   // spi configure
@@ -56,14 +57,14 @@ typedef struct SCliMsg {
 } SCliMsg;
 
 typedef struct SCliThrdObj {
-  TdThread   thread;
+  TdThread    thread;
   uv_loop_t*  loop;
   SAsyncPool* asyncPool;
   uv_timer_t  timer;
   void*       pool;  // conn pool
 
   // msg queue
-  queue           msg;
+  queue         msg;
   TdThreadMutex msgMtx;
 
   uint64_t nextTimeout;  // next timeout
@@ -181,12 +182,11 @@ static void         destroyThrdObj(SCliThrdObj* pThrd);
 static void* cliWorkThread(void* arg);
 
 bool cliMaySendCachedMsg(SCliConn* conn) {
-  if (taosArrayGetSize(conn->cliMsgs) > 0) {
+  if (!transQueueEmpty(&conn->cliMsgs)) {
     cliSend(conn);
     return true;
-  } else {
-    return false;
   }
+  return false;
 }
 void cliHandleResp(SCliConn* conn) {
   SCliThrdObj* pThrd = conn->hostThrd;
@@ -195,6 +195,7 @@ void cliHandleResp(SCliConn* conn) {
   STransMsgHead* pHead = (STransMsgHead*)(conn->readBuf.buf);
   pHead->code = htonl(pHead->code);
   pHead->msgLen = htonl(pHead->msgLen);
+
   STransMsg transMsg = {0};
   transMsg.contLen = transContLenFromMsg(pHead->msgLen);
   transMsg.pCont = transContFromHead((char*)pHead);
@@ -204,11 +205,7 @@ void cliHandleResp(SCliConn* conn) {
 
   CONN_SHOULD_RELEASE(conn, pHead);
 
-  SCliMsg* pMsg = NULL;
-  if (taosArrayGetSize(conn->cliMsgs) > 0) {
-    pMsg = taosArrayGetP(conn->cliMsgs, 0);
-    taosArrayRemove(conn->cliMsgs, 0);
-  }
+  SCliMsg* pMsg = transQueuePop(&conn->cliMsgs);
 
   STransConnCtx* pCtx = pMsg ? pMsg->ctx : NULL;
   if (pMsg == NULL && !CONN_NO_PERSIST_BY_APP(conn)) {
@@ -264,7 +261,7 @@ _RETURN:
 }
 
 void cliHandleExcept(SCliConn* pConn) {
-  if (taosArrayGetSize(pConn->cliMsgs) == 0) {
+  if (transQueueEmpty(&pConn->cliMsgs)) {
     if (pConn->broken == true || CONN_NO_PERSIST_BY_APP(pConn)) {
       transUnrefCliHandle(pConn);
       return;
@@ -274,11 +271,7 @@ void cliHandleExcept(SCliConn* pConn) {
   STrans*      pTransInst = pThrd->pTransInst;
 
   do {
-    SCliMsg* pMsg = NULL;
-    if (taosArrayGetSize(pConn->cliMsgs) > 0) {
-      pMsg = taosArrayGetP(pConn->cliMsgs, 0);
-      taosArrayRemove(pConn->cliMsgs, 0);
-    }
+    SCliMsg* pMsg = transQueuePop(&pConn->cliMsgs);
 
     STransConnCtx* pCtx = pMsg ? pMsg->ctx : NULL;
 
@@ -303,7 +296,7 @@ void cliHandleExcept(SCliConn* pConn) {
     }
     destroyCmsg(pMsg);
     tTrace("%s cli conn %p start to destroy", CONN_GET_INST_LABEL(pConn), pConn);
-  } while (taosArrayGetSize(pConn->cliMsgs) > 0);
+  } while (!transQueueEmpty(&pConn->cliMsgs));
 
   transUnrefCliHandle(pConn);
 }
@@ -380,21 +373,20 @@ static void addConnToPool(void* pool, SCliConn* conn) {
   SCliThrdObj* thrd = conn->hostThrd;
   CONN_HANDLE_THREAD_QUIT(thrd);
 
-  char key[128] = {0};
+  STrans* pTransInst = ((SCliThrdObj*)conn->hostThrd)->pTransInst;
+  conn->expireTime = taosGetTimestampMs() + CONN_PERSIST_TIME(pTransInst->idleTime);
+  transCtxCleanup(&conn->ctx);
+  transQueueClear(&conn->cliMsgs);
+  conn->status = ConnNormal;
 
-  transCtxDestroy(&conn->ctx);
+  char key[128] = {0};
   tstrncpy(key, conn->ip, strlen(conn->ip));
   tstrncpy(key + strlen(key), (char*)(&conn->port), sizeof(conn->port));
   tTrace("cli conn %p added to conn pool, read buf cap: %d", conn, conn->readBuf.cap);
 
-  STrans* pTransInst = ((SCliThrdObj*)conn->hostThrd)->pTransInst;
-
-  conn->expireTime = taosGetTimestampMs() + CONN_PERSIST_TIME(pTransInst->idleTime);
   SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key));
-  conn->status = ConnNormal;
   // list already create before
   assert(plist != NULL);
-  taosArrayClear(conn->cliMsgs);
   QUEUE_PUSH(&plist->conn, &conn->conn);
   assert(!QUEUE_IS_EMPTY(&plist->conn));
 }
@@ -445,7 +437,8 @@ static SCliConn* cliCreateConn(SCliThrdObj* pThrd) {
 
   conn->writeReq.data = conn;
   conn->connReq.data = conn;
-  conn->cliMsgs = taosArrayInit(2, sizeof(void*));
+
+  transQueueInit(&conn->cliMsgs, NULL);
   QUEUE_INIT(&conn->conn);
   conn->hostThrd = pThrd;
   conn->status = ConnNormal;
@@ -465,18 +458,18 @@ static void cliDestroy(uv_handle_t* handle) {
   SCliConn* conn = handle->data;
   free(conn->ip);
   free(conn->stream);
-  transCtxDestroy(&conn->ctx);
-  taosArrayDestroy(conn->cliMsgs);
+  transCtxCleanup(&conn->ctx);
+  transQueueDestroy(&conn->cliMsgs);
   tTrace("%s cli conn %p destroy successfully", CONN_GET_INST_LABEL(conn), conn);
   free(conn);
 }
 static bool cliHandleNoResp(SCliConn* conn) {
-  bool    res = false;
-  SArray* msgs = conn->cliMsgs;
-  if (taosArrayGetSize(msgs) > 0) {
-    SCliMsg* pMsg = taosArrayGetP(msgs, 0);
+  bool res = false;
+  if (!transQueueEmpty(&conn->cliMsgs)) {
+    SCliMsg* pMsg = transQueueGet(&conn->cliMsgs);
     if (REQUEST_NO_RESP(&pMsg->msg)) {
-      taosArrayRemove(msgs, 0);
+      transQueuePop(&conn->cliMsgs);
+      // taosArrayRemove(msgs, 0);
       destroyCmsg(pMsg);
       res = true;
     }
@@ -509,8 +502,9 @@ static void cliSendCb(uv_write_t* req, int status) {
 void cliSend(SCliConn* pConn) {
   CONN_HANDLE_BROKEN(pConn);
 
-  assert(taosArrayGetSize(pConn->cliMsgs) > 0);
-  SCliMsg*       pCliMsg = taosArrayGetP(pConn->cliMsgs, 0);
+  // assert(taosArrayGetSize(pConn->cliMsgs) > 0);
+  assert(!transQueueEmpty(&pConn->cliMsgs));
+  SCliMsg*       pCliMsg = transQueueGet(&pConn->cliMsgs);
   STransConnCtx* pCtx = pCliMsg->ctx;
 
   SCliThrdObj* pThrd = pConn->hostThrd;
@@ -600,9 +594,8 @@ static void cliHandleRelease(SCliMsg* pMsg, SCliThrdObj* pThrd) {
 
   if (T_REF_VAL_GET(conn) == 2) {
     transUnrefCliHandle(conn);
-    taosArrayPush(conn->cliMsgs, &pMsg);
-    if (taosArrayGetSize(conn->cliMsgs) >= 2) {
-      return;  // send one by one
+    if (!transQueuePush(&conn->cliMsgs, pMsg)) {
+      return;
     }
     cliSend(conn);
   } else {
@@ -643,17 +636,14 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
     conn->hThrdIdx = pCtx->hThrdIdx;
 
     transCtxMerge(&conn->ctx, &pCtx->appCtx);
-    if (taosArrayGetSize(conn->cliMsgs) > 0) {
-      taosArrayPush(conn->cliMsgs, &pMsg);
+    if (!transQueuePush(&conn->cliMsgs, pMsg)) {
       return;
     }
-
-    taosArrayPush(conn->cliMsgs, &pMsg);
     transDestroyBuffer(&conn->readBuf);
     cliSend(conn);
   } else {
     conn = cliCreateConn(pThrd);
-    taosArrayPush(conn->cliMsgs, &pMsg);
+    transQueuePush(&conn->cliMsgs, pMsg);
 
     conn->hThrdIdx = pCtx->hThrdIdx;
     conn->ip = strdup(pMsg->ctx->ip);

source/libs/transport/src/transComm.c

@@ -228,7 +228,7 @@ void transCtxInit(STransCtx* ctx) {
   // init transCtx
   ctx->args = taosHashInit(2, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UINT), true, HASH_NO_LOCK);
 }
-void transCtxDestroy(STransCtx* ctx) {
+void transCtxCleanup(STransCtx* ctx) {
   if (ctx->args == NULL) {
     return;
   }
@@ -276,4 +276,49 @@ void* transCtxDumpVal(STransCtx* ctx, int32_t key) {
   return (void*)ret;
 }
 
+void transQueueInit(STransQueue* queue, void (*free)(void* arg)) {
+  queue->q = taosArrayInit(2, sizeof(void*));
+  queue->free = free;
+}
+bool transQueuePush(STransQueue* queue, void* arg) {
+  taosArrayPush(queue->q, &arg);
+  if (taosArrayGetSize(queue->q) > 1) {
+    return false;
+  }
+  return true;
+}
+void* transQueuePop(STransQueue* queue) {
+  if (taosArrayGetSize(queue->q) == 0) {
+    return NULL;
+  }
+  void* ptr = taosArrayGetP(queue->q, 0);
+  taosArrayRemove(queue->q, 0);
+  return ptr;
+}
+
+void* transQueueGet(STransQueue* queue) {
+  if (taosArrayGetSize(queue->q) == 0) {
+    return NULL;
+  }
+  void* ptr = taosArrayGetP(queue->q, 0);
+  return ptr;
+}
+bool transQueueEmpty(STransQueue* queue) {
+  //
+  return taosArrayGetSize(queue->q) == 0;
+}
+void transQueueClear(STransQueue* queue) {
+  if (queue->free != NULL) {
+    for (int i = 0; i < taosArrayGetSize(queue->q); i++) {
+      void* p = taosArrayGetP(queue->q, i);
+      queue->free(p);
+    }
+  }
+  taosArrayClear(queue->q);
+}
+void transQueueDestroy(STransQueue* queue) {
+  transQueueClear(queue);
+  taosArrayDestroy(queue->q);
+}
+
 #endif

source/libs/transport/src/transSrv.c

@@ -37,7 +37,7 @@ typedef struct SSrvConn {
   void*       pTransInst;  // rpc init
   void*       ahandle;     //
   void*       hostThrd;
-  SArray*     srvMsgs;
+  STransQueue srvMsgs;
 
   SSrvRegArg regArg;
   bool       broken;  // conn broken;
@@ -62,12 +62,12 @@ typedef struct SSrvMsg {
 } SSrvMsg;
 
 typedef struct SWorkThrdObj {
-  TdThread       thread;
-  uv_pipe_t*      pipe;
-  uv_os_fd_t      fd;
-  uv_loop_t*      loop;
-  SAsyncPool*     asyncPool;
-  queue           msg;
+  TdThread      thread;
+  uv_pipe_t*    pipe;
+  uv_os_fd_t    fd;
+  uv_loop_t*    loop;
+  SAsyncPool*   asyncPool;
+  queue         msg;
   TdThreadMutex msgMtx;
 
   queue conn;
@@ -76,7 +76,7 @@ typedef struct SWorkThrdObj {
 } SWorkThrdObj;
 
 typedef struct SServerObj {
-  TdThread  thread;
+  TdThread   thread;
   uv_tcp_t   server;
   uv_loop_t* loop;
 
@@ -106,8 +106,7 @@ static const char* notify = "a";
       srvMsg->msg = tmsg;                                          \
       srvMsg->type = Release;                                      \
       srvMsg->pConn = conn;                                        \
-      taosArrayPush(conn->srvMsgs, &srvMsg);                       \
-      if (taosArrayGetSize(conn->srvMsgs) > 1) {                   \
+      if (!transQueuePush(&conn->srvMsgs, srvMsg)) {               \
         return;                                                    \
       }                                                            \
       uvStartSendRespInternal(srvMsg);                             \
@@ -271,20 +270,16 @@ void uvOnSendCb(uv_write_t* req, int status) {
   transClearBuffer(&conn->readBuf);
   if (status == 0) {
     tTrace("server conn %p data already was written on stream", conn);
-    if (conn->srvMsgs != NULL) {
-      assert(taosArrayGetSize(conn->srvMsgs) >= 1);
-      SSrvMsg* msg = taosArrayGetP(conn->srvMsgs, 0);
-      tTrace("server conn %p sending msg size: %d", conn, (int)taosArrayGetSize(conn->srvMsgs));
-      taosArrayRemove(conn->srvMsgs, 0);
+    if (!transQueueEmpty(&conn->srvMsgs)) {
+      SSrvMsg* msg = transQueuePop(&conn->srvMsgs);
       if (msg->type == Release && conn->status != ConnNormal) {
         conn->status = ConnNormal;
         transUnrefSrvHandle(conn);
       }
       destroySmsg(msg);
       // send second data, just use for push
-      if (taosArrayGetSize(conn->srvMsgs) > 0) {
-        tTrace("resent server conn %p sending msg size: %d", conn, (int)taosArrayGetSize(conn->srvMsgs));
-        msg = (SSrvMsg*)taosArrayGetP(conn->srvMsgs, 0);
+      if (!transQueueEmpty(&conn->srvMsgs)) {
+        msg = (SSrvMsg*)transQueueGet(&conn->srvMsgs);
         if (msg->type == Register && conn->status == ConnAcquire) {
           conn->regArg.notifyCount = 0;
           conn->regArg.init = 1;
@@ -294,7 +289,7 @@ void uvOnSendCb(uv_write_t* req, int status) {
             (pTransInst->cfp)(pTransInst->parent, &(conn->regArg.msg), NULL);
             memset(&conn->regArg, 0, sizeof(conn->regArg));
           }
-          taosArrayRemove(conn->srvMsgs, 0);
+          transQueuePop(&conn->srvMsgs);
           free(msg);
         } else {
           uvStartSendRespInternal(msg);
@@ -373,10 +368,7 @@ static void uvStartSendResp(SSrvMsg* smsg) {
     transUnrefSrvHandle(pConn);
   }
 
-  taosArrayPush(pConn->srvMsgs, &smsg);
-  if (taosArrayGetSize(pConn->srvMsgs) > 1) {
-    tDebug("server conn %p send data to client %s:%d, local info: %s:%d", pConn, taosInetNtoa(pConn->addr.sin_addr),
-           ntohs(pConn->addr.sin_port), taosInetNtoa(pConn->locaddr.sin_addr), ntohs(pConn->locaddr.sin_port));
+  if (!transQueuePush(&pConn->srvMsgs, smsg)) {
     return;
   }
   uvStartSendRespInternal(smsg);
@@ -608,14 +600,15 @@ static SSrvConn* createConn(void* hThrd) {
   QUEUE_INIT(&pConn->queue);
 
   QUEUE_PUSH(&pThrd->conn, &pConn->queue);
-  pConn->srvMsgs = taosArrayInit(2, sizeof(void*));  //
-  tTrace("server conn %p created", pConn);
+
+  transQueueInit(&pConn->srvMsgs, NULL);
 
   memset(&pConn->regArg, 0, sizeof(pConn->regArg));
   pConn->broken = false;
   pConn->status = ConnNormal;
 
   transRefSrvHandle(pConn);
+  tTrace("server conn %p created", pConn);
   return pConn;
 }
 
@@ -625,11 +618,7 @@ static void destroyConn(SSrvConn* conn, bool clear) {
   }
   transDestroyBuffer(&conn->readBuf);
 
-  for (int i = 0; i < taosArrayGetSize(conn->srvMsgs); i++) {
-    SSrvMsg* msg = taosArrayGetP(conn->srvMsgs, i);
-    destroySmsg(msg);
-  }
-  conn->srvMsgs = taosArrayDestroy(conn->srvMsgs);
+  transQueueDestroy(&conn->srvMsgs);
   if (clear) {
     tTrace("server conn %p to be destroyed", conn);
     uv_shutdown_t* req = malloc(sizeof(uv_shutdown_t));
@@ -724,8 +713,7 @@ void uvHandleRelease(SSrvMsg* msg, SWorkThrdObj* thrd) {
   // release handle to rpc init
   SSrvConn* conn = msg->pConn;
   if (conn->status == ConnAcquire) {
-    taosArrayPush(conn->srvMsgs, &msg);
-    if (taosArrayGetSize(conn->srvMsgs) > 1) {
+    if (!transQueuePush(&conn->srvMsgs, msg)) {
       return;
     }
     uvStartSendRespInternal(msg);
@@ -744,8 +732,7 @@ void uvHandleRegister(SSrvMsg* msg, SWorkThrdObj* thrd) {
   SSrvConn* conn = msg->pConn;
   tDebug("server conn %p register brokenlink callback", conn);
   if (conn->status == ConnAcquire) {
-    if (taosArrayGetSize(conn->srvMsgs) > 0) {
-      taosArrayPush(conn->srvMsgs, &msg);
+    if (!transQueuePush(&conn->srvMsgs, msg)) {
       return;
     }
     conn->regArg.notifyCount = 0;

source/libs/transport/test/transportTests.cc

@@ -144,7 +144,7 @@ class TransCtxEnv : public ::testing::Test {
     // TODO
   }
   virtual void TearDown() {
-    transCtxDestroy(ctx);
+    transCtxCleanup(ctx);
     // formate
   }
   STransCtx *ctx;

source/os/src/osSystem.c

@@ -29,61 +29,6 @@
 struct termios oldtio;
 #endif
 
-int32_t taosSystem(const char *cmd, char *buf, int32_t bufSize) {
-#if defined(WINDOWS)
-  FILE *fp;
-  if (cmd == NULL) {
-    // printf("taosSystem cmd is NULL!");
-    return -1;
-  }
-
-  if ((fp = _popen(cmd, "r")) == NULL) {
-    // printf("popen cmd:%s error: %s", cmd, strerror(errno));
-    return -1;
-  } else {
-    while (fgets(buf, bufSize, fp)) {
-      // printf("popen result:%s", buf);
-    }
-
-    if (!_pclose(fp)) {
-      // printf("close popen file pointer fp error!");
-      return -1;
-    } else {
-      // printf("popen res is :%d", res);
-    }
-
-    return 0;
-  }
-#elif defined(_TD_DARWIN_64)
-  printf("no support funtion");
-  return -1;
-#else
-  FILE *fp;
-  int32_t res;
-  if (cmd == NULL) {
-    // printf("taosSystem cmd is NULL!");
-    return -1;
-  }
-
-  if ((fp = popen(cmd, "r")) == NULL) {
-    // printf("popen cmd:%s error: %s", cmd, strerror(errno));
-    return -1;
-  } else {
-    while (fgets(buf, bufSize, fp)) {
-      // printf("popen result:%s", buf);
-    }
-
-    if ((res = pclose(fp)) == -1) {
-      // printf("close popen file pointer fp error!");
-    } else {
-      // printf("popen res is :%d", res);
-    }
-
-    return res;
-  }
-#endif
-}
-
 void* taosLoadDll(const char* filename) {
 #if defined(WINDOWS)
   return NULL;
@@ -103,7 +48,7 @@ void* taosLoadDll(const char* filename) {
 }
 
 void* taosLoadSym(void* handle, char* name) {
-#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
+#if defined(WINDOWS)
   return NULL;
 #elif defined(_TD_DARWIN_64)
   return NULL;
@@ -123,7 +68,7 @@ void* taosLoadSym(void* handle, char* name) {
 }
 
 void  taosCloseDll(void* handle) {
-#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
+#if defined(WINDOWS)
   return;
 #elif defined(_TD_DARWIN_64)
   return;
@@ -135,7 +80,7 @@ void  taosCloseDll(void* handle) {
 }
 
 int taosSetConsoleEcho(bool on) {
-#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
+#if defined(WINDOWS)
   HANDLE hStdin = GetStdHandle(STD_INPUT_HANDLE);
   DWORD  mode = 0;
   GetConsoleMode(hStdin, &mode);
@@ -146,28 +91,6 @@ int taosSetConsoleEcho(bool on) {
   }
   SetConsoleMode(hStdin, mode);
 
-  return 0;
-#elif defined(_TD_DARWIN_64)
-#define ECHOFLAGS (ECHO | ECHOE | ECHOK | ECHONL)
-  int            err;
-  struct termios term;
-
-  if (tcgetattr(STDIN_FILENO, &term) == -1) {
-    perror("Cannot get the attribution of the terminal");
-    return -1;
-  }
-
-  if (on)
-    term.c_lflag |= ECHOFLAGS;
-  else
-    term.c_lflag &= ~ECHOFLAGS;
-
-  err = tcsetattr(STDIN_FILENO, TCSAFLUSH, &term);
-  if (err == -1 && err == EINTR) {
-    perror("Cannot set the attribution of the terminal");
-    return -1;
-  }
-
   return 0;
 #else
 #define ECHOFLAGS (ECHO | ECHOE | ECHOK | ECHONL)
@@ -186,7 +109,7 @@ int taosSetConsoleEcho(bool on) {
 
   err = tcsetattr(STDIN_FILENO, TCSAFLUSH, &term);
   if (err == -1 || err == EINTR) {
-    //printf("Cannot set the attribution of the terminal");
+    printf("Cannot set the attribution of the terminal");
     return -1;
   }
 
@@ -195,35 +118,8 @@ int taosSetConsoleEcho(bool on) {
 }
 
 void setTerminalMode() {
-#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
-
-#elif defined(_TD_DARWIN_64)
-  struct termios newtio;
-
-  /* if (atexit() != 0) { */
-  /*     fprintf(stderr, "Error register exit function!\n"); */
-  /*     exit(EXIT_FAILURE); */
-  /* } */
-
-  memcpy(&newtio, &oldtio, sizeof(oldtio));
-
-  // Set new terminal attributes.
-  newtio.c_iflag &= ~(IXON | IXOFF | ICRNL | INLCR | IGNCR | IMAXBEL | ISTRIP);
-  newtio.c_iflag |= IGNBRK;
-
-  // newtio.c_oflag &= ~(OPOST|ONLCR|OCRNL|ONLRET);
-  newtio.c_oflag |= OPOST;
-  newtio.c_oflag |= ONLCR;
-  newtio.c_oflag &= ~(OCRNL | ONLRET);
-
-  newtio.c_lflag &= ~(IEXTEN | ICANON | ECHO | ECHOE | ECHONL | ECHOCTL | ECHOPRT | ECHOKE | ISIG);
-  newtio.c_cc[VMIN] = 1;
-  newtio.c_cc[VTIME] = 0;
+#if defined(WINDOWS)
 
-  if (tcsetattr(0, TCSANOW, &newtio) != 0) {
-    fprintf(stderr, "Fail to set terminal properties!\n");
-    exit(EXIT_FAILURE);
-  }
 #else
   struct termios newtio;
 
@@ -256,19 +152,7 @@ void setTerminalMode() {
 
 int32_t getOldTerminalMode() {
 #if defined(WINDOWS)
-
-#elif defined(_TD_DARWIN_64)
-  /* Make sure stdin is a terminal. */
-  if (!isatty(STDIN_FILENO)) {
-    return -1;
-  }
-
-  // Get the parameter of current terminal
-  if (tcgetattr(0, &oldtio) != 0) {
-    return -1;
-  }
-
-  return 1;
+  
 #else
   /* Make sure stdin is a terminal. */
   if (!isatty(STDIN_FILENO)) {
@@ -285,13 +169,8 @@ int32_t getOldTerminalMode() {
 }
 
 void resetTerminalMode() {
-#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
+#if defined(WINDOWS)
 
-#elif defined(_TD_DARWIN_64)
-  if (tcsetattr(0, TCSANOW, &oldtio) != 0) {
-    fprintf(stderr, "Fail to reset the terminal properties!\n");
-    exit(EXIT_FAILURE);
-  }
 #else
   if (tcsetattr(0, TCSANOW, &oldtio) != 0) {
     fprintf(stderr, "Fail to reset the terminal properties!\n");

tests/script/sh/massiveTable/compileVersion.sh

@@ -68,7 +68,7 @@ gitPullBranchInfo $TDengineBrVer
 compileTDengineVersion
 
 taos_dir=${projectDir}/debug/tools/shell
-taosd_dir=${projectDir}/debug/source/dnode/mgmt/daemon
+taosd_dir=${projectDir}/debug/source/dnode/mgmt/main
 exec_process_dir=${projectDir}/debug/tests/test/c
 
 rm -f /usr/bin/taos

tests/script/tsim/insert/null.sim

+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+system sh/exec.sh -n dnode1 -s start
+sleep 50
+sql connect
+
+print =============== create database
+sql create database d0
+sql show databases
+if $rows != 1 then 
+  return -1
+endi
+
+print $data00 $data01 $data02
+
+sql use d0
+
+print =============== create super table, include column type for count/sum/min/max/first
+sql create table if not exists stb (ts timestamp, c1 int, c2 float, c3 double, c4 bigint) tags (t1 int unsigned)
+
+sql show stables
+if $rows != 1 then 
+  return -1
+endi
+
+print =============== create child table
+sql create table ct1 using stb tags(1000)
+sql create table ct2 using stb tags(2000)
+sql create table ct3 using stb tags(3000)
+
+sql show tables
+if $rows != 3 then 
+  return -1
+endi
+
+print =============== insert data, include NULL
+sql insert into ct1 values (now+0s, 10, 2.0, 3.0, 90)(now+1s, NULL, NULL, NULL, NULL)(now+2s, NULL, 2.1, 3.1, 91)(now+3s, 11, NULL, 3.2, 92)(now+4s, 12, 2.2, NULL, 93)(now+5s, 13, 2.3, 3.3, NULL)
+sql insert into ct1 values (now+6s, NULL, 2.4, 3.4, 94)
+sql insert into ct1 values (now+7s, 14, NULL, 3.5, 95)
+sql insert into ct1 values (now+8s, 15, 2.5, NULL, 96)
+sql insert into ct1 values (now+9s, 16, 2.6, 3.6, NULL)
+sql insert into ct1 values (now+10s, NULL, NULL, NULL, NULL)
+sql insert into ct1 values (now+11s, -2147483648, 2.7, 3.7, 97)
+
+#===================================================================
+#===================================================================
+print =============== query data from child table
+sql select * from ct1
+print ===> select * from ct1
+print ===> rows: $rows
+print ===> rows0: $data00 $data01 $data02 $data03 $data04
+print ===> rows1: $data10 $data11 $data12 $data13 $data14
+print ===> rows2: $data20 $data21 $data22 $data23 $data24
+print ===> rows3: $data30 $data31 $data32 $data33 $data34
+print ===> rows4: $data40 $data41 $data42 $data43 $data44
+if $rows != 12 then
+  return -1
+endi
+if $data01 != 10 then 
+  return -1
+endi
+if $data02 != 2.00000 then 
+  return -1
+endi
+if $data03 != 3.000000000 then 
+  return -1
+endi
+#if $data41 != -14 then 
+#  return -1
+#endi
+#if $data42 != -2.40000 then 
+#  return -1
+#endi
+#if $data43 != -3.400000000 then 
+#  return -1
+#endi
+
+
+print =============== select count(*) from child table
+sql select count(*) from ct1
+print ===> select count(*) from ct1
+print ===> rows: $rows
+print ===> rows0: $data00 $data01 $data02 $data03 $data04
+if $rows != 1 then 
+  return -1
+endi
+
+print $data00 $data01 $data02
+if $data00 != 4 then 
+  return -1
+endi
+
+print =============== select count(column) from child table
+sql select count(ts), count(c1), count(c2), count(c3) from ct1
+print ===> select count(ts), count(c1), count(c2), count(c3) from ct1
+print ===> rows: $rows
+print ===> rows0: $data00 $data01 $data02 $data03 $data04
+
+if $data00 != 4 then 
+  return -1
+endi
+if $data01 != 4 then 
+  return -1
+endi
+if $data02 != 4 then 
+  return -1
+endi
+if $data03 != 4 then 
+  return -1
+endi
+
+#print =============== select first(*)/first(column) from child table
+#sql select first(*) from ct1
+#sql select first(ts), first(c1), first(c2), first(c3) from ct1
+
+print =============== select min(column) from child table
+sql select min(c1), min(c2), min(c3) from ct1
+print ===> select min(c1), min(c2), min(c3) from ct1
+print ===> rows: $rows
+print ===> rows0: $data00 $data01 $data02 $data03 $data04
+if $rows != 1 then 
+  return -1
+endi
+if $data00 != 10 then 
+  return -1
+endi
+if $data01 != 2.00000 then 
+  return -1
+endi
+if $data02 != 3.000000000 then 
+  return -1
+endi
+
+print =============== select max(column) from child table
+sql select max(c1), max(c2), max(c3) from ct1
+print ===> select max(c1), max(c2), max(c3) from ct1
+print ===> rows: $rows
+print ===> rows0: $data00 $data01 $data02 $data03 $data04
+if $rows != 1 then 
+  return -1
+endi
+if $data00 != 13 then 
+  return -1
+endi
+if $data01 != 2.30000 then 
+  return -1
+endi
+if $data02 != 3.300000000 then 
+  return -1
+endi
+
+print =============== select sum(column) from child table
+sql select sum(c1), sum(c2), sum(c3) from ct1
+print ===> select sum(c1), sum(c2), sum(c3) from ct1
+print ===> rows: $rows
+print ===> rows0: $data00 $data01 $data02 $data03 $data04
+if $rows != 1 then 
+  return -1
+endi
+if $data00 != 46 then 
+  return -1
+endi
+if $data01 != 8.599999905 then 
+  return -1
+endi
+if $data02 != 12.600000000 then 
+  return -1
+endi
+
+print =============== select column, from child table
+sql select c1, c2, c3 from ct1
+print ===> select c1, c2, c3 from ct1
+print ===> rows: $rows
+print ===> rows0: $data00 $data01 $data02 $data03 $data04
+#if $rows != 4 then 
+#  return -1
+#endi
+#if $data00 != 10 then 
+#  return -1
+#endi
+#if $data01 != 2.00000 then 
+#  return -1
+#endi
+#if $data02 != 3.000000000 then 
+#  return -1
+#endi
+#if $data10 != 11 then 
+#  return -1
+#endi
+#if $data11 != 2.10000 then 
+#  return -1
+#endi
+#if $data12 != 3.100000000 then 
+#  return -1
+#endi
+#if $data30 != 13 then 
+#  return -1
+#endi
+#if $data31 != 2.30000 then 
+#  return -1
+#endi
+#if $data32 != 3.300000000 then 
+#  return -1
+#endi
+#===================================================================
+#===================================================================
+
+
+return 
+
+#print =============== query data from stb
+#sql select * from stb
+#print ===> 
+#print ===> rows: $rows
+#print ===> rows0: $data00 $data01 $data02 $data03 $data04
+#if $rows != 4 then 
+#  return -1
+#endi
+#print =============== select count(*) from supter table
+#sql select count(*) from stb
+#if $rows != 1 then 
+#  return -1
+#endi
+#
+#print $data00 $data01 $data02
+#if $data00 != 8 then 
+#  return -1
+#endi
+#
+#print =============== select count(column) from supter table
+#sql select count(ts), count(c1), count(c2), count(c3) from stb
+#print $data00 $data01 $data02  $data03
+#if $data00 != 8 then 
+#  return -1
+#endi
+#if $data01 != 8 then 
+#  return -1
+#endi
+#if $data02 != 8 then 
+#  return -1
+#endi
+#if $data03 != 8 then 
+#  return -1
+#endi
+
+
+#===================================================================
+#===================================================================
+
+print =============== stop and restart taosd, then again do query above
+system sh/exec.sh -n dnode1 -s stop -x SIGINT
+system sh/exec.sh -n dnode1 -s start
+
+sleep 2000
+sql select * from ct1
+if $rows != 4 then         # after fix bug, modify 4 to 7
+  return -1
+endi
+if $data01 != 10 then 
+  return -1
+endi
+if $data02 != 2.00000 then 
+  return -1
+endi
+if $data03 != 3.000000000 then 
+  return -1
+endi
+#if $data41 != -14 then 
+#  return -1
+#endi
+#if $data42 != -2.40000 then 
+#  return -1
+#endi
+#if $data43 != -3.400000000 then 
+#  return -1
+#endi
+
+
+print =============== select count(*) from child table
+sql select count(*) from ct1
+if $rows != 1 then 
+  return -1
+endi
+
+print $data00 $data01 $data02
+if $data00 != 4 then 
+  return -1
+endi
+
+print =============== select count(column) from child table
+sql select count(ts), count(c1), count(c2), count(c3) from ct1
+print $data00 $data01 $data02  $data03
+if $data00 != 4 then 
+  return -1
+endi
+if $data01 != 4 then 
+  return -1
+endi
+if $data02 != 4 then 
+  return -1
+endi
+if $data03 != 4 then 
+  return -1
+endi
+
+#print =============== select first(*)/first(column) from child table
+#sql select first(*) from ct1
+#sql select first(ts), first(c1), first(c2), first(c3) from ct1
+
+print =============== select min(column) from child table
+sql select min(c1), min(c2), min(c3) from ct1
+print $data00 $data01 $data02  $data03
+if $rows != 1 then 
+  return -1
+endi
+if $data00 != 10 then 
+  return -1
+endi
+if $data01 != 2.00000 then 
+  return -1
+endi
+if $data02 != 3.000000000 then 
+  return -1
+endi
+
+print =============== select max(column) from child table
+sql select max(c1), max(c2), max(c3) from ct1
+print $data00 $data01 $data02  $data03
+if $rows != 1 then 
+  return -1
+endi
+if $data00 != 13 then 
+  return -1
+endi
+if $data01 != 2.30000 then 
+  return -1
+endi
+if $data02 != 3.300000000 then 
+  return -1
+endi
+
+print =============== select sum(column) from child table
+sql select sum(c1), sum(c2), sum(c3) from ct1
+print $data00 $data01 $data02  $data03
+if $rows != 1 then 
+  return -1
+endi
+if $data00 != 46 then 
+  return -1
+endi
+if $data01 != 8.599999905 then 
+  return -1
+endi
+if $data02 != 12.600000000 then 
+  return -1
+endi
+
+#system sh/exec.sh -n dnode1 -s stop -x SIGINT

tests/script/tsim/query/interval-offset.sim

@@ -31,60 +31,91 @@ if $rows != 2 then
   return -1
 endi 
 
-print =============== insert data into child table
+print =============== insert data into child table ct1 (s)
 sql insert into ct1 values ( '2022-01-01 01:01:01.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:01:06.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:01:10.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:01:16.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:01:20.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:01:26.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:01:30.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:01:36.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:01:40.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:01:46.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:01:50.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:01:56.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:02:00.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:02:06.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:02:10.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:02:16.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:02:20.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:02:26.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:02:30.000', 1 )
-sql insert into ct1 values ( '2022-01-01 01:02:36.000', 1 )
+sql insert into ct1 values ( '2022-01-01 01:01:06.000', 2 )
+sql insert into ct1 values ( '2022-01-01 01:01:10.000', 3 )
+sql insert into ct1 values ( '2022-01-01 01:01:16.000', 4 )
+sql insert into ct1 values ( '2022-01-01 01:01:20.000', 5 )
+sql insert into ct1 values ( '2022-01-01 01:01:26.000', 6 )
+sql insert into ct1 values ( '2022-01-01 01:01:30.000', 7 )
+sql insert into ct1 values ( '2022-01-01 01:01:36.000', 8 )
 
 sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(*) from ct1 interval(10s, 2s)
 print ===> select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(*) from ct1 interval(10s, 2s)
-print ===> $rows $data00 $data01 $data02 $data03 $data04
-if $rows != 10 then
+print ===> rows: $rows
+print ===> rows0: $data00 $data01 $data02 $data03 $data04
+print ===> rows1: $data10 $data11 $data12 $data13 $data14
+print ===> rows2: $data20 $data21 $data22 $data23 $data24
+print ===> rows3: $data30 $data31 $data32 $data33 $data34
+print ===> rows4: $data40 $data41 $data42 $data43 $data44
+if $rows != 5 then
+  return -1
+endi 
+if $data00 != 1 then
+  return -1
+endi 
+if $data40 != 1 then
+  return -1
+endi 
+
+sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(*) from ct1 interval(10s, 2s) sliding(10s)
+print ===> select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(*) from ct1 interval(10s, 2s) sliding(10s)
+print ===> rows: $rows
+print ===> rows0: $data00 $data01 $data02 $data03 $data04
+print ===> rows1: $data10 $data11 $data12 $data13 $data14
+print ===> rows2: $data20 $data21 $data22 $data23 $data24
+print ===> rows3: $data30 $data31 $data32 $data33 $data34
+print ===> rows4: $data40 $data41 $data42 $data43 $data44
+if $rows != 5 then
+  return -1
+endi 
+if $data00 != 1 then
+  return -1
+endi 
+if $data40 != 1 then
+  return -1
+endi 
+
+sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(*) from ct1 interval(10s, 2s) sliding(5s)
+print ===> select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(*) from ct1 interval(10s, 2s) sliding(5s)
+print ===> rows: $rows
+print ===> rows0: $data00 $data01 $data02 $data03 $data04
+print ===> rows1: $data10 $data11 $data12 $data13 $data14
+print ===> rows2: $data20 $data21 $data22 $data23 $data24
+print ===> rows3: $data30 $data31 $data32 $data33 $data34
+print ===> rows4: $data40 $data41 $data42 $data43 $data44
+print ===> rows5: $data50 $data51 $data52 $data53 $data54
+print ===> rows6: $data60 $data61 $data62 $data63 $data64
+print ===> rows7: $data70 $data71 $data72 $data73 $data74
+if $rows != 8 then
   return -1
 endi 
 if $data00 != 2 then
   return -1
 endi 
-if $data04 != 2 then
+if $data70 != 1 then
   return -1
 endi 
 
+print =============== insert data into child table ct2 (d)
 sql insert into ct2 values ( '2022-01-01 01:00:01.000', 1 )
-sql insert into ct2 values ( '2022-01-01 12:00:01.000', 2 )
-sql insert into ct2 values ( '2022-01-01 23:00:01.000', 3 )
-sql insert into ct2 values ( '2022-01-02 10:00:01.000', 1 )
-sql insert into ct2 values ( '2022-01-03 10:00:01.000', 1 )
-sql insert into ct2 values ( '2022-01-04 10:00:01.000', 1 )
-sql insert into ct2 values ( '2022-01-05 10:00:01.000', 1 )
-sql insert into ct2 values ( '2022-01-06 10:00:01.000', 1 )
-sql insert into ct2 values ( '2022-01-07 10:00:01.000', 1 )
-sql insert into ct2 values ( '2022-01-08 10:00:01.000', 1 )
-sql insert into ct2 values ( '2022-01-09 10:00:01.000', 1 )
-sql insert into ct2 values ( '2022-01-10 10:00:01.000', 1 )
+sql insert into ct2 values ( '2022-01-01 10:00:01.000', 2 )
+sql insert into ct2 values ( '2022-01-01 20:00:01.000', 3 )
+sql insert into ct2 values ( '2022-01-02 10:00:01.000', 4 )
+sql insert into ct2 values ( '2022-01-02 20:00:01.000', 5 )
+sql insert into ct2 values ( '2022-01-03 10:00:01.000', 6 )
+sql insert into ct2 values ( '2022-01-03 20:00:01.000', 7 )
 
 sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(*) from ct2 interval(1d, 2h)
-print ===> select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(*) from ct2 interval(1d, 2w)
+print ===> select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(*) from ct2 interval(1d, 2h)
+print ===> rows: $rows
 print ===> rows0: $data00 $data01 $data02 $data03 $data04
 print ===> rows1: $data10 $data11 $data12 $data13 $data14
 print ===> rows2: $data20 $data21 $data22 $data23 $data24
-if $rows != 11 then
+print ===> rows3: $data30 $data31 $data32 $data33 $data34
+print ===> rows4: $data40 $data41 $data42 $data43 $data44
+if $rows != 4 then
   return -1
 endi 
 if $data00 != 1 then
@@ -94,8 +125,37 @@ if $data10 != 2 then
   return -1
 endi
 
+sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(*) from ct2 interval(1d, 2h) sliding(12h)
+print ===> select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(*) from ct2 interval(1d, 2h) sliding(12h)
+print ===> rows: $rows
+print ===> rows0: $data00 $data01 $data02 $data03 $data04
+print ===> rows1: $data10 $data11 $data12 $data13 $data14
+print ===> rows2: $data20 $data21 $data22 $data23 $data24
+print ===> rows3: $data30 $data31 $data32 $data33 $data34
+print ===> rows4: $data40 $data41 $data42 $data43 $data44
+print ===> rows5: $data50 $data51 $data52 $data53 $data54
+print ===> rows6: $data60 $data61 $data62 $data63 $data64
+print ===> rows7: $data70 $data71 $data72 $data73 $data74
+if $rows != 7 then
+  return -1
+endi 
+if $data00 != 2 then
+  return -1
+endi 
+if $data60 != 1 then
+  return -1
+endi
+
 return
 
+
+
+
+
+
+
+
+
 sql select count(*) from car interval(1n, 10d) order by ts desc
 #        tdSql.checkData(0, 1, 1)
 #        tdSql.checkData(1, 1, 2)

tools/shell/src/shellLinux.c

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#define __USE_XOPEN
-#include "os.h"
-#include "tglobal.h"
-#include "shell.h"
-#include "shellCommand.h"
-#include "tbase64.h"
-#include "tlog.h"
-#include "version.h"
-
-#include <wordexp.h>
-#include <argp.h>
-#include <termio.h>
-
-#define OPT_ABORT 1 /* �Cabort */
-
-int indicator = 1;
-struct termios oldtio;
-
-void insertChar(Command *cmd, char *c, int size);
-const char *argp_program_version = version;
-const char *argp_program_bug_address = "<support@taosdata.com>";
-static char doc[] = "";
-static char args_doc[] = "";
-static struct argp_option options[] = {
-  {"host",       'h', "HOST",       0,                   "TDengine server FQDN to connect. The default host is localhost."},
-  {"password",   'p', 0,   0,                   "The password to use when connecting to the server."},
-  {"port",       'P', "PORT",       0,                   "The TCP/IP port number to use for the connection."},
-  {"user",       'u', "USER",       0,                   "The user name to use when connecting to the server."},
-  {"auth",       'A', "Auth",       0,                   "The auth string to use when connecting to the server."},
-  {"config-dir", 'c', "CONFIG_DIR", 0,                   "Configuration directory."},
-  {"dump-config", 'C', 0,           0,                   "Dump configuration."},
-  {"commands",   's', "COMMANDS",   0,                   "Commands to run without enter the shell."},
-  {"raw-time",   'r', 0,            0,                   "Output time as uint64_t."},
-  {"file",       'f', "FILE",       0,                   "Script to run without enter the shell."},
-  {"directory",  'D', "DIRECTORY",  0,                   "Use multi-thread to import all SQL files in the directory separately."},
-  {"thread",     'T', "THREADNUM",  0,                   "Number of threads when using multi-thread to import data."},
-  {"check",      'k', "CHECK",      0,                   "Check tables."},
-  {"database",   'd', "DATABASE",   0,                   "Database to use when connecting to the server."},
-  {"timezone",   'z', "TIMEZONE",   0,                   "Time zone of the shell, default is local."},
-  {"netrole",    'n', "NETROLE",    0,                   "Net role when network connectivity test, default is startup, options: client|server|rpc|startup|sync|speen|fqdn."},
-  {"pktlen",     'l', "PKTLEN",     0,                   "Packet length used for net test, default is 1000 bytes."},
-  {"pktnum",     'N', "PKTNUM",     0,                   "Packet numbers used for net test, default is 100."},
-  {"pkttype",    'S', "PKTTYPE",    0,                   "Packet type used for net test, default is TCP."},
-  {0}};
-
-static error_t parse_opt(int key, char *arg, struct argp_state *state) {
-  /* Get the input argument from argp_parse, which we
-  know is a pointer to our arguments structure. */
-  SShellArguments *arguments = state->input;
-  wordexp_t full_path;
-
-  switch (key) {
-    case 'h':
-      arguments->host = arg;
-      break;
-    case 'p':
-      break;
-    case 'P':
-      if (arg) {
-        arguments->port  = atoi(arg);
-      } else {
-        fprintf(stderr, "Invalid port\n");
-        return -1;
-      }
-
-      break;
-    case 'z':
-      arguments->timezone = arg;
-      break;
-    case 'u':
-      arguments->user = arg;
-      break;
-    case 'A':
-      arguments->auth = arg;
-      break;
-    case 'c':
-      if (wordexp(arg, &full_path, 0) != 0) {
-        fprintf(stderr, "Invalid path %s\n", arg);
-        return -1;
-      }
-      if (strlen(full_path.we_wordv[0]) >= TSDB_FILENAME_LEN) {
-        fprintf(stderr, "config file path: %s overflow max len %d\n", full_path.we_wordv[0], TSDB_FILENAME_LEN - 1);
-        wordfree(&full_path);
-        return -1;
-      }
-      tstrncpy(configDir, full_path.we_wordv[0], TSDB_FILENAME_LEN);
-      wordfree(&full_path);
-      break;
-    case 'C':
-      arguments->dump_config = true;
-      break;
-    case 's':
-      arguments->commands = arg;
-      break;
-    case 'r':
-      arguments->is_raw_time = true;
-      break;
-    case 'f':
-      if ((0 == strlen(arg)) || (wordexp(arg, &full_path, 0) != 0)) {
-        fprintf(stderr, "Invalid path %s\n", arg);
-        return -1;
-      }
-      tstrncpy(arguments->file, full_path.we_wordv[0], TSDB_FILENAME_LEN);
-      wordfree(&full_path);
-      break;
-    case 'D':
-      if (wordexp(arg, &full_path, 0) != 0) {
-        fprintf(stderr, "Invalid path %s\n", arg);
-        return -1;
-      }
-      tstrncpy(arguments->dir, full_path.we_wordv[0], TSDB_FILENAME_LEN);
-      wordfree(&full_path);
-      break;
-    case 'T':
-      if (arg) {
-        arguments->threadNum = atoi(arg);
-      } else {
-        fprintf(stderr, "Invalid number of threads\n");
-        return -1;
-      }
-      break;
-    case 'k':
-      arguments->check = atoi(arg);
-      break;
-    case 'd':
-      arguments->database = arg;
-      break;
-    case 'n':
-      arguments->netTestRole = arg;
-      break;
-    case 'l':
-      if (arg) {
-        arguments->pktLen = atoi(arg);
-      } else {
-        fprintf(stderr, "Invalid packet length\n");
-        return -1;
-      }
-      break;
-    case 'N':
-      if (arg) {
-        arguments->pktNum = atoi(arg);
-      } else {
-        fprintf(stderr, "Invalid packet number\n");
-        return -1;
-      }
-      break;
-    case 'S':
-      arguments->pktType = arg;
-      break;
-    case OPT_ABORT:
-      arguments->abort = 1;
-      break;
-    default:
-      return ARGP_ERR_UNKNOWN;
-  }
-  return 0;
-}
-
-/* Our argp parser. */
-static struct argp argp = {options, parse_opt, args_doc, doc};
-
-char      LINUXCLIENT_VERSION[] = "Welcome to the TDengine shell from %s, Client Version:%s\n"
-                             "Copyright (c) 2020 by TAOS Data, Inc. All rights reserved.\n\n";
-char g_password[SHELL_MAX_PASSWORD_LEN];
-
-static void parse_args(
-        int argc, char *argv[], SShellArguments *arguments) {
-    for (int i = 1; i < argc; i++) {
-        if ((strncmp(argv[i], "-p", 2) == 0)
-              || (strncmp(argv[i], "--password", 10) == 0)) {
-            printf(LINUXCLIENT_VERSION, tsOsName, taos_get_client_info());
-            if ((strlen(argv[i]) == 2)
-                  || (strncmp(argv[i], "--password", 10) == 0)) {
-                printf("Enter password: ");
-                taosSetConsoleEcho(false);
-                if (scanf("%20s", g_password) > 1) {
-                    fprintf(stderr, "password reading error\n");
-                }
-                taosSetConsoleEcho(true);
-                if (EOF == getchar()) {
-                    fprintf(stderr, "getchar() return EOF\n");
-                }
-            } else {
-                tstrncpy(g_password, (char *)(argv[i] + 2), SHELL_MAX_PASSWORD_LEN);
-                strcpy(argv[i], "-p");
-            }
-            arguments->password = g_password;
-            arguments->is_use_passwd = true;
-        }
-    }
-}
-
-void shellParseArgument(int argc, char *argv[], SShellArguments *arguments) {
-  static char verType[32] = {0};
-  sprintf(verType, "version: %s\n", version);
-
-  argp_program_version = verType;
-
-  if (argc > 1) {
-    parse_args(argc, argv, arguments);
-  }
-
-  argp_parse(&argp, argc, argv, 0, 0, arguments);
-  if (arguments->abort) {
-    #ifndef _ALPINE
-    #if 0
-      error(10, 0, "ABORTED");
-    #endif  
-    #else
-      abort();
-    #endif
-  }
-}
-
-int32_t shellReadCommand(TAOS *con, char *command) {
-  unsigned hist_counter = history.hend;
-  char utf8_array[10] = "\0";
-  Command cmd;
-  memset(&cmd, 0, sizeof(cmd));
-  cmd.buffer = (char *)calloc(1, MAX_COMMAND_SIZE);
-  cmd.command = (char *)calloc(1, MAX_COMMAND_SIZE);
-  showOnScreen(&cmd);
-
-  // Read input.
-  char c;
-  while (1) {
-    c = (char)getchar(); // getchar() return an 'int' value
-
-    if (c == EOF) {
-      return c;
-    }
-
-    if (c < 0) {  // For UTF-8
-      int count = countPrefixOnes(c);
-      utf8_array[0] = c;
-      for (int k = 1; k < count; k++) {
-        c = (char)getchar();
-        utf8_array[k] = c;
-      }
-      insertChar(&cmd, utf8_array, count);
-    } else if (c < '\033') {
-      // Ctrl keys.  TODO: Implement ctrl combinations
-      switch (c) {
-        case 1:  // ctrl A
-          positionCursorHome(&cmd);
-          break;
-        case 3:
-          printf("\n");
-          resetCommand(&cmd, "");
-          kill(0, SIGINT);
-          break;
-        case 4:  // EOF or Ctrl+D
-          printf("\n");
-          taos_close(con);
-          // write the history
-          write_history();
-          exitShell();
-          break;
-        case 5:  // ctrl E
-          positionCursorEnd(&cmd);
-          break;
-        case 8:
-          backspaceChar(&cmd);
-          break;
-        case '\n':
-        case '\r':
-          printf("\n");
-          if (isReadyGo(&cmd)) {
-            sprintf(command, "%s%s", cmd.buffer, cmd.command);
-            tfree(cmd.buffer);
-            tfree(cmd.command);
-            return 0;
-          } else {
-            updateBuffer(&cmd);
-          }
-          break;
-        case 11:  // Ctrl + K;
-          clearLineAfter(&cmd);
-          break;
-        case 12:  // Ctrl + L;
-          system("clear");
-          showOnScreen(&cmd);
-          break;
-        case 21:  // Ctrl + U;
-          clearLineBefore(&cmd);
-          break;
-      }
-    } else if (c == '\033') {
-      c = (char)getchar();
-      switch (c) {
-        case '[':
-          c = (char)getchar();
-          switch (c) {
-            case 'A':  // Up arrow
-              if (hist_counter != history.hstart) {
-                hist_counter = (hist_counter + MAX_HISTORY_SIZE - 1) % MAX_HISTORY_SIZE;
-                resetCommand(&cmd, (history.hist[hist_counter] == NULL) ? "" : history.hist[hist_counter]);
-              }
-              break;
-            case 'B':  // Down arrow
-              if (hist_counter != history.hend) {
-                int next_hist = (hist_counter + 1) % MAX_HISTORY_SIZE;
-
-                if (next_hist != history.hend) {
-                  resetCommand(&cmd, (history.hist[next_hist] == NULL) ? "" : history.hist[next_hist]);
-                } else {
-                  resetCommand(&cmd, "");
-                }
-                hist_counter = next_hist;
-              }
-              break;
-            case 'C':  // Right arrow
-              moveCursorRight(&cmd);
-              break;
-            case 'D':  // Left arrow
-              moveCursorLeft(&cmd);
-              break;
-            case '1':
-              if ((c = (char)getchar()) == '~') {
-                // Home key
-                positionCursorHome(&cmd);
-              }
-              break;
-            case '2':
-              if ((c = (char)getchar()) == '~') {
-                // Insert key
-              }
-              break;
-            case '3':
-              if ((c = (char)getchar()) == '~') {
-                // Delete key
-                deleteChar(&cmd);
-              }
-              break;
-            case '4':
-              if ((c = (char)getchar()) == '~') {
-                // End key
-                positionCursorEnd(&cmd);
-              }
-              break;
-            case '5':
-              if ((c = (char)getchar()) == '~') {
-                // Page up key
-              }
-              break;
-            case '6':
-              if ((c = (char)getchar()) == '~') {
-                // Page down key
-              }
-              break;
-            case 72:
-              // Home key
-              positionCursorHome(&cmd);
-              break;
-            case 70:
-              // End key
-              positionCursorEnd(&cmd);
-              break;
-          }
-          break;
-      }
-    } else if (c == 0x7f) {
-      // press delete key
-      backspaceChar(&cmd);
-    } else {
-      insertChar(&cmd, &c, 1);
-    }
-  }
-
-  return 0;
-}
-
-void *shellLoopQuery(void *arg) {
-  if (indicator) {
-    getOldTerminalMode();
-    indicator = 0;
-  }
-
-  TAOS *con = (TAOS *)arg;
-
-  setThreadName("shellLoopQuery");
-
-  taosThreadCleanupPush(cleanup_handler, NULL);
-
-  char *command = malloc(MAX_COMMAND_SIZE);
-  if (command == NULL){
-    uError("failed to malloc command");
-    return NULL;
-  }
-
-  int32_t err = 0;
-  
-  do {
-    // Read command from shell.
-    memset(command, 0, MAX_COMMAND_SIZE);
-    setTerminalMode();
-    err = shellReadCommand(con, command);
-    if (err) {
-      break;
-    }
-    resetTerminalMode();
-  } while (shellRunCommand(con, command) == 0);
-  
-  tfree(command);
-  exitShell();
-
-  taosThreadCleanupPop(1);
-  
-  return NULL;
-}
-
-void get_history_path(char *_history) { snprintf(_history, TSDB_FILENAME_LEN, "%s/%s", getenv("HOME"), HISTORY_FILE); }
-
-void clearScreen(int ecmd_pos, int cursor_pos) {
-  struct winsize w;
-  if (ioctl(0, TIOCGWINSZ, &w) < 0 || w.ws_col == 0 || w.ws_row == 0) {
-    //fprintf(stderr, "No stream device, and use default value(col 120, row 30)\n");
-    w.ws_col = 120;
-    w.ws_row = 30;
-  }
-
-  int cursor_x = cursor_pos / w.ws_col;
-  int cursor_y = cursor_pos % w.ws_col;
-  int command_x = ecmd_pos / w.ws_col;
-  positionCursor(cursor_y, LEFT);
-  positionCursor(command_x - cursor_x, DOWN);
-  fprintf(stdout, "\033[2K");
-  for (int i = 0; i < command_x; i++) {
-    positionCursor(1, UP);
-    fprintf(stdout, "\033[2K");
-  }
-  fflush(stdout);
-}
-
-void showOnScreen(Command *cmd) {
-  struct winsize w;
-  if (ioctl(0, TIOCGWINSZ, &w) < 0 || w.ws_col == 0 || w.ws_row == 0) {
-    //fprintf(stderr, "No stream device\n");
-    w.ws_col = 120;
-    w.ws_row = 30;
-  }
-
-  TdWchar wc;
-  int size = 0;
-
-  // Print out the command.
-  char *total_string = malloc(MAX_COMMAND_SIZE);
-  memset(total_string, '\0', MAX_COMMAND_SIZE);
-  if (strcmp(cmd->buffer, "") == 0) {
-    sprintf(total_string, "%s%s", PROMPT_HEADER, cmd->command);
-  } else {
-    sprintf(total_string, "%s%s", CONTINUE_PROMPT, cmd->command);
-  }
-
-  int remain_column = w.ws_col;
-  /* size = cmd->commandSize + prompt_size; */
-  for (char *str = total_string; size < cmd->commandSize + prompt_size;) {
-    int ret = taosMbToWchar(&wc, str, MB_CUR_MAX);
-    if (ret < 0) break;
-    size += ret;
-    /* assert(size >= 0); */
-    int width = taosWcharWidth(wc);
-    if (remain_column > width) {
-      printf("%lc", wc);
-      remain_column -= width;
-    } else {
-      if (remain_column == width) {
-        printf("%lc\n\r", wc);
-        remain_column = w.ws_col;
-      } else {
-        printf("\n\r%lc", wc);
-        remain_column = w.ws_col - width;
-      }
-    }
-
-    str = total_string + size;
-  }
-
-  free(total_string);
-  /* for (int i = 0; i < size; i++){ */
-  /*     char c = total_string[i]; */
-  /*     if (k % w.ws_col == 0) { */
-  /*         printf("%c\n\r", c); */
-  /*     } */
-  /*     else { */
-  /*         printf("%c", c); */
-  /*     } */
-  /*     k += 1; */
-  /* } */
-
-  // Position the cursor
-  int cursor_pos = cmd->screenOffset + prompt_size;
-  int ecmd_pos = cmd->endOffset + prompt_size;
-
-  int cursor_x = cursor_pos / w.ws_col;
-  int cursor_y = cursor_pos % w.ws_col;
-  // int cursor_y = cursor % w.ws_col;
-  int command_x = ecmd_pos / w.ws_col;
-  int command_y = ecmd_pos % w.ws_col;
-  // int command_y = (command.size() + prompt_size) % w.ws_col;
-  positionCursor(command_y, LEFT);
-  positionCursor(command_x, UP);
-  positionCursor(cursor_x, DOWN);
-  positionCursor(cursor_y, RIGHT);
-  fflush(stdout);
-}
-
-void cleanup_handler(void *arg) { resetTerminalMode(); }
-
-void exitShell() {
-  /*int32_t ret =*/ resetTerminalMode();
-  taos_cleanup();
-  exit(EXIT_SUCCESS);
-}

tools/shell/src/shellMain.c

@@ -13,13 +13,524 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#define __USE_XOPEN
 #include "os.h"
 #include "shell.h"
 #include "tglobal.h"
+#include "shellCommand.h"
+#include "tbase64.h"
+#include "tlog.h"
+#include "version.h"
+
+#include <wordexp.h>
+#include <argp.h>
+#include <termio.h>
+
+#define OPT_ABORT 1 /* abort */
+
+
+int indicator = 1;
+
+void insertChar(Command *cmd, char *c, int size);
+const char *argp_program_version = version;
+const char *argp_program_bug_address = "<support@taosdata.com>";
+static char doc[] = "";
+static char args_doc[] = "";
 
 TdThread pid;
 static tsem_t cancelSem;
 
+static struct argp_option options[] = {
+  {"host",       'h', "HOST",       0,                   "TDengine server FQDN to connect. The default host is localhost."},
+  {"password",   'p', 0,   0,                   "The password to use when connecting to the server."},
+  {"port",       'P', "PORT",       0,                   "The TCP/IP port number to use for the connection."},
+  {"user",       'u', "USER",       0,                   "The user name to use when connecting to the server."},
+  {"auth",       'A', "Auth",       0,                   "The auth string to use when connecting to the server."},
+  {"config-dir", 'c', "CONFIG_DIR", 0,                   "Configuration directory."},
+  {"dump-config", 'C', 0,           0,                   "Dump configuration."},
+  {"commands",   's', "COMMANDS",   0,                   "Commands to run without enter the shell."},
+  {"raw-time",   'r', 0,            0,                   "Output time as uint64_t."},
+  {"file",       'f', "FILE",       0,                   "Script to run without enter the shell."},
+  {"directory",  'D', "DIRECTORY",  0,                   "Use multi-thread to import all SQL files in the directory separately."},
+  {"thread",     'T', "THREADNUM",  0,                   "Number of threads when using multi-thread to import data."},
+  {"check",      'k', "CHECK",      0,                   "Check tables."},
+  {"database",   'd', "DATABASE",   0,                   "Database to use when connecting to the server."},
+  {"timezone",   'z', "TIMEZONE",   0,                   "Time zone of the shell, default is local."},
+  {"netrole",    'n', "NETROLE",    0,                   "Net role when network connectivity test, default is startup, options: client|server|rpc|startup|sync|speen|fqdn."},
+  {"pktlen",     'l', "PKTLEN",     0,                   "Packet length used for net test, default is 1000 bytes."},
+  {"pktnum",     'N', "PKTNUM",     0,                   "Packet numbers used for net test, default is 100."},
+  {"pkttype",    'S', "PKTTYPE",    0,                   "Packet type used for net test, default is TCP."},
+  {0}};
+
+static error_t parse_opt(int key, char *arg, struct argp_state *state) {
+  /* Get the input argument from argp_parse, which we
+  know is a pointer to our arguments structure. */
+  SShellArguments *arguments = state->input;
+  wordexp_t full_path;
+
+  switch (key) {
+    case 'h':
+      arguments->host = arg;
+      break;
+    case 'p':
+      break;
+    case 'P':
+      if (arg) {
+        arguments->port  = atoi(arg);
+      } else {
+        fprintf(stderr, "Invalid port\n");
+        return -1;
+      }
+
+      break;
+    case 'z':
+      arguments->timezone = arg;
+      break;
+    case 'u':
+      arguments->user = arg;
+      break;
+    case 'A':
+      arguments->auth = arg;
+      break;
+    case 'c':
+      if (wordexp(arg, &full_path, 0) != 0) {
+        fprintf(stderr, "Invalid path %s\n", arg);
+        return -1;
+      }
+      if (strlen(full_path.we_wordv[0]) >= TSDB_FILENAME_LEN) {
+        fprintf(stderr, "config file path: %s overflow max len %d\n", full_path.we_wordv[0], TSDB_FILENAME_LEN - 1);
+        wordfree(&full_path);
+        return -1;
+      }
+      tstrncpy(configDir, full_path.we_wordv[0], TSDB_FILENAME_LEN);
+      wordfree(&full_path);
+      break;
+    case 'C':
+      arguments->dump_config = true;
+      break;
+    case 's':
+      arguments->commands = arg;
+      break;
+    case 'r':
+      arguments->is_raw_time = true;
+      break;
+    case 'f':
+      if ((0 == strlen(arg)) || (wordexp(arg, &full_path, 0) != 0)) {
+        fprintf(stderr, "Invalid path %s\n", arg);
+        return -1;
+      }
+      tstrncpy(arguments->file, full_path.we_wordv[0], TSDB_FILENAME_LEN);
+      wordfree(&full_path);
+      break;
+    case 'D':
+      if (wordexp(arg, &full_path, 0) != 0) {
+        fprintf(stderr, "Invalid path %s\n", arg);
+        return -1;
+      }
+      tstrncpy(arguments->dir, full_path.we_wordv[0], TSDB_FILENAME_LEN);
+      wordfree(&full_path);
+      break;
+    case 'T':
+      if (arg) {
+        arguments->threadNum = atoi(arg);
+      } else {
+        fprintf(stderr, "Invalid number of threads\n");
+        return -1;
+      }
+      break;
+    case 'k':
+      arguments->check = atoi(arg);
+      break;
+    case 'd':
+      arguments->database = arg;
+      break;
+    case 'n':
+      arguments->netTestRole = arg;
+      break;
+    case 'l':
+      if (arg) {
+        arguments->pktLen = atoi(arg);
+      } else {
+        fprintf(stderr, "Invalid packet length\n");
+        return -1;
+      }
+      break;
+    case 'N':
+      if (arg) {
+        arguments->pktNum = atoi(arg);
+      } else {
+        fprintf(stderr, "Invalid packet number\n");
+        return -1;
+      }
+      break;
+    case 'S':
+      arguments->pktType = arg;
+      break;
+    case OPT_ABORT:
+      arguments->abort = 1;
+      break;
+    default:
+      return ARGP_ERR_UNKNOWN;
+  }
+  return 0;
+}
+
+/* Our argp parser. */
+static struct argp argp = {options, parse_opt, args_doc, doc};
+
+char      LINUXCLIENT_VERSION[] = "Welcome to the TDengine shell from %s, Client Version:%s\n"
+                             "Copyright (c) 2020 by TAOS Data, Inc. All rights reserved.\n\n";
+char g_password[SHELL_MAX_PASSWORD_LEN];
+
+static void parse_args(
+        int argc, char *argv[], SShellArguments *arguments) {
+    for (int i = 1; i < argc; i++) {
+        if ((strncmp(argv[i], "-p", 2) == 0)
+              || (strncmp(argv[i], "--password", 10) == 0)) {
+            printf(LINUXCLIENT_VERSION, tsOsName, taos_get_client_info());
+            if ((strlen(argv[i]) == 2)
+                  || (strncmp(argv[i], "--password", 10) == 0)) {
+                printf("Enter password: ");
+                taosSetConsoleEcho(false);
+                if (scanf("%20s", g_password) > 1) {
+                    fprintf(stderr, "password reading error\n");
+                }
+                taosSetConsoleEcho(true);
+                if (EOF == getchar()) {
+                    fprintf(stderr, "getchar() return EOF\n");
+                }
+            } else {
+                tstrncpy(g_password, (char *)(argv[i] + 2), SHELL_MAX_PASSWORD_LEN);
+                strcpy(argv[i], "-p");
+            }
+            arguments->password = g_password;
+            arguments->is_use_passwd = true;
+        }
+    }
+}
+
+void shellParseArgument(int argc, char *argv[], SShellArguments *arguments) {
+  static char verType[32] = {0};
+  sprintf(verType, "version: %s\n", version);
+
+  argp_program_version = verType;
+
+  if (argc > 1) {
+    parse_args(argc, argv, arguments);
+  }
+
+  argp_parse(&argp, argc, argv, 0, 0, arguments);
+  if (arguments->abort) {
+    #ifndef _ALPINE
+    #if 0
+      error(10, 0, "ABORTED");
+    #endif  
+    #else
+      abort();
+    #endif
+  }
+}
+
+int32_t shellReadCommand(TAOS *con, char *command) {
+  unsigned hist_counter = history.hend;
+  char utf8_array[10] = "\0";
+  Command cmd;
+  memset(&cmd, 0, sizeof(cmd));
+  cmd.buffer = (char *)calloc(1, MAX_COMMAND_SIZE);
+  cmd.command = (char *)calloc(1, MAX_COMMAND_SIZE);
+  showOnScreen(&cmd);
+
+  // Read input.
+  char c;
+  while (1) {
+    c = (char)getchar(); // getchar() return an 'int' value
+
+    if (c == EOF) {
+      return c;
+    }
+
+    if (c < 0) {  // For UTF-8
+      int count = countPrefixOnes(c);
+      utf8_array[0] = c;
+      for (int k = 1; k < count; k++) {
+        c = (char)getchar();
+        utf8_array[k] = c;
+      }
+      insertChar(&cmd, utf8_array, count);
+    } else if (c < '\033') {
+      // Ctrl keys.  TODO: Implement ctrl combinations
+      switch (c) {
+        case 1:  // ctrl A
+          positionCursorHome(&cmd);
+          break;
+        case 3:
+          printf("\n");
+          resetCommand(&cmd, "");
+          kill(0, SIGINT);
+          break;
+        case 4:  // EOF or Ctrl+D
+          printf("\n");
+          taos_close(con);
+          // write the history
+          write_history();
+          exitShell();
+          break;
+        case 5:  // ctrl E
+          positionCursorEnd(&cmd);
+          break;
+        case 8:
+          backspaceChar(&cmd);
+          break;
+        case '\n':
+        case '\r':
+          printf("\n");
+          if (isReadyGo(&cmd)) {
+            sprintf(command, "%s%s", cmd.buffer, cmd.command);
+            tfree(cmd.buffer);
+            tfree(cmd.command);
+            return 0;
+          } else {
+            updateBuffer(&cmd);
+          }
+          break;
+        case 11:  // Ctrl + K;
+          clearLineAfter(&cmd);
+          break;
+        case 12:  // Ctrl + L;
+          system("clear");
+          showOnScreen(&cmd);
+          break;
+        case 21:  // Ctrl + U;
+          clearLineBefore(&cmd);
+          break;
+      }
+    } else if (c == '\033') {
+      c = (char)getchar();
+      switch (c) {
+        case '[':
+          c = (char)getchar();
+          switch (c) {
+            case 'A':  // Up arrow
+              if (hist_counter != history.hstart) {
+                hist_counter = (hist_counter + MAX_HISTORY_SIZE - 1) % MAX_HISTORY_SIZE;
+                resetCommand(&cmd, (history.hist[hist_counter] == NULL) ? "" : history.hist[hist_counter]);
+              }
+              break;
+            case 'B':  // Down arrow
+              if (hist_counter != history.hend) {
+                int next_hist = (hist_counter + 1) % MAX_HISTORY_SIZE;
+
+                if (next_hist != history.hend) {
+                  resetCommand(&cmd, (history.hist[next_hist] == NULL) ? "" : history.hist[next_hist]);
+                } else {
+                  resetCommand(&cmd, "");
+                }
+                hist_counter = next_hist;
+              }
+              break;
+            case 'C':  // Right arrow
+              moveCursorRight(&cmd);
+              break;
+            case 'D':  // Left arrow
+              moveCursorLeft(&cmd);
+              break;
+            case '1':
+              if ((c = (char)getchar()) == '~') {
+                // Home key
+                positionCursorHome(&cmd);
+              }
+              break;
+            case '2':
+              if ((c = (char)getchar()) == '~') {
+                // Insert key
+              }
+              break;
+            case '3':
+              if ((c = (char)getchar()) == '~') {
+                // Delete key
+                deleteChar(&cmd);
+              }
+              break;
+            case '4':
+              if ((c = (char)getchar()) == '~') {
+                // End key
+                positionCursorEnd(&cmd);
+              }
+              break;
+            case '5':
+              if ((c = (char)getchar()) == '~') {
+                // Page up key
+              }
+              break;
+            case '6':
+              if ((c = (char)getchar()) == '~') {
+                // Page down key
+              }
+              break;
+            case 72:
+              // Home key
+              positionCursorHome(&cmd);
+              break;
+            case 70:
+              // End key
+              positionCursorEnd(&cmd);
+              break;
+          }
+          break;
+      }
+    } else if (c == 0x7f) {
+      // press delete key
+      backspaceChar(&cmd);
+    } else {
+      insertChar(&cmd, &c, 1);
+    }
+  }
+
+  return 0;
+}
+
+void *shellLoopQuery(void *arg) {
+  if (indicator) {
+    getOldTerminalMode();
+    indicator = 0;
+  }
+
+  TAOS *con = (TAOS *)arg;
+
+  setThreadName("shellLoopQuery");
+
+  taosThreadCleanupPush(cleanup_handler, NULL);
+
+  char *command = malloc(MAX_COMMAND_SIZE);
+  if (command == NULL){
+    uError("failed to malloc command");
+    return NULL;
+  }
+
+  int32_t err = 0;
+  
+  do {
+    // Read command from shell.
+    memset(command, 0, MAX_COMMAND_SIZE);
+    setTerminalMode();
+    err = shellReadCommand(con, command);
+    if (err) {
+      break;
+    }
+    resetTerminalMode();
+  } while (shellRunCommand(con, command) == 0);
+  
+  tfree(command);
+  exitShell();
+
+  taosThreadCleanupPop(1);
+  
+  return NULL;
+}
+
+void get_history_path(char *_history) { snprintf(_history, TSDB_FILENAME_LEN, "%s/%s", getenv("HOME"), HISTORY_FILE); }
+
+void clearScreen(int ecmd_pos, int cursor_pos) {
+  struct winsize w;
+  if (ioctl(0, TIOCGWINSZ, &w) < 0 || w.ws_col == 0 || w.ws_row == 0) {
+    //fprintf(stderr, "No stream device, and use default value(col 120, row 30)\n");
+    w.ws_col = 120;
+    w.ws_row = 30;
+  }
+
+  int cursor_x = cursor_pos / w.ws_col;
+  int cursor_y = cursor_pos % w.ws_col;
+  int command_x = ecmd_pos / w.ws_col;
+  positionCursor(cursor_y, LEFT);
+  positionCursor(command_x - cursor_x, DOWN);
+  fprintf(stdout, "\033[2K");
+  for (int i = 0; i < command_x; i++) {
+    positionCursor(1, UP);
+    fprintf(stdout, "\033[2K");
+  }
+  fflush(stdout);
+}
+
+void showOnScreen(Command *cmd) {
+  struct winsize w;
+  if (ioctl(0, TIOCGWINSZ, &w) < 0 || w.ws_col == 0 || w.ws_row == 0) {
+    //fprintf(stderr, "No stream device\n");
+    w.ws_col = 120;
+    w.ws_row = 30;
+  }
+
+  TdWchar wc;
+  int size = 0;
+
+  // Print out the command.
+  char *total_string = malloc(MAX_COMMAND_SIZE);
+  memset(total_string, '\0', MAX_COMMAND_SIZE);
+  if (strcmp(cmd->buffer, "") == 0) {
+    sprintf(total_string, "%s%s", PROMPT_HEADER, cmd->command);
+  } else {
+    sprintf(total_string, "%s%s", CONTINUE_PROMPT, cmd->command);
+  }
+
+  int remain_column = w.ws_col;
+  /* size = cmd->commandSize + prompt_size; */
+  for (char *str = total_string; size < cmd->commandSize + prompt_size;) {
+    int ret = taosMbToWchar(&wc, str, MB_CUR_MAX);
+    if (ret < 0) break;
+    size += ret;
+    /* assert(size >= 0); */
+    int width = taosWcharWidth(wc);
+    if (remain_column > width) {
+      printf("%lc", wc);
+      remain_column -= width;
+    } else {
+      if (remain_column == width) {
+        printf("%lc\n\r", wc);
+        remain_column = w.ws_col;
+      } else {
+        printf("\n\r%lc", wc);
+        remain_column = w.ws_col - width;
+      }
+    }
+
+    str = total_string + size;
+  }
+
+  free(total_string);
+  /* for (int i = 0; i < size; i++){ */
+  /*     char c = total_string[i]; */
+  /*     if (k % w.ws_col == 0) { */
+  /*         printf("%c\n\r", c); */
+  /*     } */
+  /*     else { */
+  /*         printf("%c", c); */
+  /*     } */
+  /*     k += 1; */
+  /* } */
+
+  // Position the cursor
+  int cursor_pos = cmd->screenOffset + prompt_size;
+  int ecmd_pos = cmd->endOffset + prompt_size;
+
+  int cursor_x = cursor_pos / w.ws_col;
+  int cursor_y = cursor_pos % w.ws_col;
+  // int cursor_y = cursor % w.ws_col;
+  int command_x = ecmd_pos / w.ws_col;
+  int command_y = ecmd_pos % w.ws_col;
+  // int command_y = (command.size() + prompt_size) % w.ws_col;
+  positionCursor(command_y, LEFT);
+  positionCursor(command_x, UP);
+  positionCursor(cursor_x, DOWN);
+  positionCursor(cursor_y, RIGHT);
+  fflush(stdout);
+}
+
+void cleanup_handler(void *arg) { resetTerminalMode(); }
+
+void exitShell() {
+  /*int32_t ret =*/ resetTerminalMode();
+  taos_cleanup();
+  exit(EXIT_SUCCESS);
+}
 void shellQueryInterruptHandler(int32_t signum, void *sigInfo, void *context) {
   tsem_post(&cancelSem);
 }