From c9279cdc115636da78bf041323ad464dbaaf1070 Mon Sep 17 00:00:00 2001
From: yihaoDeng <yhdeng@taosdata.com>
Date: Tue, 6 Apr 2021 17:04:31 +0800
Subject: [PATCH 01/10] refactor rpc

---
 src/rpc/src/rpcTcp.c | 155 +++++++++++++++++++++++++++++++++++++++----
 1 file changed, 143 insertions(+), 12 deletions(-)

diff --git a/src/rpc/src/rpcTcp.c b/src/rpc/src/rpcTcp.c
index 3162ab2e4c..286ed223c7 100644
--- a/src/rpc/src/rpcTcp.c
+++ b/src/rpc/src/rpcTcp.c
@@ -21,6 +21,13 @@
 #include "rpcLog.h"
 #include "rpcHead.h"
 #include "rpcTcp.h"
+#include "tlist.h"
+
+typedef struct SConnItem {
+  SOCKET         fd; 
+  uint32_t       ip;
+  uint16_t       port;
+} SConnItem;
 
 typedef struct SFdObj {
   void              *signature;
@@ -38,6 +45,12 @@ typedef struct SThreadObj {
   pthread_t       thread;
   SFdObj *        pHead;
   pthread_mutex_t mutex;
+  // receive the notify from dispatch thread 
+
+  int             notifyReceiveFd;
+  int             notifySendFd;
+  SList           *connQueue;  
+
   uint32_t        ip;
   bool            stop;
   EpollFd         pollFd;
@@ -69,6 +82,7 @@ typedef struct {
 } SServerObj;
 
 static void   *taosProcessTcpData(void *param);
+static void   *taosProcessServerTcpData(void *param);
 static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, SOCKET fd);
 static void    taosFreeFdObj(SFdObj *pFdObj);
 static void    taosReportBrokenLink(SFdObj *pFdObj);
@@ -124,6 +138,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
     tstrncpy(pThreadObj->label, label, sizeof(pThreadObj->label));
     pThreadObj->shandle = shandle;
     pThreadObj->stop = false;
+    pThreadObj->connQueue = tdListNew(sizeof(SConnItem));
   }
 
   // initialize mutex, thread, fd which may fail
@@ -142,7 +157,25 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
       break;
     }
 
-    code = pthread_create(&(pThreadObj->thread), &thattr, taosProcessTcpData, (void *)(pThreadObj));
+    int fds[2];
+    if (pipe(fds)) {
+      tError("%s failed to create pipe", label);
+      code = -1;
+      break;
+    }
+
+    pThreadObj->notifyReceiveFd = fds[0];
+    pThreadObj->notifySendFd    = fds[1];
+    struct epoll_event event;
+    event.events  = EPOLLIN | EPOLLRDHUP; 
+    event.data.fd = pThreadObj->notifyReceiveFd;
+    if (epoll_ctl(pThreadObj->pollFd, EPOLL_CTL_ADD, pThreadObj->notifyReceiveFd , &event) < 0) {
+      tError("%s failed to create pipe", label);
+      code = -1;
+      break;
+    }
+
+    code = pthread_create(&(pThreadObj->thread), &thattr, taosProcessServerTcpData, (void *)(pThreadObj));
     if (code != 0) {
       tError("%s failed to create TCP process data thread(%s)", label, strerror(errno));
       break;
@@ -275,17 +308,12 @@ static void *taosAcceptTcpConnection(void *arg) {
     // pick up the thread to handle this connection
     pThreadObj = pServerObj->pThreadObj[threadId];
 
-    SFdObj *pFdObj = taosMallocFdObj(pThreadObj, connFd);
-    if (pFdObj) {
-      pFdObj->ip = caddr.sin_addr.s_addr;
-      pFdObj->port = htons(caddr.sin_port);
-      tDebug("%s new TCP connection from %s:%hu, fd:%d FD:%p numOfFds:%d", pServerObj->label,
-              taosInetNtoa(caddr.sin_addr), pFdObj->port, connFd, pFdObj, pThreadObj->numOfFds);
-    } else {
-      taosCloseSocket(connFd);
-      tError("%s failed to malloc FdObj(%s) for connection from:%s:%hu", pServerObj->label, strerror(errno),
-             taosInetNtoa(caddr.sin_addr), htons(caddr.sin_port));
-    }
+    pthread_mutex_lock(&(pThreadObj->mutex));
+    SConnItem item = {.fd = connFd, .ip = caddr.sin_addr.s_addr, .port = htons(caddr.sin_port)};     
+    tdListAppend(pThreadObj->connQueue, &item);     
+    pthread_mutex_unlock(&(pThreadObj->mutex));
+    
+    write(pThreadObj->notifySendFd, "", 1);
 
     // pick up next thread for next connection
     threadId++;
@@ -591,6 +619,109 @@ static void *taosProcessTcpData(void *param) {
   return NULL;
 }
 
+static void *taosProcessServerTcpData(void *param) {
+  SThreadObj        *pThreadObj = param;
+  SFdObj            *pFdObj;
+  struct epoll_event events[maxEvents];
+  SRecvInfo          recvInfo;
+
+  char               bb[1];
+#ifdef __APPLE__
+  taos_block_sigalrm();
+#endif // __APPLE__
+  while (1) {
+    int fdNum = epoll_wait(pThreadObj->pollFd, events, maxEvents, TAOS_EPOLL_WAIT_TIME);
+    if (pThreadObj->stop) {
+      tDebug("%s TCP thread get stop event, exiting...", pThreadObj->label);
+      break;
+    }
+    if (fdNum < 0) continue;
+
+    for (int i = 0; i < fdNum; ++i) {
+      if (events[i].data.fd == pThreadObj->notifyReceiveFd) {
+        if (events[i].events & EPOLLIN) {
+          read(pThreadObj->notifyReceiveFd, bb, 1);
+
+          pthread_mutex_lock(&(pThreadObj->mutex));
+          SListNode *head = tdListPopHead(pThreadObj->connQueue);  
+          pthread_mutex_unlock(&(pThreadObj->mutex));
+
+          SConnItem item = {0};  
+          tdListNodeGetData(pThreadObj->connQueue, head, &item);   
+          tfree(head);
+          
+          // register fd on epoll 
+          SFdObj *pFdObj = taosMallocFdObj(pThreadObj, item.fd);
+          if (pFdObj) {
+            pFdObj->ip   = item.ip;
+            pFdObj->port = item.port; 
+            tDebug("%s new TCP connection from %u:%hu, fd:%d FD:%p numOfFds:%d", pThreadObj->label,
+              pFdObj->ip, pFdObj->port, item.fd, pFdObj, pThreadObj->numOfFds);
+          } else {
+            taosCloseSocket(item.fd);
+            tError("%s failed to malloc FdObj(%s) for connection from:%u:%hu", pThreadObj->label, strerror(errno),
+               pFdObj->ip, pFdObj->port);
+          }
+        } 
+        continue; 
+      }
+      pFdObj = events[i].data.ptr;
+
+      if (events[i].events & EPOLLERR) {
+        tDebug("%s %p FD:%p epoll errors", pThreadObj->label, pFdObj->thandle, pFdObj);
+        taosReportBrokenLink(pFdObj);
+        continue;
+      }
+
+      if (events[i].events & EPOLLRDHUP) {
+        tDebug("%s %p FD:%p RD hang up", pThreadObj->label, pFdObj->thandle, pFdObj);
+        taosReportBrokenLink(pFdObj);
+        continue;
+      }
+
+      if (events[i].events & EPOLLHUP) {
+        tDebug("%s %p FD:%p hang up", pThreadObj->label, pFdObj->thandle, pFdObj);
+        taosReportBrokenLink(pFdObj);
+        continue;
+      }
+
+      if (taosReadTcpData(pFdObj, &recvInfo) < 0) {
+        shutdown(pFdObj->fd, SHUT_WR);
+        continue;
+      }
+
+      pFdObj->thandle = (*(pThreadObj->processData))(&recvInfo);
+      if (pFdObj->thandle == NULL) taosFreeFdObj(pFdObj);
+    }
+
+    if (pThreadObj->stop) break;
+  }
+
+  if (pThreadObj->connQueue) {
+    pThreadObj->connQueue = tdListFree(pThreadObj->connQueue); 
+  }
+  // close pipe
+  close(pThreadObj->notifySendFd);
+  close(pThreadObj->notifyReceiveFd);
+
+  if (pThreadObj->pollFd >=0) {
+    EpollClose(pThreadObj->pollFd);
+    pThreadObj->pollFd = -1;
+  }
+
+  while (pThreadObj->pHead) {
+    SFdObj *pFdObj = pThreadObj->pHead;
+    pThreadObj->pHead = pFdObj->next;
+    taosReportBrokenLink(pFdObj);
+  }
+
+  pthread_mutex_destroy(&(pThreadObj->mutex));
+  tDebug("%s TCP thread exits ...", pThreadObj->label);
+  tfree(pThreadObj);
+
+  return NULL;
+}
+
 static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, SOCKET fd) {
   struct epoll_event event;
 

From 57e77b5f416eba4164196d6c8438fd6af119d6ab Mon Sep 17 00:00:00 2001
From: yihaoDeng <yhdeng@taosdata.com>
Date: Thu, 8 Apr 2021 12:10:51 +0800
Subject: [PATCH 02/10] enlarge default time

---
 src/rpc/src/rpcMain.c | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/rpc/src/rpcMain.c b/src/rpc/src/rpcMain.c
index 133ae6d0ab..58d611ebb5 100644
--- a/src/rpc/src/rpcMain.c
+++ b/src/rpc/src/rpcMain.c
@@ -295,7 +295,7 @@ void *rpcOpen(const SRpcInit *pInit) {
       return NULL;
     }
   } else {
-    pRpc->pCache = rpcOpenConnCache(pRpc->sessions, rpcCloseConn, pRpc->tmrCtrl, pRpc->idleTime); 
+    pRpc->pCache = rpcOpenConnCache(pRpc->sessions, rpcCloseConn, pRpc->tmrCtrl, pRpc->idleTime * 30); 
     if ( pRpc->pCache == NULL ) {
       tError("%s failed to init connection cache", pRpc->label);
       rpcClose(pRpc);
@@ -470,7 +470,7 @@ void rpcSendResponse(const SRpcMsg *pRsp) {
   taosTmrStopA(&pConn->pTimer);
 
   // set the idle timer to monitor the activity
-  taosTmrReset(rpcProcessIdleTimer, pRpc->idleTime, pConn, pRpc->tmrCtrl, &pConn->pIdleTimer);
+  taosTmrReset(rpcProcessIdleTimer, pRpc->idleTime * 30, pConn, pRpc->tmrCtrl, &pConn->pIdleTimer);
   rpcSendMsgToPeer(pConn, msg, msgLen);
 
   // if not set to secured, set it expcet NOT_READY case, since client wont treat it as secured

From 3a14b45f40977f6d46f67c17d8019f1a66678aaa Mon Sep 17 00:00:00 2001
From: yihaoDeng <yhdeng@taosdata.com>
Date: Fri, 9 Apr 2021 14:56:25 +0800
Subject: [PATCH 03/10] fixbug crash

---
 src/rpc/src/rpcMain.c | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/src/rpc/src/rpcMain.c b/src/rpc/src/rpcMain.c
index 58d611ebb5..98d5c1ed54 100644
--- a/src/rpc/src/rpcMain.c
+++ b/src/rpc/src/rpcMain.c
@@ -997,8 +997,8 @@ static SRpcConn *rpcProcessMsgHead(SRpcInfo *pRpc, SRecvInfo *pRecv, SRpcReqCont
     }
 
     if ( rpcIsReq(pHead->msgType) ) {
-      terrno = rpcProcessReqHead(pConn, pHead);
       pConn->connType = pRecv->connType;
+      terrno = rpcProcessReqHead(pConn, pHead);
 
       // stop idle timer
       taosTmrStopA(&pConn->pIdleTimer);  
@@ -1367,7 +1367,8 @@ static void rpcProcessConnError(void *param, void *id) {
   
   tDebug("%s %p, connection error happens", pRpc->label, pContext->ahandle);
 
-  if (pContext->numOfTry >= pContext->epSet.numOfEps) {
+  if (pContext->numOfTry >= pContext->epSet.numOfEps 
+      || pContex->msgType == TSDB_MSG_TYPE_FETCH) {
     rpcMsg.msgType = pContext->msgType+1;
     rpcMsg.ahandle = pContext->ahandle;
     rpcMsg.code = pContext->code;

From 1511e0c32f13255564836f9d9b9e48255b8b615f Mon Sep 17 00:00:00 2001
From: yihaoDeng <yhdeng@taosdata.com>
Date: Fri, 9 Apr 2021 15:01:29 +0800
Subject: [PATCH 04/10] fixbug crash

---
 src/rpc/src/rpcMain.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/rpc/src/rpcMain.c b/src/rpc/src/rpcMain.c
index 98d5c1ed54..3e8c1d9180 100644
--- a/src/rpc/src/rpcMain.c
+++ b/src/rpc/src/rpcMain.c
@@ -1368,7 +1368,7 @@ static void rpcProcessConnError(void *param, void *id) {
   tDebug("%s %p, connection error happens", pRpc->label, pContext->ahandle);
 
   if (pContext->numOfTry >= pContext->epSet.numOfEps 
-      || pContex->msgType == TSDB_MSG_TYPE_FETCH) {
+      || pContext->msgType == TSDB_MSG_TYPE_FETCH) {
     rpcMsg.msgType = pContext->msgType+1;
     rpcMsg.ahandle = pContext->ahandle;
     rpcMsg.code = pContext->code;

From f5db431c19973e56a11c1e644e521634a7d83fb2 Mon Sep 17 00:00:00 2001
From: yihaoDeng <yhdeng@taosdata.com>
Date: Sat, 10 Apr 2021 17:11:58 +0800
Subject: [PATCH 05/10] reset timeout

---
 src/rpc/src/rpcMain.c | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/rpc/src/rpcMain.c b/src/rpc/src/rpcMain.c
index 7205bafd7d..cd6b3cf80c 100644
--- a/src/rpc/src/rpcMain.c
+++ b/src/rpc/src/rpcMain.c
@@ -295,7 +295,7 @@ void *rpcOpen(const SRpcInit *pInit) {
       return NULL;
     }
   } else {
-    pRpc->pCache = rpcOpenConnCache(pRpc->sessions, rpcCloseConn, pRpc->tmrCtrl, pRpc->idleTime*30); 
+    pRpc->pCache = rpcOpenConnCache(pRpc->sessions, rpcCloseConn, pRpc->tmrCtrl, pRpc->idleTime); 
     if ( pRpc->pCache == NULL ) {
       tError("%s failed to init connection cache", pRpc->label);
       rpcClose(pRpc);
@@ -470,7 +470,7 @@ void rpcSendResponse(const SRpcMsg *pRsp) {
   taosTmrStopA(&pConn->pTimer);
 
   // set the idle timer to monitor the activity
-  taosTmrReset(rpcProcessIdleTimer, pRpc->idleTime*30, pConn, pRpc->tmrCtrl, &pConn->pIdleTimer);
+  taosTmrReset(rpcProcessIdleTimer, pRpc->idleTime, pConn, pRpc->tmrCtrl, &pConn->pIdleTimer);
   rpcSendMsgToPeer(pConn, msg, msgLen);
 
   // if not set to secured, set it expcet NOT_READY case, since client wont treat it as secured

From 35c7506dd096edd534c5b93eef4b4afc6ac647cc Mon Sep 17 00:00:00 2001
From: yihaoDeng <yhdeng@taosdata.com>
Date: Sat, 10 Apr 2021 22:48:35 +0800
Subject: [PATCH 06/10] reset tcp

---
 src/rpc/src/rpcTcp.c | 155 ++++---------------------------------------
 1 file changed, 12 insertions(+), 143 deletions(-)

diff --git a/src/rpc/src/rpcTcp.c b/src/rpc/src/rpcTcp.c
index 286ed223c7..3162ab2e4c 100644
--- a/src/rpc/src/rpcTcp.c
+++ b/src/rpc/src/rpcTcp.c
@@ -21,13 +21,6 @@
 #include "rpcLog.h"
 #include "rpcHead.h"
 #include "rpcTcp.h"
-#include "tlist.h"
-
-typedef struct SConnItem {
-  SOCKET         fd; 
-  uint32_t       ip;
-  uint16_t       port;
-} SConnItem;
 
 typedef struct SFdObj {
   void              *signature;
@@ -45,12 +38,6 @@ typedef struct SThreadObj {
   pthread_t       thread;
   SFdObj *        pHead;
   pthread_mutex_t mutex;
-  // receive the notify from dispatch thread 
-
-  int             notifyReceiveFd;
-  int             notifySendFd;
-  SList           *connQueue;  
-
   uint32_t        ip;
   bool            stop;
   EpollFd         pollFd;
@@ -82,7 +69,6 @@ typedef struct {
 } SServerObj;
 
 static void   *taosProcessTcpData(void *param);
-static void   *taosProcessServerTcpData(void *param);
 static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, SOCKET fd);
 static void    taosFreeFdObj(SFdObj *pFdObj);
 static void    taosReportBrokenLink(SFdObj *pFdObj);
@@ -138,7 +124,6 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
     tstrncpy(pThreadObj->label, label, sizeof(pThreadObj->label));
     pThreadObj->shandle = shandle;
     pThreadObj->stop = false;
-    pThreadObj->connQueue = tdListNew(sizeof(SConnItem));
   }
 
   // initialize mutex, thread, fd which may fail
@@ -157,25 +142,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
       break;
     }
 
-    int fds[2];
-    if (pipe(fds)) {
-      tError("%s failed to create pipe", label);
-      code = -1;
-      break;
-    }
-
-    pThreadObj->notifyReceiveFd = fds[0];
-    pThreadObj->notifySendFd    = fds[1];
-    struct epoll_event event;
-    event.events  = EPOLLIN | EPOLLRDHUP; 
-    event.data.fd = pThreadObj->notifyReceiveFd;
-    if (epoll_ctl(pThreadObj->pollFd, EPOLL_CTL_ADD, pThreadObj->notifyReceiveFd , &event) < 0) {
-      tError("%s failed to create pipe", label);
-      code = -1;
-      break;
-    }
-
-    code = pthread_create(&(pThreadObj->thread), &thattr, taosProcessServerTcpData, (void *)(pThreadObj));
+    code = pthread_create(&(pThreadObj->thread), &thattr, taosProcessTcpData, (void *)(pThreadObj));
     if (code != 0) {
       tError("%s failed to create TCP process data thread(%s)", label, strerror(errno));
       break;
@@ -308,12 +275,17 @@ static void *taosAcceptTcpConnection(void *arg) {
     // pick up the thread to handle this connection
     pThreadObj = pServerObj->pThreadObj[threadId];
 
-    pthread_mutex_lock(&(pThreadObj->mutex));
-    SConnItem item = {.fd = connFd, .ip = caddr.sin_addr.s_addr, .port = htons(caddr.sin_port)};     
-    tdListAppend(pThreadObj->connQueue, &item);     
-    pthread_mutex_unlock(&(pThreadObj->mutex));
-    
-    write(pThreadObj->notifySendFd, "", 1);
+    SFdObj *pFdObj = taosMallocFdObj(pThreadObj, connFd);
+    if (pFdObj) {
+      pFdObj->ip = caddr.sin_addr.s_addr;
+      pFdObj->port = htons(caddr.sin_port);
+      tDebug("%s new TCP connection from %s:%hu, fd:%d FD:%p numOfFds:%d", pServerObj->label,
+              taosInetNtoa(caddr.sin_addr), pFdObj->port, connFd, pFdObj, pThreadObj->numOfFds);
+    } else {
+      taosCloseSocket(connFd);
+      tError("%s failed to malloc FdObj(%s) for connection from:%s:%hu", pServerObj->label, strerror(errno),
+             taosInetNtoa(caddr.sin_addr), htons(caddr.sin_port));
+    }
 
     // pick up next thread for next connection
     threadId++;
@@ -619,109 +591,6 @@ static void *taosProcessTcpData(void *param) {
   return NULL;
 }
 
-static void *taosProcessServerTcpData(void *param) {
-  SThreadObj        *pThreadObj = param;
-  SFdObj            *pFdObj;
-  struct epoll_event events[maxEvents];
-  SRecvInfo          recvInfo;
-
-  char               bb[1];
-#ifdef __APPLE__
-  taos_block_sigalrm();
-#endif // __APPLE__
-  while (1) {
-    int fdNum = epoll_wait(pThreadObj->pollFd, events, maxEvents, TAOS_EPOLL_WAIT_TIME);
-    if (pThreadObj->stop) {
-      tDebug("%s TCP thread get stop event, exiting...", pThreadObj->label);
-      break;
-    }
-    if (fdNum < 0) continue;
-
-    for (int i = 0; i < fdNum; ++i) {
-      if (events[i].data.fd == pThreadObj->notifyReceiveFd) {
-        if (events[i].events & EPOLLIN) {
-          read(pThreadObj->notifyReceiveFd, bb, 1);
-
-          pthread_mutex_lock(&(pThreadObj->mutex));
-          SListNode *head = tdListPopHead(pThreadObj->connQueue);  
-          pthread_mutex_unlock(&(pThreadObj->mutex));
-
-          SConnItem item = {0};  
-          tdListNodeGetData(pThreadObj->connQueue, head, &item);   
-          tfree(head);
-          
-          // register fd on epoll 
-          SFdObj *pFdObj = taosMallocFdObj(pThreadObj, item.fd);
-          if (pFdObj) {
-            pFdObj->ip   = item.ip;
-            pFdObj->port = item.port; 
-            tDebug("%s new TCP connection from %u:%hu, fd:%d FD:%p numOfFds:%d", pThreadObj->label,
-              pFdObj->ip, pFdObj->port, item.fd, pFdObj, pThreadObj->numOfFds);
-          } else {
-            taosCloseSocket(item.fd);
-            tError("%s failed to malloc FdObj(%s) for connection from:%u:%hu", pThreadObj->label, strerror(errno),
-               pFdObj->ip, pFdObj->port);
-          }
-        } 
-        continue; 
-      }
-      pFdObj = events[i].data.ptr;
-
-      if (events[i].events & EPOLLERR) {
-        tDebug("%s %p FD:%p epoll errors", pThreadObj->label, pFdObj->thandle, pFdObj);
-        taosReportBrokenLink(pFdObj);
-        continue;
-      }
-
-      if (events[i].events & EPOLLRDHUP) {
-        tDebug("%s %p FD:%p RD hang up", pThreadObj->label, pFdObj->thandle, pFdObj);
-        taosReportBrokenLink(pFdObj);
-        continue;
-      }
-
-      if (events[i].events & EPOLLHUP) {
-        tDebug("%s %p FD:%p hang up", pThreadObj->label, pFdObj->thandle, pFdObj);
-        taosReportBrokenLink(pFdObj);
-        continue;
-      }
-
-      if (taosReadTcpData(pFdObj, &recvInfo) < 0) {
-        shutdown(pFdObj->fd, SHUT_WR);
-        continue;
-      }
-
-      pFdObj->thandle = (*(pThreadObj->processData))(&recvInfo);
-      if (pFdObj->thandle == NULL) taosFreeFdObj(pFdObj);
-    }
-
-    if (pThreadObj->stop) break;
-  }
-
-  if (pThreadObj->connQueue) {
-    pThreadObj->connQueue = tdListFree(pThreadObj->connQueue); 
-  }
-  // close pipe
-  close(pThreadObj->notifySendFd);
-  close(pThreadObj->notifyReceiveFd);
-
-  if (pThreadObj->pollFd >=0) {
-    EpollClose(pThreadObj->pollFd);
-    pThreadObj->pollFd = -1;
-  }
-
-  while (pThreadObj->pHead) {
-    SFdObj *pFdObj = pThreadObj->pHead;
-    pThreadObj->pHead = pFdObj->next;
-    taosReportBrokenLink(pFdObj);
-  }
-
-  pthread_mutex_destroy(&(pThreadObj->mutex));
-  tDebug("%s TCP thread exits ...", pThreadObj->label);
-  tfree(pThreadObj);
-
-  return NULL;
-}
-
 static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, SOCKET fd) {
   struct epoll_event event;
 

From ea771cee5915aa46e6ab2d2e5564fc584b3e2305 Mon Sep 17 00:00:00 2001
From: Shuduo Sang <sangshuduo@gmail.com>
Date: Mon, 12 Apr 2021 14:35:32 +0800
Subject: [PATCH 07/10] [TD-3752]<fix>: taosdemo columns+tags+ts must less then
 1024. (#5782)

Co-authored-by: Shuduo Sang <sdsang@taosdata.com>
---
 src/kit/taosdemo/taosdemo.c | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/src/kit/taosdemo/taosdemo.c b/src/kit/taosdemo/taosdemo.c
index 49e550cc01..815a842d60 100644
--- a/src/kit/taosdemo/taosdemo.c
+++ b/src/kit/taosdemo/taosdemo.c
@@ -2980,7 +2980,7 @@ static bool getColumnAndTagTypeFromInsertJsonFile(
   }
 
   int columnSize = cJSON_GetArraySize(columns);
-  if (columnSize > MAX_COLUMN_COUNT) {
+  if ((columnSize + 1/* ts */) > MAX_COLUMN_COUNT) {
     errorPrint("%s() LN%d, failed to read json, column size overflow, max column size is %d\n",
             __func__, __LINE__, MAX_COLUMN_COUNT);
     goto PARSE_OVER;
@@ -3038,8 +3038,8 @@ static bool getColumnAndTagTypeFromInsertJsonFile(
     }
   }
 
-  if (index > MAX_COLUMN_COUNT) {
-    errorPrint("%s() LN%d, failed to read json, column size overflow, max column size is %d\n",
+  if ((index + 1 /* ts */) > MAX_COLUMN_COUNT) {
+    errorPrint("%s() LN%d, failed to read json, column size overflow, allowed max column size is %d\n",
             __func__, __LINE__, MAX_COLUMN_COUNT);
     goto PARSE_OVER;
   }
@@ -3110,16 +3110,16 @@ static bool getColumnAndTagTypeFromInsertJsonFile(
   }
 
   if (index > MAX_TAG_COUNT) {
-    errorPrint("%s() LN%d, failed to read json, tags size overflow, max tag size is %d\n",
+    errorPrint("%s() LN%d, failed to read json, tags size overflow, allowed max tag count is %d\n",
         __func__, __LINE__, MAX_TAG_COUNT);
     goto PARSE_OVER;
   }
 
   superTbls->tagCount = index;
 
-  if ((superTbls->columnCount + superTbls->tagCount) > MAX_COLUMN_COUNT) {
-    errorPrint("%s() LN%d, columns + tags is more than max columns count: %d\n",
-        __func__, __LINE__, MAX_TAG_COUNT);
+  if ((superTbls->columnCount + superTbls->tagCount + 1 /* ts */) > MAX_COLUMN_COUNT) {
+    errorPrint("%s() LN%d, columns + tags is more than allowed max columns count: %d\n",
+        __func__, __LINE__, MAX_COLUMN_COUNT);
     goto PARSE_OVER;
   }
   ret = true;

From b6ee0c8fc77ae35a47ca510530b8f99f368a4be1 Mon Sep 17 00:00:00 2001
From: Steven Li <stevenli@taosdata.com>
Date: Tue, 13 Apr 2021 05:49:22 +0000
Subject: [PATCH 08/10] Moved over changes from feature/crash_gen to
 feature/crash_gen_master

---
 tests/pytest/crash_gen/README.md          |  44 ++++-
 tests/pytest/crash_gen/crash_gen_main.py  |  13 +-
 tests/pytest/crash_gen/service_manager.py | 219 ++++++++++++++++------
 3 files changed, 206 insertions(+), 70 deletions(-)

diff --git a/tests/pytest/crash_gen/README.md b/tests/pytest/crash_gen/README.md
index 6788ab1a63..8cdddd5b57 100644
--- a/tests/pytest/crash_gen/README.md
+++ b/tests/pytest/crash_gen/README.md
@@ -6,6 +6,25 @@ To effectively test and debug our TDengine product, we have developed a simple t
 exercise various functions of the system in a randomized fashion, hoping to expose 
 maximum number of problems, hopefully without a pre-determined scenario.
 
+# Features
+
+This tool can run as a test client with the following features:
+
+1. Any number of concurrent threads
+1. Any number of test steps/loops
+1. Auto-create and writing to multiple databases
+1. Ignore specific error codes
+1. Write small or large data blocks
+1. Auto-generate out-of-sequence data, if needed
+1. Verify the result of write operations
+1. Concurrent writing to a shadow database for later data verification
+1. User specified number of replicas to use, against clusters
+
+This tool can also use to start a TDengine service, either in stand-alone mode or 
+cluster mode. The features include:
+
+1. User specified number of D-Nodes to create/use.
+
 # Preparation
 
 To run this tool, please ensure the followed preparation work is done first.
@@ -16,7 +35,7 @@ To run this tool, please ensure the followed preparation work is done first.
     Ubuntu 20.04LTS as our own development environment, and suggest you also use such
     an environment if possible.
 
-# Simple Execution
+# Simple Execution as Client Test Tool
 
 To run the tool with the simplest method, follow the steps below:
 
@@ -28,19 +47,21 @@ To run the tool with the simplest method, follow the steps below:
 
 That's it!
 
-# Running Clusters
+# Running Server-side Clusters
 
 This tool also makes it easy to test/verify the clustering capabilities of TDengine. You
 can start a cluster quite easily with the following command:
 
 ```
 $ cd tests/pytest/
-$ ./crash_gen.sh -e -o 3
+$ rm -rf ../../build/cluster_dnode_?; ./crash_gen.sh -e -o 3   # first part optional
 ```
 
 The `-e` option above tells the tool to start the service, and do not run any tests, while 
 the `-o 3` option tells the tool to start 3 DNodes and join them together in a cluster. 
-Obviously you can adjust the the number here.
+Obviously you can adjust the the number here. The `rm -rf` command line is optional 
+to clean up previous cluster data, so that we can start from a clean state with no data
+at all.
 
 ## Behind the Scenes
 
@@ -89,8 +110,9 @@ The exhaustive features of the tool is available through the `-h` option:
 
 ```
 $ ./crash_gen.sh -h
-usage: crash_gen_bootstrap.py [-h] [-a] [-b MAX_DBS] [-c CONNECTOR_TYPE] [-d] [-e] [-g IGNORE_ERRORS] [-i MAX_REPLICAS] [-l] [-n] [-o NUM_DNODES] [-p] [-r]
-                              [-s MAX_STEPS] [-t NUM_THREADS] [-v] [-x]
+usage: crash_gen_bootstrap.py [-h] [-a] [-b MAX_DBS] [-c CONNECTOR_TYPE] [-d] [-e] [-g IGNORE_ERRORS] 
+    [-i NUM_REPLICAS] [-k] [-l] [-m] [-n]
+    [-o NUM_DNODES] [-p] [-r] [-s MAX_STEPS] [-t NUM_THREADS] [-v] [-w] [-x]
 
 TDengine Auto Crash Generator (PLEASE NOTICE the Prerequisites Below)
 ---------------------------------------------------------------------
@@ -109,11 +131,14 @@ optional arguments:
   -e, --run-tdengine    Run TDengine service in foreground (default: false)
   -g IGNORE_ERRORS, --ignore-errors IGNORE_ERRORS
                         Ignore error codes, comma separated, 0x supported (default: None)
-  -i MAX_REPLICAS, --max-replicas MAX_REPLICAS
-                        Maximum number of replicas to use, when testing against clusters. (default: 1)
+  -i NUM_REPLICAS, --num-replicas NUM_REPLICAS
+                        Number (fixed) of replicas to use, when testing against clusters. (default: 1)
+  -k, --track-memory-leaks
+                        Use Valgrind tool to track memory leaks (default: false)
   -l, --larger-data     Write larger amount of data during write operations (default: false)
+  -m, --mix-oos-data    Mix out-of-sequence data into the test data stream (default: true)
   -n, --dynamic-db-table-names
-                        Use non-fixed names for dbs/tables, useful for multi-instance executions (default: false)
+                        Use non-fixed names for dbs/tables, for -b, useful for multi-instance executions (default: false)
   -o NUM_DNODES, --num-dnodes NUM_DNODES
                         Number of Dnodes to initialize, used with -e option. (default: 1)
   -p, --per-thread-db-connection
@@ -124,6 +149,7 @@ optional arguments:
   -t NUM_THREADS, --num-threads NUM_THREADS
                         Number of threads to run (default: 10)
   -v, --verify-data     Verify data written in a number of places by reading back (default: false)
+  -w, --use-shadow-db   Use a shaddow database to verify data integrity (default: false)
   -x, --continue-on-exception
                         Continue execution after encountering unexpected/disallowed errors/exceptions (default: false)
 ```
diff --git a/tests/pytest/crash_gen/crash_gen_main.py b/tests/pytest/crash_gen/crash_gen_main.py
index 0fe94bad1d..44295e8bee 100755
--- a/tests/pytest/crash_gen/crash_gen_main.py
+++ b/tests/pytest/crash_gen/crash_gen_main.py
@@ -1574,9 +1574,9 @@ class TaskCreateDb(StateTransitionTask):
     def _executeInternal(self, te: TaskExecutor, wt: WorkerThread):
         # was: self.execWtSql(wt, "create database db")
         repStr = ""
-        if gConfig.max_replicas != 1:
+        if gConfig.num_replicas != 1:
             # numReplica = Dice.throw(gConfig.max_replicas) + 1 # 1,2 ... N
-            numReplica = gConfig.max_replicas # fixed, always
+            numReplica = gConfig.num_replicas # fixed, always
             repStr = "replica {}".format(numReplica)
         updatePostfix = "update 1" if gConfig.verify_data else "" # allow update only when "verify data" is active
         dbName = self._db.getName()
@@ -2394,11 +2394,16 @@ class MainExec:
             help='Ignore error codes, comma separated, 0x supported (default: None)')
         parser.add_argument(
             '-i',
-            '--max-replicas',
+            '--num-replicas',
             action='store',
             default=1,
             type=int,
-            help='Maximum number of replicas to use, when testing against clusters. (default: 1)')
+            help='Number (fixed) of replicas to use, when testing against clusters. (default: 1)')
+        parser.add_argument(
+            '-k',
+            '--track-memory-leaks',
+            action='store_true',
+            help='Use Valgrind tool to track memory leaks (default: false)')
         parser.add_argument(
             '-l',
             '--larger-data',
diff --git a/tests/pytest/crash_gen/service_manager.py b/tests/pytest/crash_gen/service_manager.py
index ae6f8d5d3a..cdbf2db4da 100644
--- a/tests/pytest/crash_gen/service_manager.py
+++ b/tests/pytest/crash_gen/service_manager.py
@@ -19,6 +19,7 @@ from queue import Queue, Empty
 
 from .misc import Logging, Status, CrashGenError, Dice, Helper, Progress
 from .db import DbConn, DbTarget
+import crash_gen.settings 
 
 class TdeInstance():
     """
@@ -132,6 +133,7 @@ keep 36500
 walLevel 1
 #
 # maxConnections 100
+quorum 2
 """
         cfgContent = cfgTemplate.format_map(cfgValues)
         f = open(cfgFile, "w")
@@ -164,7 +166,12 @@ walLevel 1
         return "127.0.0.1"
 
     def getServiceCmdLine(self): # to start the instance
-        return [self.getExecFile(), '-c', self.getCfgDir()] # used in subproce.Popen()
+        cmdLine = []
+        if crash_gen.settings.gConfig.track_memory_leaks:
+            Logging.info("Invoking VALGRIND on service...")
+            cmdLine = ['valgrind', '--leak-check=yes']
+        cmdLine += ["exec " + self.getExecFile(), '-c', self.getCfgDir()] # used in subproce.Popen()        
+        return cmdLine
     
     def _getDnodes(self, dbc):
         dbc.query("show dnodes")
@@ -202,7 +209,7 @@ walLevel 1
         self.generateCfgFile() # service side generates config file, client does not
         self.rotateLogs()
 
-        self._smThread.start(self.getServiceCmdLine())
+        self._smThread.start(self.getServiceCmdLine(), self.getLogDir()) # May raise exceptions
 
     def stop(self):
         self._smThread.stop()
@@ -225,7 +232,7 @@ class TdeSubProcess:
     # RET_SUCCESS = -4
 
     def __init__(self):
-        self.subProcess = None
+        self.subProcess = None # type: subprocess.Popen
         # if tInst is None:
         #     raise CrashGenError("Empty instance not allowed in TdeSubProcess")
         # self._tInst = tInst # Default create at ServiceManagerThread
@@ -263,7 +270,7 @@ class TdeSubProcess:
         # print("Starting TDengine with env: ", myEnv.items())
         # print("Starting TDengine via Shell: {}".format(cmdLineStr))
 
-        useShell = True    
+        useShell = True # Needed to pass environments into it
         self.subProcess = subprocess.Popen(
             # ' '.join(cmdLine) if useShell else cmdLine,
             # shell=useShell,
@@ -276,12 +283,12 @@ class TdeSubProcess:
             env=myEnv
             )  # had text=True, which interferred with reading EOF
 
-    STOP_SIGNAL = signal.SIGKILL # signal.SIGKILL/SIGINT # What signal to use (in kill) to stop a taosd process?
+    STOP_SIGNAL = signal.SIGINT # signal.SIGKILL/SIGINT # What signal to use (in kill) to stop a taosd process?
     SIG_KILL_RETCODE = 137 # ref: https://stackoverflow.com/questions/43268156/process-finished-with-exit-code-137-in-pycharm
 
     def stop(self):
         """
-        Stop a sub process, and try to return a meaningful return code.
+        Stop a sub process, DO NOT return anything, process all conditions INSIDE
 
         Common POSIX signal values (from man -7 signal):
         SIGHUP           1
@@ -301,40 +308,99 @@ class TdeSubProcess:
         """
         if not self.subProcess:
             Logging.error("Sub process already stopped")
-            return  # -1
+            return
 
         retCode = self.subProcess.poll() # ret -N means killed with signal N, otherwise it's from exit(N)
         if retCode:  # valid return code, process ended
-            retCode = -retCode # only if valid
+            # retCode = -retCode # only if valid
             Logging.warning("TSP.stop(): process ended itself")
             self.subProcess = None
-            return retCode
+            return
 
         # process still alive, let's interrupt it
-        Logging.info("Terminate running process, send SIG_{} and wait...".format(self.STOP_SIGNAL))
-        # sub process should end, then IPC queue should end, causing IO thread to end        
-        topSubProc = psutil.Process(self.subProcess.pid)
-        for child in topSubProc.children(recursive=True):  # or parent.children() for recursive=False
-            child.send_signal(self.STOP_SIGNAL)
-            time.sleep(0.2) # 200 ms
-        # topSubProc.send_signal(sig) # now kill the main sub process (likely the Shell)
+        self._stopForSure(self.subProcess, self.STOP_SIGNAL) # success if no exception
+        self.subProcess = None
 
-        self.subProcess.send_signal(self.STOP_SIGNAL) # main sub process (likely the Shell)
-        self.subProcess.wait(20)
-        retCode = self.subProcess.returncode # should always be there
-        # May throw subprocess.TimeoutExpired exception above, therefore
-        # The process is guranteed to have ended by now
-        self.subProcess = None       
-        if retCode == self.SIG_KILL_RETCODE:
-            Logging.info("TSP.stop(): sub proc KILLED, as expected")
-        elif retCode == (- self.STOP_SIGNAL):
-            Logging.info("TSP.stop(), sub process STOPPED, as expected")
-        elif retCode != 0: # != (- signal.SIGINT):
-            Logging.error("TSP.stop(): Failed to stop sub proc properly w/ SIG {}, retCode={}".format(
-                self.STOP_SIGNAL, retCode))
-        else:
-            Logging.info("TSP.stop(): sub proc successfully terminated with SIG {}".format(self.STOP_SIGNAL))
-        return - retCode
+        # sub process should end, then IPC queue should end, causing IO thread to end        
+
+    @classmethod
+    def _stopForSure(cls, proc: subprocess.Popen, sig: int):
+        ''' 
+        Stop a process and all sub processes with a singal, and SIGKILL if necessary
+        '''
+        def doKillTdService(proc: subprocess.Popen, sig: int):
+            Logging.info("Killing sub-sub process {} with signal {}".format(proc.pid, sig))
+            proc.send_signal(sig)
+            try:            
+                retCode = proc.wait(20)
+                if (- retCode) == signal.SIGSEGV: # Crashed
+                    Logging.warning("Process {} CRASHED, please check CORE file!".format(proc.pid))
+                elif (- retCode) == sig : 
+                    Logging.info("TD service terminated with expected return code {}".format(sig))
+                else:
+                    Logging.warning("TD service terminated, EXPECTING ret code {}, got {}".format(sig, -retCode))
+                return True # terminated successfully
+            except subprocess.TimeoutExpired as err:
+                Logging.warning("Failed to kill sub-sub process {} with signal {}".format(proc.pid, sig))
+            return False # failed to terminate
+
+
+        def doKillChild(child: psutil.Process, sig: int):
+            Logging.info("Killing sub-sub process {} with signal {}".format(child.pid, sig))
+            child.send_signal(sig)
+            try:            
+                retCode = child.wait(20)
+                if (- retCode) == signal.SIGSEGV: # Crashed
+                    Logging.warning("Process {} CRASHED, please check CORE file!".format(child.pid))
+                elif (- retCode) == sig : 
+                    Logging.info("Sub-sub process terminated with expected return code {}".format(sig))
+                else:
+                    Logging.warning("Process terminated, EXPECTING ret code {}, got {}".format(sig, -retCode))
+                return True # terminated successfully
+            except psutil.TimeoutExpired as err:
+                Logging.warning("Failed to kill sub-sub process {} with signal {}".format(child.pid, sig))
+            return False # did not terminate
+
+        def doKill(proc: subprocess.Popen, sig: int):
+            pid = proc.pid
+            try:
+                topSubProc = psutil.Process(pid)
+                for child in topSubProc.children(recursive=True):  # or parent.children() for recursive=False
+                    Logging.warning("Unexpected child to be killed")
+                    doKillChild(child, sig)
+            except psutil.NoSuchProcess as err:
+                Logging.info("Process not found, can't kill, pid = {}".format(pid))
+            
+            return doKillTdService(proc, sig)
+            # TODO: re-examine if we need to kill the top process, which is always the SHELL for now
+            # try:
+            #     proc.wait(1) # SHELL process here, may throw subprocess.TimeoutExpired exception
+            #     # expRetCode = self.SIG_KILL_RETCODE if sig==signal.SIGKILL else (-sig)
+            #     # if retCode == expRetCode:
+            #     #     Logging.info("Process terminated with expected return code {}".format(retCode))
+            #     # else:
+            #     #     Logging.warning("Process terminated, EXPECTING ret code {}, got {}".format(expRetCode, retCode))
+            #     # return True # success
+            # except subprocess.TimeoutExpired as err:
+            #     Logging.warning("Failed to kill process {} with signal {}".format(pid, sig))
+            # return False # failed to kill
+
+        def softKill(proc, sig):
+            return doKill(proc, sig)
+
+        def hardKill(proc):
+            return doKill(proc, signal.SIGKILL)
+
+        
+
+        pid = proc.pid
+        Logging.info("Terminate running processes under {}, with SIG #{} and wait...".format(pid, sig))
+        if softKill(proc, sig):            
+            return# success
+        if sig != signal.SIGKILL: # really was soft above            
+            if hardKill(proc):
+                return
+        raise CrashGenError("Failed to stop process, pid={}".format(pid))
 
 class ServiceManager:
     PAUSE_BETWEEN_IPC_CHECK = 1.2  # seconds between checks on STDOUT of sub process
@@ -560,7 +626,8 @@ class ServiceManagerThread:
         # self._tInstNum = tInstNum # instance serial number in cluster, ZERO based
         # self._tInst    = tInst or TdeInstance() # Need an instance
 
-        self._thread = None # The actual thread, # type: threading.Thread
+        self._thread  = None # The actual thread, # type: threading.Thread
+        self._thread2 = None # watching stderr
         self._status = Status(Status.STATUS_STOPPED) # The status of the underlying service, actually.
 
     def __repr__(self):
@@ -568,11 +635,20 @@ class ServiceManagerThread:
             self.getStatus(), self._tdeSubProcess)
 
     def getStatus(self):
+        '''
+        Get the status of the process being managed. (misnomer alert!)
+        '''
         return self._status
 
     # Start the thread (with sub process), and wait for the sub service
     # to become fully operational
-    def start(self, cmdLine):
+    def start(self, cmdLine : str, logDir: str):
+        '''
+        Request the manager thread to start a new sub process, and manage it.
+
+        :param cmdLine: the command line to invoke
+        :param logDir: the logging directory, to hold stdout/stderr files
+        '''
         if self._thread:
             raise RuntimeError("Unexpected _thread")
         if self._tdeSubProcess:
@@ -582,20 +658,30 @@ class ServiceManagerThread:
 
         self._status.set(Status.STATUS_STARTING)
         self._tdeSubProcess = TdeSubProcess()
-        self._tdeSubProcess.start(cmdLine)
+        self._tdeSubProcess.start(cmdLine) # TODO: verify process is running
 
         self._ipcQueue = Queue()
         self._thread = threading.Thread( # First thread captures server OUTPUT
             target=self.svcOutputReader,
-            args=(self._tdeSubProcess.getStdOut(), self._ipcQueue))
+            args=(self._tdeSubProcess.getStdOut(), self._ipcQueue, logDir))
         self._thread.daemon = True  # thread dies with the program
         self._thread.start()
+        time.sleep(0.01)
+        if not self._thread.is_alive(): # What happened?
+            Logging.info("Failed to started process to monitor STDOUT")
+            self.stop()
+            raise CrashGenError("Failed to start thread to monitor STDOUT")
+        Logging.info("Successfully started process to monitor STDOUT")
 
         self._thread2 = threading.Thread( # 2nd thread captures server ERRORs
             target=self.svcErrorReader,
-            args=(self._tdeSubProcess.getStdErr(), self._ipcQueue))
+            args=(self._tdeSubProcess.getStdErr(), self._ipcQueue, logDir))
         self._thread2.daemon = True  # thread dies with the program
         self._thread2.start()
+        time.sleep(0.01)
+        if not self._thread2.is_alive():
+            self.stop()
+            raise CrashGenError("Failed to start thread to monitor STDERR")
 
         # wait for service to start
         for i in range(0, 100):
@@ -643,7 +729,7 @@ class ServiceManagerThread:
             Logging.info("Service already stopped")
             return
         if self.getStatus().isStopping():
-            Logging.info("Service is already being stopped")
+            Logging.info("Service is already being stopped, pid: {}".format(self._tdeSubProcess.getPid()))
             return
         # Linux will send Control-C generated SIGINT to the TDengine process
         # already, ref:
@@ -653,14 +739,14 @@ class ServiceManagerThread:
 
         self._status.set(Status.STATUS_STOPPING)
         # retCode = self._tdeSubProcess.stop()
-        try:
-            retCode = self._tdeSubProcess.stop()
-            # print("Attempted to stop sub process, got return code: {}".format(retCode))
-            if retCode == signal.SIGSEGV : # SGV
-                Logging.error("[[--ERROR--]]: TDengine service SEGV fault (check core file!)")
-        except subprocess.TimeoutExpired as err:
-            Logging.info("Time out waiting for TDengine service process to exit")
-        else:    
+        # try:
+        #     retCode = self._tdeSubProcess.stop()
+        #     # print("Attempted to stop sub process, got return code: {}".format(retCode))
+        #     if retCode == signal.SIGSEGV : # SGV
+        #         Logging.error("[[--ERROR--]]: TDengine service SEGV fault (check core file!)")
+        # except subprocess.TimeoutExpired as err:
+        #     Logging.info("Time out waiting for TDengine service process to exit")
+        if not self._tdeSubProcess.stop(): # everything withing
             if self._tdeSubProcess.isRunning():  # still running, should now never happen
                 Logging.error("FAILED to stop sub process, it is still running... pid = {}".format(
                     self._tdeSubProcess.getPid()))
@@ -683,16 +769,18 @@ class ServiceManagerThread:
             raise RuntimeError(
                 "SMT.Join(): Unexpected status: {}".format(self._status))
 
-        if self._thread:
-            self._thread.join()
-            self._thread = None
-            self._status.set(Status.STATUS_STOPPED)
-            # STD ERR thread
-            self._thread2.join()
-            self._thread2 = None
+        if self._thread or self._thread2 :
+            if self._thread:
+                self._thread.join()
+                self._thread = None
+            if self._thread2: # STD ERR thread            
+                self._thread2.join()
+                self._thread2 = None
         else:
             print("Joining empty thread, doing nothing")
 
+        self._status.set(Status.STATUS_STOPPED)
+
     def _trimQueue(self, targetSize):
         if targetSize <= 0:
             return  # do nothing
@@ -739,11 +827,22 @@ class ServiceManagerThread:
         print(pBar, end="", flush=True)
         print('\b\b\b\b', end="", flush=True)
 
-    def svcOutputReader(self, out: IO, queue):
+    def svcOutputReader(self, out: IO, queue, logDir: str):
+        '''
+        The infinite routine that processes the STDOUT stream for the sub process being managed.
+
+        :param out: the IO stream object used to fetch the data from
+        :param queue: the queue where we dump the roughly parsed line-by-line data
+        :param logDir: where we should dump a verbatim output file
+        '''
+        os.makedirs(logDir, exist_ok=True)
+        logFile = os.path.join(logDir,'stdout.log')
+        fOut = open(logFile, 'wb')
         # Important Reference: https://stackoverflow.com/questions/375427/non-blocking-read-on-a-subprocess-pipe-in-python
         # print("This is the svcOutput Reader...")
         # for line in out :
         for line in iter(out.readline, b''):
+            fOut.write(line)
             # print("Finished reading a line: {}".format(line))
             # print("Adding item to queue...")
             try:
@@ -772,10 +871,16 @@ class ServiceManagerThread:
             # queue.put(line)
         # meaning sub process must have died
         Logging.info("EOF for TDengine STDOUT: {}".format(self))
-        out.close()
+        out.close() # Close the stream
+        fOut.close() # Close the output file
 
-    def svcErrorReader(self, err: IO, queue):
+    def svcErrorReader(self, err: IO, queue, logDir: str):
+        os.makedirs(logDir, exist_ok=True)
+        logFile = os.path.join(logDir,'stderr.log')
+        fErr = open(logFile, 'wb')
         for line in iter(err.readline, b''):
+            fErr.write(line)
             Logging.info("TDengine STDERR: {}".format(line))
         Logging.info("EOF for TDengine STDERR: {}".format(self))
-        err.close()
\ No newline at end of file
+        err.close()
+        fErr.close()
\ No newline at end of file

From 640062a53d571e2e791e019b1f722b0e5a6019bd Mon Sep 17 00:00:00 2001
From: Shuduo Sang <sangshuduo@gmail.com>
Date: Tue, 13 Apr 2021 16:24:42 +0800
Subject: [PATCH 09/10] Hotfix/sangshuduo/td 3401 query statistic (#5795)

* [TD-3401]<fix>: taosdemo query statistic.

refactor func name.

* [TD-3401]<fix>: taosdemo query statistic.

refactor func name 2.

* [TD-3401]<fix>: taosdemo support query statistic.

implementation.

* cleanup

Co-authored-by: Shuduo Sang <sdsang@taosdata.com>
---
 src/kit/taosdemo/insert-interlace.json        |   1 -
 src/kit/taosdemo/insert.json                  |   1 -
 src/kit/taosdemo/query.json                   |  37 ++-
 src/kit/taosdemo/taosdemo.c                   | 265 ++++++++++--------
 .../taosdemo/src/main/resources/insert.json   |   2 -
 .../cluster/clusterEnvSetup/insert.json       |   2 -
 tests/pytest/perfbenchmark/bug3433.py         |   4 -
 tests/pytest/query/query1970YearsAf.py        |   4 +-
 tests/pytest/tools/insert-interlace.json      |   1 -
 .../insert-tblimit-tboffset-createdb.json     |   2 -
 .../insert-tblimit-tboffset-insertrec.json    |   2 -
 .../pytest/tools/insert-tblimit-tboffset.json |   2 -
 .../tools/insert-tblimit-tboffset0.json       |   2 -
 .../tools/insert-tblimit1-tboffset.json       |   2 -
 tests/pytest/tools/taosdemo-sampledata.json   |   2 -
 tests/pytest/tools/taosdemoPerformance.py     |   2 -
 16 files changed, 183 insertions(+), 148 deletions(-)

diff --git a/src/kit/taosdemo/insert-interlace.json b/src/kit/taosdemo/insert-interlace.json
index 344db4fd00..cf3e1de2f4 100644
--- a/src/kit/taosdemo/insert-interlace.json
+++ b/src/kit/taosdemo/insert-interlace.json
@@ -40,7 +40,6 @@
             "data_source": "rand",
             "insert_mode": "taosc",
             "insert_rows": 1000,
-            "multi_thread_write_one_tbl": "no",
             "interlace_rows": 20,
             "max_sql_len": 1024000,
             "disorder_ratio": 0,
diff --git a/src/kit/taosdemo/insert.json b/src/kit/taosdemo/insert.json
index f0e3ab1d50..5f152a2d0c 100644
--- a/src/kit/taosdemo/insert.json
+++ b/src/kit/taosdemo/insert.json
@@ -40,7 +40,6 @@
             "data_source": "rand",
             "insert_mode": "taosc",
             "insert_rows": 100000,
-            "multi_thread_write_one_tbl": "no",
             "interlace_rows": 0,
             "max_sql_len": 1024000,
             "disorder_ratio": 0,
diff --git a/src/kit/taosdemo/query.json b/src/kit/taosdemo/query.json
index 33ac120bda..d84f997c32 100644
--- a/src/kit/taosdemo/query.json
+++ b/src/kit/taosdemo/query.json
@@ -1,5 +1,5 @@
 {
-  "filetype":"query",
+  "filetype": "query",
   "cfgdir": "/etc/taos",
   "host": "127.0.0.1",
   "port": 6030,
@@ -7,13 +7,30 @@
   "password": "taosdata",
   "confirm_parameter_prompt": "yes",
   "databases": "dbx",
-  "specified_table_query":
-    {"query_interval":1, "concurrent":4,
-     "sqls": [{"sql": "select last_row(*) from stb where color='red'", "result": "./query_res0.txt"},
-              {"sql": "select count(*) from stb_01", "result": "./query_res1.txt"}]
-    },
-  "super_table_query": 
-    {"stblname": "stb", "query_interval":1, "threads":4,
-     "sqls": [{"sql": "select last_row(*) from xxxx", "result": "./query_res2.txt"}]
-    }
+  "query_times": 1,
+  "specified_table_query": {
+    "query_interval": 1,
+    "concurrent": 4,
+    "sqls": [
+      {
+        "sql": "select last_row(*) from stb where color='red'",
+        "result": "./query_res0.txt"
+      },
+      {
+        "sql": "select count(*) from stb_01",
+        "result": "./query_res1.txt"
+      }
+    ]
+  },
+  "super_table_query": {
+    "stblname": "stb",
+    "query_interval": 1,
+    "threads": 4,
+    "sqls": [
+      {
+        "sql": "select last_row(*) from xxxx",
+        "result": "./query_res2.txt"
+      }
+    ]
+  }
 }
diff --git a/src/kit/taosdemo/taosdemo.c b/src/kit/taosdemo/taosdemo.c
index 815a842d60..dd2891fd32 100644
--- a/src/kit/taosdemo/taosdemo.c
+++ b/src/kit/taosdemo/taosdemo.c
@@ -366,6 +366,7 @@ typedef struct SpecifiedQueryInfo_S {
   char         sql[MAX_QUERY_SQL_COUNT][MAX_QUERY_SQL_LENGTH+1];
   char         result[MAX_QUERY_SQL_COUNT][MAX_FILE_NAME_LEN+1];
   TAOS_SUB*    tsub[MAX_QUERY_SQL_COUNT];
+  int          totalQueried;
 } SpecifiedQueryInfo;
 
 typedef struct SuperQueryInfo_S {
@@ -385,6 +386,7 @@ typedef struct SuperQueryInfo_S {
   TAOS_SUB*    tsub[MAX_QUERY_SQL_COUNT];
 
   char*        childTblName;
+  int          totalQueried;
 } SuperQueryInfo;
 
 typedef struct SQueryMetaInfo_S {
@@ -398,6 +400,7 @@ typedef struct SQueryMetaInfo_S {
 
   SpecifiedQueryInfo  specifiedQueryInfo;
   SuperQueryInfo    superQueryInfo;
+  int          totalQueried;
 } SQueryMetaInfo;
 
 typedef struct SThreadInfo_S {
@@ -2602,8 +2605,8 @@ static int createDatabasesAndStables() {
 
 static void* createTable(void *sarg)
 {
-  threadInfo *winfo = (threadInfo *)sarg;
-  SSuperTable* superTblInfo = winfo->superTblInfo;
+  threadInfo *pThreadInfo = (threadInfo *)sarg;
+  SSuperTable* superTblInfo = pThreadInfo->superTblInfo;
 
   int64_t  lastPrintTime = taosGetTimestampMs();
 
@@ -2621,15 +2624,15 @@ static void* createTable(void *sarg)
 
   verbosePrint("%s() LN%d: Creating table from %d to %d\n",
           __func__, __LINE__,
-          winfo->start_table_from, winfo->end_table_to);
+          pThreadInfo->start_table_from, pThreadInfo->end_table_to);
 
-  for (int i = winfo->start_table_from; i <= winfo->end_table_to; i++) {
+  for (int i = pThreadInfo->start_table_from; i <= pThreadInfo->end_table_to; i++) {
     if (0 == g_Dbs.use_metric) {
       snprintf(buffer, buff_len,
               "create table if not exists %s.%s%d %s;",
-              winfo->db_name,
+              pThreadInfo->db_name,
               g_args.tb_prefix, i,
-              winfo->cols);
+              pThreadInfo->cols);
     } else {
       if (superTblInfo == NULL) {
         errorPrint("%s() LN%d, use metric, but super table info is NULL\n",
@@ -2658,8 +2661,8 @@ static void* createTable(void *sarg)
         len += snprintf(buffer + len,
                 buff_len - len,
                 "if not exists %s.%s%d using %s.%s tags %s ",
-                winfo->db_name, superTblInfo->childTblPrefix,
-                i, winfo->db_name,
+                pThreadInfo->db_name, superTblInfo->childTblPrefix,
+                i, pThreadInfo->db_name,
                 superTblInfo->sTblName, tagsValBuf);
         free(tagsValBuf);
         batchNum++;
@@ -2673,7 +2676,7 @@ static void* createTable(void *sarg)
 
     len = 0;
     verbosePrint("%s() LN%d %s\n", __func__, __LINE__, buffer);
-    if (0 != queryDbExec(winfo->taos, buffer, NO_INSERT_TYPE, false)){
+    if (0 != queryDbExec(pThreadInfo->taos, buffer, NO_INSERT_TYPE, false)){
       errorPrint( "queryDbExec() failed. buffer:\n%s\n", buffer);
       free(buffer);
       return NULL;
@@ -2682,14 +2685,14 @@ static void* createTable(void *sarg)
     int64_t  currentPrintTime = taosGetTimestampMs();
     if (currentPrintTime - lastPrintTime > 30*1000) {
       printf("thread[%d] already create %d - %d tables\n",
-              winfo->threadID, winfo->start_table_from, i);
+              pThreadInfo->threadID, pThreadInfo->start_table_from, i);
       lastPrintTime = currentPrintTime;
     }
   }
 
   if (0 != len) {
     verbosePrint("%s() %d buffer: %s\n", __func__, __LINE__, buffer);
-    if (0 != queryDbExec(winfo->taos, buffer, NO_INSERT_TYPE, false)) {
+    if (0 != queryDbExec(pThreadInfo->taos, buffer, NO_INSERT_TYPE, false)) {
       errorPrint( "queryDbExec() failed. buffer:\n%s\n", buffer);
     }
   }
@@ -3694,7 +3697,7 @@ static bool getMetaFromInsertJsonFile(cJSON* root) {
             __func__, __LINE__);
         goto PARSE_OVER;
       }
-
+/*
       cJSON *multiThreadWriteOneTbl =
           cJSON_GetObjectItem(stbInfo, "multi_thread_write_one_tbl"); // no , yes
       if (multiThreadWriteOneTbl
@@ -3711,7 +3714,7 @@ static bool getMetaFromInsertJsonFile(cJSON* root) {
         printf("ERROR: failed to read json, multiThreadWriteOneTbl not found\n");
         goto PARSE_OVER;
       }
-
+*/
       cJSON* interlaceRows = cJSON_GetObjectItem(stbInfo, "interlace_rows");
       if (interlaceRows && interlaceRows->type == cJSON_Number) {
         g_Dbs.db[i].superTbls[j].interlaceRows = interlaceRows->valueint;
@@ -5159,45 +5162,45 @@ free_and_statistics_2:
 
 static void* syncWrite(void *sarg) {
 
-  threadInfo *winfo = (threadInfo *)sarg;
-  SSuperTable* superTblInfo = winfo->superTblInfo;
+  threadInfo *pThreadInfo = (threadInfo *)sarg;
+  SSuperTable* superTblInfo = pThreadInfo->superTblInfo;
 
   int interlaceRows = superTblInfo?superTblInfo->interlaceRows:g_args.interlace_rows;
 
   if (interlaceRows > 0) {
     // interlace mode
-    return syncWriteInterlace(winfo);
+    return syncWriteInterlace(pThreadInfo);
   } else {
     // progressive mode
-    return syncWriteProgressive(winfo);
+    return syncWriteProgressive(pThreadInfo);
   }
 }
 
 static void callBack(void *param, TAOS_RES *res, int code) {
-  threadInfo* winfo = (threadInfo*)param;
-  SSuperTable* superTblInfo = winfo->superTblInfo;
+  threadInfo* pThreadInfo = (threadInfo*)param;
+  SSuperTable* superTblInfo = pThreadInfo->superTblInfo;
 
   int insert_interval =
       superTblInfo?superTblInfo->insertInterval:g_args.insert_interval;
   if (insert_interval) {
-    winfo->et = taosGetTimestampUs();
-    if (((winfo->et - winfo->st)/1000) < insert_interval) {
-      taosMsleep(insert_interval - (winfo->et - winfo->st)/1000); // ms
+    pThreadInfo->et = taosGetTimestampUs();
+    if (((pThreadInfo->et - pThreadInfo->st)/1000) < insert_interval) {
+      taosMsleep(insert_interval - (pThreadInfo->et - pThreadInfo->st)/1000); // ms
     }
   }
 
-  char *buffer = calloc(1, winfo->superTblInfo->maxSqlLen);
+  char *buffer = calloc(1, pThreadInfo->superTblInfo->maxSqlLen);
   char data[MAX_DATA_SIZE];
   char *pstr = buffer;
-  pstr += sprintf(pstr, "insert into %s.%s%d values", winfo->db_name, winfo->tb_prefix,
-          winfo->start_table_from);
-//  if (winfo->counter >= winfo->superTblInfo->insertRows) {
-  if (winfo->counter >= g_args.num_of_RPR) {
-    winfo->start_table_from++;
-    winfo->counter = 0;
+  pstr += sprintf(pstr, "insert into %s.%s%d values", pThreadInfo->db_name, pThreadInfo->tb_prefix,
+          pThreadInfo->start_table_from);
+//  if (pThreadInfo->counter >= pThreadInfo->superTblInfo->insertRows) {
+  if (pThreadInfo->counter >= g_args.num_of_RPR) {
+    pThreadInfo->start_table_from++;
+    pThreadInfo->counter = 0;
   }
-  if (winfo->start_table_from > winfo->end_table_to) {
-    tsem_post(&winfo->lock_sem);
+  if (pThreadInfo->start_table_from > pThreadInfo->end_table_to) {
+    tsem_post(&pThreadInfo->lock_sem);
     free(buffer);
     taos_free_result(res);
     return;
@@ -5205,46 +5208,46 @@ static void callBack(void *param, TAOS_RES *res, int code) {
 
   for (int i = 0; i < g_args.num_of_RPR; i++) {
     int rand_num = taosRandom() % 100;
-    if (0 != winfo->superTblInfo->disorderRatio
-            && rand_num < winfo->superTblInfo->disorderRatio) {
-      int64_t d = winfo->lastTs - (taosRandom() % winfo->superTblInfo->disorderRange + 1);
-      generateRowData(data, d, winfo->superTblInfo);
+    if (0 != pThreadInfo->superTblInfo->disorderRatio
+            && rand_num < pThreadInfo->superTblInfo->disorderRatio) {
+      int64_t d = pThreadInfo->lastTs - (taosRandom() % pThreadInfo->superTblInfo->disorderRange + 1);
+      generateRowData(data, d, pThreadInfo->superTblInfo);
     } else {
-      generateRowData(data, winfo->lastTs += 1000, winfo->superTblInfo);
+      generateRowData(data, pThreadInfo->lastTs += 1000, pThreadInfo->superTblInfo);
     }
     pstr += sprintf(pstr, "%s", data);
-    winfo->counter++;
+    pThreadInfo->counter++;
 
-    if (winfo->counter >= winfo->superTblInfo->insertRows) {
+    if (pThreadInfo->counter >= pThreadInfo->superTblInfo->insertRows) {
       break;
     }
   }
 
   if (insert_interval) {
-    winfo->st = taosGetTimestampUs();
+    pThreadInfo->st = taosGetTimestampUs();
   }
-  taos_query_a(winfo->taos, buffer, callBack, winfo);
+  taos_query_a(pThreadInfo->taos, buffer, callBack, pThreadInfo);
   free(buffer);
 
   taos_free_result(res);
 }
 
 static void *asyncWrite(void *sarg) {
-  threadInfo *winfo = (threadInfo *)sarg;
-  SSuperTable* superTblInfo = winfo->superTblInfo;
+  threadInfo *pThreadInfo = (threadInfo *)sarg;
+  SSuperTable* superTblInfo = pThreadInfo->superTblInfo;
 
-  winfo->st = 0;
-  winfo->et = 0;
-  winfo->lastTs = winfo->start_time;
+  pThreadInfo->st = 0;
+  pThreadInfo->et = 0;
+  pThreadInfo->lastTs = pThreadInfo->start_time;
 
   int insert_interval =
       superTblInfo?superTblInfo->insertInterval:g_args.insert_interval;
   if (insert_interval) {
-    winfo->st = taosGetTimestampUs();
+    pThreadInfo->st = taosGetTimestampUs();
   }
-  taos_query_a(winfo->taos, "show databases", callBack, winfo);
+  taos_query_a(pThreadInfo->taos, "show databases", callBack, pThreadInfo);
 
-  tsem_wait(&(winfo->lock_sem));
+  tsem_wait(&(pThreadInfo->lock_sem));
 
   return NULL;
 }
@@ -5774,10 +5777,10 @@ static int insertTestProcess() {
   return 0;
 }
 
-static void *specifiedQueryProcess(void *sarg) {
-  threadInfo *winfo = (threadInfo *)sarg;
+static void *specifiedTableQuery(void *sarg) {
+  threadInfo *pThreadInfo = (threadInfo *)sarg;
 
-  if (winfo->taos == NULL) {
+  if (pThreadInfo->taos == NULL) {
     TAOS * taos = NULL;
     taos = taos_connect(g_queryInfo.host,
           g_queryInfo.user,
@@ -5786,17 +5789,17 @@ static void *specifiedQueryProcess(void *sarg) {
           g_queryInfo.port);
     if (taos == NULL) {
       errorPrint("[%d] Failed to connect to TDengine, reason:%s\n",
-            winfo->threadID, taos_errstr(NULL));
+            pThreadInfo->threadID, taos_errstr(NULL));
       return NULL;
     } else {
-      winfo->taos = taos;
+      pThreadInfo->taos = taos;
     }
   }
 
   char sqlStr[MAX_DB_NAME_SIZE + 5];
   sprintf(sqlStr, "use %s", g_queryInfo.dbName);
-  if (0 != queryDbExec(winfo->taos, sqlStr, NO_INSERT_TYPE, false)) {
-    taos_close(winfo->taos);
+  if (0 != queryDbExec(pThreadInfo->taos, sqlStr, NO_INSERT_TYPE, false)) {
+    taos_close(pThreadInfo->taos);
     errorPrint( "use database %s failed!\n\n",
                 g_queryInfo.dbName);
     return NULL;
@@ -5807,11 +5810,15 @@ static void *specifiedQueryProcess(void *sarg) {
 
   int queryTimes = g_queryInfo.specifiedQueryInfo.queryTimes;
 
+  int totalQueried = 0;
+  int64_t  lastPrintTime = taosGetTimestampMs();
+  int64_t  startTs = taosGetTimestampMs();
+
   while(queryTimes --) {
     if (g_queryInfo.specifiedQueryInfo.rate && (et - st) <
             (int64_t)g_queryInfo.specifiedQueryInfo.rate*1000) {
       taosMsleep(g_queryInfo.specifiedQueryInfo.rate*1000 - (et - st)); // ms
-      //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, winfo->start_table_from, winfo->end_table_to);
+      //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, pThreadInfo->start_table_from, pThreadInfo->end_table_to);
     }
 
     st = taosGetTimestampUs();
@@ -5819,13 +5826,13 @@ static void *specifiedQueryProcess(void *sarg) {
     if (0 == strncasecmp(g_queryInfo.queryMode, "taosc", 5)) {
       int64_t t1 = taosGetTimestampUs();
       char tmpFile[MAX_FILE_NAME_LEN*2] = {0};
-      if (g_queryInfo.specifiedQueryInfo.result[winfo->querySeq][0] != 0) {
+      if (g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq][0] != 0) {
         sprintf(tmpFile, "%s-%d",
-                g_queryInfo.specifiedQueryInfo.result[winfo->querySeq],
-                winfo->threadID);
+                g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq],
+                pThreadInfo->threadID);
       }
-      selectAndGetResult(winfo->taos,
-          g_queryInfo.specifiedQueryInfo.sql[winfo->querySeq], tmpFile);
+      selectAndGetResult(pThreadInfo->taos,
+          g_queryInfo.specifiedQueryInfo.sql[pThreadInfo->querySeq], tmpFile);
       int64_t t2 = taosGetTimestampUs();
       printf("=[taosc] thread[%"PRId64"] complete one sql, Spent %f s\n",
               taosGetSelfPthreadId(), (t2 - t1)/1000000.0);
@@ -5833,25 +5840,37 @@ static void *specifiedQueryProcess(void *sarg) {
       int64_t t1 = taosGetTimestampUs();
       int retCode = postProceSql(g_queryInfo.host,
               g_queryInfo.port,
-              g_queryInfo.specifiedQueryInfo.sql[winfo->querySeq]);
+              g_queryInfo.specifiedQueryInfo.sql[pThreadInfo->querySeq]);
+      if (0 != retCode) {
+        printf("====restful return fail, threadID[%d]\n", pThreadInfo->threadID);
+        return NULL;
+      }
       int64_t t2 = taosGetTimestampUs();
       printf("=[restful] thread[%"PRId64"] complete one sql, Spent %f s\n",
               taosGetSelfPthreadId(), (t2 - t1)/1000000.0);
 
-      if (0 != retCode) {
-        printf("====restful return fail, threadID[%d]\n", winfo->threadID);
-        return NULL;
-      }
     }
+    totalQueried ++;
+    g_queryInfo.specifiedQueryInfo.totalQueried ++;
 
     et = taosGetTimestampUs();
     printf("==thread[%"PRId64"] complete all sqls to specify tables once queries duration:%.6fs\n\n",
             taosGetSelfPthreadId(), (double)(et - st)/1000.0);
+
+    int64_t  currentPrintTime = taosGetTimestampMs();
+    int64_t  endTs = taosGetTimestampMs();
+    if (currentPrintTime - lastPrintTime > 30*1000) {
+      printf("thread[%d] has currently completed queries: %d, QPS: %10.2f\n",
+                    pThreadInfo->threadID,
+                    totalQueried,
+                    totalQueried/((endTs-startTs)/1000.0));
+    }
+    lastPrintTime = currentPrintTime;
   }
   return NULL;
 }
 
-static void replaceSubTblName(char* inSql, char* outSql, int tblIndex) {
+static void replaceChildTblName(char* inSql, char* outSql, int tblIndex) {
   char sourceString[32] = "xxxx";
   char subTblName[MAX_TB_NAME_SIZE*3];
   sprintf(subTblName, "%s.%s",
@@ -5873,11 +5892,11 @@ static void replaceSubTblName(char* inSql, char* outSql, int tblIndex) {
   //printf("3: %s\n", outSql);
 }
 
-static void *superQueryProcess(void *sarg) {
+static void *superTableQuery(void *sarg) {
   char sqlstr[1024];
-  threadInfo *winfo = (threadInfo *)sarg;
+  threadInfo *pThreadInfo = (threadInfo *)sarg;
 
-  if (winfo->taos == NULL) {
+  if (pThreadInfo->taos == NULL) {
     TAOS * taos = NULL;
     taos = taos_connect(g_queryInfo.host,
           g_queryInfo.user,
@@ -5886,10 +5905,10 @@ static void *superQueryProcess(void *sarg) {
           g_queryInfo.port);
     if (taos == NULL) {
       errorPrint("[%d] Failed to connect to TDengine, reason:%s\n",
-            winfo->threadID, taos_errstr(NULL));
+            pThreadInfo->threadID, taos_errstr(NULL));
       return NULL;
     } else {
-      winfo->taos = taos;
+      pThreadInfo->taos = taos;
     }
   }
 
@@ -5897,33 +5916,49 @@ static void *superQueryProcess(void *sarg) {
   int64_t et = (int64_t)g_queryInfo.superQueryInfo.rate*1000;
 
   int queryTimes = g_queryInfo.superQueryInfo.queryTimes;
+  int totalQueried = 0;
+  int64_t  startTs = taosGetTimestampMs();
 
+  int64_t  lastPrintTime = taosGetTimestampMs();
   while(queryTimes --) {
     if (g_queryInfo.superQueryInfo.rate
             && (et - st) < (int64_t)g_queryInfo.superQueryInfo.rate*1000) {
       taosMsleep(g_queryInfo.superQueryInfo.rate*1000 - (et - st)); // ms
-      //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, winfo->start_table_from, winfo->end_table_to);
+      //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, pThreadInfo->start_table_from, pThreadInfo->end_table_to);
     }
 
     st = taosGetTimestampUs();
-    for (int i = winfo->start_table_from; i <= winfo->end_table_to; i++) {
+    for (int i = pThreadInfo->start_table_from; i <= pThreadInfo->end_table_to; i++) {
       for (int j = 0; j < g_queryInfo.superQueryInfo.sqlCount; j++) {
         memset(sqlstr,0,sizeof(sqlstr));
-        replaceSubTblName(g_queryInfo.superQueryInfo.sql[j], sqlstr, i);
+        replaceChildTblName(g_queryInfo.superQueryInfo.sql[j], sqlstr, i);
         char tmpFile[MAX_FILE_NAME_LEN*2] = {0};
         if (g_queryInfo.superQueryInfo.result[j][0] != 0) {
           sprintf(tmpFile, "%s-%d",
                   g_queryInfo.superQueryInfo.result[j],
-                  winfo->threadID);
+                  pThreadInfo->threadID);
         }
-        selectAndGetResult(winfo->taos, sqlstr, tmpFile);
+        selectAndGetResult(pThreadInfo->taos, sqlstr, tmpFile);
+
+        totalQueried++;
+        g_queryInfo.superQueryInfo.totalQueried ++;
+
+        int64_t  currentPrintTime = taosGetTimestampMs();
+        int64_t  endTs = taosGetTimestampMs();
+        if (currentPrintTime - lastPrintTime > 30*1000) {
+          printf("thread[%d] has currently completed queries: %d, QPS: %10.2f\n",
+                    pThreadInfo->threadID,
+                    totalQueried,
+                    totalQueried/((endTs-startTs)/1000.0));
+        }
+        lastPrintTime = currentPrintTime;
       }
     }
     et = taosGetTimestampUs();
     printf("####thread[%"PRId64"] complete all sqls to allocate all sub-tables[%d - %d] once queries duration:%.4fs\n\n",
             taosGetSelfPthreadId(),
-            winfo->start_table_from,
-            winfo->end_table_to,
+            pThreadInfo->start_table_from,
+            pThreadInfo->end_table_to,
             (double)(et - st)/1000000.0);
   }
 
@@ -5969,6 +6004,8 @@ static int queryTestProcess() {
   int nConcurrent = g_queryInfo.specifiedQueryInfo.concurrent;
   int nSqlCount = g_queryInfo.specifiedQueryInfo.sqlCount;
 
+  int64_t startTs = taosGetTimestampMs();
+
   if ((nSqlCount > 0) && (nConcurrent > 0)) {
 
     pids  = malloc(nConcurrent * nSqlCount * sizeof(pthread_t));
@@ -6002,7 +6039,7 @@ static int queryTestProcess() {
 
         t_info->taos = NULL;// TODO: workaround to use separate taos connection;
 
-        pthread_create(pids + i * nSqlCount + j, NULL, specifiedQueryProcess,
+        pthread_create(pids + i * nSqlCount + j, NULL, specifiedTableQuery,
             t_info);
       }
     }
@@ -6051,7 +6088,7 @@ static int queryTestProcess() {
       t_info->end_table_to = i < b ? startFrom + a : startFrom + a - 1;
       startFrom = t_info->end_table_to + 1;
       t_info->taos = NULL; // TODO: workaround to use separate taos connection;
-      pthread_create(pidsOfSub + i, NULL, superQueryProcess, t_info);
+      pthread_create(pidsOfSub + i, NULL, superTableQuery, t_info);
     }
 
     g_queryInfo.superQueryInfo.threadCnt = threads;
@@ -6078,6 +6115,14 @@ static int queryTestProcess() {
   tmfree((char*)infosOfSub);
 
 //  taos_close(taos);// TODO: workaround to use separate taos connection;
+  int64_t endTs = taosGetTimestampMs();
+
+  int totalQueried = g_queryInfo.specifiedQueryInfo.totalQueried +
+    g_queryInfo.superQueryInfo.totalQueried;
+
+  printf("==== completed total queries: %d, the QPS of all threads: %10.2f====\n",
+          totalQueried,
+          totalQueried/((endTs-startTs)/1000.0));
   return 0;
 }
 
@@ -6114,12 +6159,12 @@ static TAOS_SUB* subscribeImpl(TAOS *taos, char *sql, char* topic, char* resultF
   return tsub;
 }
 
-static void *subSubscribeProcess(void *sarg) {
-  threadInfo *winfo = (threadInfo *)sarg;
+static void *superSubscribe(void *sarg) {
+  threadInfo *pThreadInfo = (threadInfo *)sarg;
   char subSqlstr[1024];
   TAOS_SUB*    tsub[MAX_QUERY_SQL_COUNT] = {0};
 
-  if (winfo->taos == NULL) {
+  if (pThreadInfo->taos == NULL) {
     TAOS * taos = NULL;
     taos = taos_connect(g_queryInfo.host,
           g_queryInfo.user,
@@ -6128,17 +6173,17 @@ static void *subSubscribeProcess(void *sarg) {
           g_queryInfo.port);
     if (taos == NULL) {
       errorPrint("[%d] Failed to connect to TDengine, reason:%s\n",
-            winfo->threadID, taos_errstr(NULL));
+            pThreadInfo->threadID, taos_errstr(NULL));
       return NULL;
     } else {
-      winfo->taos = taos;
+      pThreadInfo->taos = taos;
     }
   }
 
   char sqlStr[MAX_TB_NAME_SIZE*2];
   sprintf(sqlStr, "use %s", g_queryInfo.dbName);
-  if (0 != queryDbExec(winfo->taos, sqlStr, NO_INSERT_TYPE, false)) {
-    taos_close(winfo->taos);
+  if (0 != queryDbExec(pThreadInfo->taos, sqlStr, NO_INSERT_TYPE, false)) {
+    taos_close(pThreadInfo->taos);
     errorPrint( "use database %s failed!\n\n",
                 g_queryInfo.dbName);
     return NULL;
@@ -6149,7 +6194,7 @@ static void *subSubscribeProcess(void *sarg) {
   do {
     //if (g_queryInfo.specifiedQueryInfo.rate && (et - st) < g_queryInfo.specifiedQueryInfo.rate*1000) {
     //  taosMsleep(g_queryInfo.specifiedQueryInfo.rate*1000 - (et - st)); // ms
-    //  //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, winfo->start_table_from, winfo->end_table_to);
+    //  //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, pThreadInfo->start_table_from, pThreadInfo->end_table_to);
     //}
 
     //st = taosGetTimestampMs();
@@ -6157,15 +6202,15 @@ static void *subSubscribeProcess(void *sarg) {
     for (int i = 0; i < g_queryInfo.superQueryInfo.sqlCount; i++) {
       sprintf(topic, "taosdemo-subscribe-%d", i);
       memset(subSqlstr,0,sizeof(subSqlstr));
-      replaceSubTblName(g_queryInfo.superQueryInfo.sql[i], subSqlstr, i);
+      replaceChildTblName(g_queryInfo.superQueryInfo.sql[i], subSqlstr, i);
       char tmpFile[MAX_FILE_NAME_LEN*2] = {0};
       if (g_queryInfo.superQueryInfo.result[i][0] != 0) {
         sprintf(tmpFile, "%s-%d",
-                g_queryInfo.superQueryInfo.result[i], winfo->threadID);
+                g_queryInfo.superQueryInfo.result[i], pThreadInfo->threadID);
       }
-      tsub[i] = subscribeImpl(winfo->taos, subSqlstr, topic, tmpFile);
+      tsub[i] = subscribeImpl(pThreadInfo->taos, subSqlstr, topic, tmpFile);
       if (NULL == tsub[i]) {
-        taos_close(winfo->taos);
+        taos_close(pThreadInfo->taos);
         return NULL;
       }
     }
@@ -6187,7 +6232,7 @@ static void *subSubscribeProcess(void *sarg) {
         if (g_queryInfo.superQueryInfo.result[i][0] != 0) {
           sprintf(tmpFile, "%s-%d",
                   g_queryInfo.superQueryInfo.result[i],
-                  winfo->threadID);
+                  pThreadInfo->threadID);
         }
         getResult(res, tmpFile);
       }
@@ -6199,15 +6244,15 @@ static void *subSubscribeProcess(void *sarg) {
     taos_unsubscribe(tsub[i], g_queryInfo.superQueryInfo.subscribeKeepProgress);
   }
 
-  taos_close(winfo->taos);
+  taos_close(pThreadInfo->taos);
   return NULL;
 }
 
-static void *superSubscribeProcess(void *sarg) {
-  threadInfo *winfo = (threadInfo *)sarg;
+static void *specifiedSubscribe(void *sarg) {
+  threadInfo *pThreadInfo = (threadInfo *)sarg;
   TAOS_SUB*    tsub[MAX_QUERY_SQL_COUNT] = {0};
 
-  if (winfo->taos == NULL) {
+  if (pThreadInfo->taos == NULL) {
     TAOS * taos = NULL;
     taos = taos_connect(g_queryInfo.host,
           g_queryInfo.user,
@@ -6216,18 +6261,18 @@ static void *superSubscribeProcess(void *sarg) {
           g_queryInfo.port);
     if (taos == NULL) {
       errorPrint("[%d] Failed to connect to TDengine, reason:%s\n",
-            winfo->threadID, taos_errstr(NULL));
+            pThreadInfo->threadID, taos_errstr(NULL));
       return NULL;
     } else {
-      winfo->taos = taos;
+      pThreadInfo->taos = taos;
     }
   }
 
   char sqlStr[MAX_TB_NAME_SIZE*2];
   sprintf(sqlStr, "use %s", g_queryInfo.dbName);
   debugPrint("%s() %d sqlStr: %s\n", __func__, __LINE__, sqlStr);
-  if (0 != queryDbExec(winfo->taos, sqlStr, NO_INSERT_TYPE, false)) {
-    taos_close(winfo->taos);
+  if (0 != queryDbExec(pThreadInfo->taos, sqlStr, NO_INSERT_TYPE, false)) {
+    taos_close(pThreadInfo->taos);
     return NULL;
   }
 
@@ -6236,7 +6281,7 @@ static void *superSubscribeProcess(void *sarg) {
   do {
     //if (g_queryInfo.specifiedQueryInfo.rate && (et - st) < g_queryInfo.specifiedQueryInfo.rate*1000) {
     //  taosMsleep(g_queryInfo.specifiedQueryInfo.rate*1000 - (et - st)); // ms
-    //  //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, winfo->start_table_from, winfo->end_table_to);
+    //  //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, pThreadInfo->start_table_from, pThreadInfo->end_table_to);
     //}
 
     //st = taosGetTimestampMs();
@@ -6246,12 +6291,12 @@ static void *superSubscribeProcess(void *sarg) {
       char tmpFile[MAX_FILE_NAME_LEN*2] = {0};
       if (g_queryInfo.superQueryInfo.result[i][0] != 0) {
         sprintf(tmpFile, "%s-%d",
-                g_queryInfo.specifiedQueryInfo.result[i], winfo->threadID);
+                g_queryInfo.specifiedQueryInfo.result[i], pThreadInfo->threadID);
       }
-      tsub[i] = subscribeImpl(winfo->taos,
+      tsub[i] = subscribeImpl(pThreadInfo->taos,
           g_queryInfo.specifiedQueryInfo.sql[i], topic, tmpFile);
       if (NULL == g_queryInfo.specifiedQueryInfo.tsub[i]) {
-        taos_close(winfo->taos);
+        taos_close(pThreadInfo->taos);
         return NULL;
       }
     }
@@ -6272,7 +6317,7 @@ static void *superSubscribeProcess(void *sarg) {
         char tmpFile[MAX_FILE_NAME_LEN*2] = {0};
         if (g_queryInfo.specifiedQueryInfo.result[i][0] != 0) {
           sprintf(tmpFile, "%s-%d",
-                  g_queryInfo.specifiedQueryInfo.result[i], winfo->threadID);
+                  g_queryInfo.specifiedQueryInfo.result[i], pThreadInfo->threadID);
         }
         getResult(res, tmpFile);
       }
@@ -6285,7 +6330,7 @@ static void *superSubscribeProcess(void *sarg) {
         g_queryInfo.specifiedQueryInfo.subscribeKeepProgress);
   }
 
-  taos_close(winfo->taos);
+  taos_close(pThreadInfo->taos);
   return NULL;
 }
 
@@ -6343,7 +6388,7 @@ static int subscribeTestProcess() {
       threadInfo *t_info = infos + i;
       t_info->threadID = i;
       t_info->taos = NULL;  // TODO: workaround to use separate taos connection;
-      pthread_create(pids + i, NULL, superSubscribeProcess, t_info);
+      pthread_create(pids + i, NULL, specifiedSubscribe, t_info);
   }
 
   //==== create sub threads for query from sub table
@@ -6386,7 +6431,7 @@ static int subscribeTestProcess() {
       t_info->end_table_to = i < b ? startFrom + a : startFrom + a - 1;
       startFrom = t_info->end_table_to + 1;
       t_info->taos = NULL; // TODO: workaround to use separate taos connection;
-      pthread_create(pidsOfSub + i, NULL, subSubscribeProcess, t_info);
+      pthread_create(pidsOfSub + i, NULL, superSubscribe, t_info);
     }
 
     g_queryInfo.superQueryInfo.threadCnt = threads;
diff --git a/tests/examples/JDBC/taosdemo/src/main/resources/insert.json b/tests/examples/JDBC/taosdemo/src/main/resources/insert.json
index 35c7773175..7578083d33 100644
--- a/tests/examples/JDBC/taosdemo/src/main/resources/insert.json
+++ b/tests/examples/JDBC/taosdemo/src/main/resources/insert.json
@@ -36,8 +36,6 @@
           "insert_mode": "taosc",
           "insert_rate": 0,
           "insert_rows": 100,
-          "multi_thread_write_one_tbl": "no",
-          "number_of_tbl_in_one_sql": 0,
           "interlace_rows": 3,
           "max_sql_len": 1024,
           "disorder_ratio": 0,
diff --git a/tests/pytest/cluster/clusterEnvSetup/insert.json b/tests/pytest/cluster/clusterEnvSetup/insert.json
index 4548ef74e6..2f3cf0f0d9 100644
--- a/tests/pytest/cluster/clusterEnvSetup/insert.json
+++ b/tests/pytest/cluster/clusterEnvSetup/insert.json
@@ -37,8 +37,6 @@
 			"insert_mode": "taosc",
 			"insert_rate": 0,
 			"insert_rows": 100000,
-			"multi_thread_write_one_tbl": "no",
-			"number_of_tbl_in_one_sql": 1,
 			"interlace_rows": 100,
 			"max_sql_len": 1024000,
 			"disorder_ratio": 0,
diff --git a/tests/pytest/perfbenchmark/bug3433.py b/tests/pytest/perfbenchmark/bug3433.py
index 80e9cce0dc..e4480df6b6 100644
--- a/tests/pytest/perfbenchmark/bug3433.py
+++ b/tests/pytest/perfbenchmark/bug3433.py
@@ -91,8 +91,6 @@ class TDTestCase:
             "data_source": "rand",
             "insert_mode": "taosc",
             "insert_rows": 1,
-            "multi_thread_write_one_tbl": "no",
-            "number_of_tbl_in_one_sql": 0,
             "rows_per_tbl": 1,
             "max_sql_len": 65480,
             "disorder_ratio": 0,
@@ -135,8 +133,6 @@ class TDTestCase:
             "data_source": "rand",
             "insert_mode": "taosc",
             "insert_rows": 1,
-            "multi_thread_write_one_tbl": "no",
-            "number_of_tbl_in_one_sql": 0,
             "rows_per_tbl": 1,
             "max_sql_len": 65480,
             "disorder_ratio": 0,
diff --git a/tests/pytest/query/query1970YearsAf.py b/tests/pytest/query/query1970YearsAf.py
index e66a79f5e6..93404afd59 100644
--- a/tests/pytest/query/query1970YearsAf.py
+++ b/tests/pytest/query/query1970YearsAf.py
@@ -82,8 +82,6 @@ class TDTestCase:
             "data_source": "rand",
             "insert_mode": "taosc",
             "insert_rows": 5000,
-            "multi_thread_write_one_tbl": "no",
-            "number_of_tbl_in_one_sql": 0,
             "rows_per_tbl": 1,
             "max_sql_len": 65480,
             "disorder_ratio": 0,
@@ -255,4 +253,4 @@ class TDTestCase:
 
 
 tdCases.addLinux(__file__, TDTestCase())
-tdCases.addWindows(__file__, TDTestCase())
\ No newline at end of file
+tdCases.addWindows(__file__, TDTestCase())
diff --git a/tests/pytest/tools/insert-interlace.json b/tests/pytest/tools/insert-interlace.json
index d4767ad064..a5c545d159 100644
--- a/tests/pytest/tools/insert-interlace.json
+++ b/tests/pytest/tools/insert-interlace.json
@@ -41,7 +41,6 @@
             "data_source": "rand",
             "insert_mode": "taosc",
             "insert_rows": 250,
-            "multi_thread_write_one_tbl": "no",
             "interlace_rows": 80,
             "max_sql_len": 1024000,
             "disorder_ratio": 0,
diff --git a/tests/pytest/tools/insert-tblimit-tboffset-createdb.json b/tests/pytest/tools/insert-tblimit-tboffset-createdb.json
index 8c3b39fb0b..9220bc1d17 100644
--- a/tests/pytest/tools/insert-tblimit-tboffset-createdb.json
+++ b/tests/pytest/tools/insert-tblimit-tboffset-createdb.json
@@ -40,8 +40,6 @@
             "data_source": "rand",
             "insert_mode": "taosc",
             "insert_rows": 0,
-            "multi_thread_write_one_tbl": "no",
-            "number_of_tbl_in_one_sql": 0,
             "max_sql_len": 1024000,
             "disorder_ratio": 0,
             "disorder_range": 1000,
diff --git a/tests/pytest/tools/insert-tblimit-tboffset-insertrec.json b/tests/pytest/tools/insert-tblimit-tboffset-insertrec.json
index a9efa7c31c..164d4fe8be 100644
--- a/tests/pytest/tools/insert-tblimit-tboffset-insertrec.json
+++ b/tests/pytest/tools/insert-tblimit-tboffset-insertrec.json
@@ -42,8 +42,6 @@
             "insert_rows": 1000,
             "childtable_limit": 33,
             "childtable_offset": 33,
-            "multi_thread_write_one_tbl": "no",
-            "number_of_tbl_in_one_sql": 0,
             "max_sql_len": 1024000,
             "disorder_ratio": 0,
             "disorder_range": 1000,
diff --git a/tests/pytest/tools/insert-tblimit-tboffset.json b/tests/pytest/tools/insert-tblimit-tboffset.json
index f3d3e864ba..0b8e0bd6c5 100644
--- a/tests/pytest/tools/insert-tblimit-tboffset.json
+++ b/tests/pytest/tools/insert-tblimit-tboffset.json
@@ -42,8 +42,6 @@
             "insert_rows": 1000,
             "childtable_limit": 33,
             "childtable_offset": 33,
-            "multi_thread_write_one_tbl": "no",
-            "number_of_tbl_in_one_sql": 0,
             "max_sql_len": 1024000,
             "disorder_ratio": 0,
             "disorder_range": 1000,
diff --git a/tests/pytest/tools/insert-tblimit-tboffset0.json b/tests/pytest/tools/insert-tblimit-tboffset0.json
index 6bf3783cb2..55d9e19055 100644
--- a/tests/pytest/tools/insert-tblimit-tboffset0.json
+++ b/tests/pytest/tools/insert-tblimit-tboffset0.json
@@ -42,8 +42,6 @@
             "insert_rows": 1000,
             "childtable_limit": 20,
             "childtable_offset": 0,
-            "multi_thread_write_one_tbl": "no",
-            "number_of_tbl_in_one_sql": 0,
             "max_sql_len": 1024000,
             "disorder_ratio": 0,
             "disorder_range": 1000,
diff --git a/tests/pytest/tools/insert-tblimit1-tboffset.json b/tests/pytest/tools/insert-tblimit1-tboffset.json
index 292902093d..3a88665661 100644
--- a/tests/pytest/tools/insert-tblimit1-tboffset.json
+++ b/tests/pytest/tools/insert-tblimit1-tboffset.json
@@ -42,8 +42,6 @@
             "insert_rows": 1000,
             "childtable_limit": 1,
             "childtable_offset": 50,
-            "multi_thread_write_one_tbl": "no",
-            "number_of_tbl_in_one_sql": 0,
             "max_sql_len": 1024000,
             "disorder_ratio": 0,
             "disorder_range": 1000,
diff --git a/tests/pytest/tools/taosdemo-sampledata.json b/tests/pytest/tools/taosdemo-sampledata.json
index d543b7e086..ce1c68b393 100644
--- a/tests/pytest/tools/taosdemo-sampledata.json
+++ b/tests/pytest/tools/taosdemo-sampledata.json
@@ -22,8 +22,6 @@
 			"insert_mode": "taosc",
 			"insert_rate": 0,
 			"insert_rows": 20,
-			"multi_thread_write_one_tbl": "no",
-			"number_of_tbl_in_one_sql": 0,
 			"max_sql_len": 1048000,
 			"timestamp_step": 1000,
 			"start_timestamp": "2020-1-1 00:00:00",
diff --git a/tests/pytest/tools/taosdemoPerformance.py b/tests/pytest/tools/taosdemoPerformance.py
index 8ce99d0969..b50180e2b3 100644
--- a/tests/pytest/tools/taosdemoPerformance.py
+++ b/tests/pytest/tools/taosdemoPerformance.py
@@ -49,8 +49,6 @@ class taosdemoPerformace:
             "batch_create_tbl_num": 10,
             "insert_mode": "taosc",
             "insert_rows": 100000,
-            "multi_thread_write_one_tbl": "no",
-            "number_of_tbl_in_one_sql": 0,
             "interlace_rows": 100,
             "max_sql_len": 1024000,
             "disorder_ratio": 0,

From db17a59b71cda7a2f46b34edc6efc0684858969f Mon Sep 17 00:00:00 2001
From: Minglei Jin <mljin@taosdata.com>
Date: Tue, 13 Apr 2021 20:24:59 +0800
Subject: [PATCH 10/10] [TD-3079]<enhance>: use multi-level storage with
 vnode's wal files

---
 src/vnode/src/vnodeMain.c | 39 ++++++++++++++++++++++++++++++++++++---
 1 file changed, 36 insertions(+), 3 deletions(-)

diff --git a/src/vnode/src/vnodeMain.c b/src/vnode/src/vnodeMain.c
index ee8ed9e0fa..0921c5ce48 100644
--- a/src/vnode/src/vnodeMain.c
+++ b/src/vnode/src/vnodeMain.c
@@ -232,9 +232,28 @@ int32_t vnodeAlter(void *vparam, SCreateVnodeMsg *pVnodeCfg) {
   return code;
 }
 
+static void vnodeFindWalRootDir(int32_t vgId, char *walRootDir) {
+  char vnodeDir[TSDB_FILENAME_LEN] = "\0";
+  snprintf(vnodeDir, TSDB_FILENAME_LEN, "/vnode/vnode%d/wal", vgId);
+
+  TDIR *tdir = tfsOpendir(vnodeDir);
+  if (!tdir) return;
+
+  const TFILE *tfile = tfsReaddir(tdir);
+  if (!tfile) {
+    tfsClosedir(tdir);
+    return;
+  }
+
+  sprintf(walRootDir, "%s/vnode/vnode%d", TFS_DISK_PATH(tfile->level, tfile->id), vgId);
+
+  tfsClosedir(tdir);
+}
+
 int32_t vnodeOpen(int32_t vgId) {
   char temp[TSDB_FILENAME_LEN * 3];
   char rootDir[TSDB_FILENAME_LEN * 2];
+  char walRootDir[TSDB_FILENAME_LEN * 2] = {0};
   snprintf(rootDir, TSDB_FILENAME_LEN * 2, "%s/vnode%d", tsVnodeDir, vgId);
 
   SVnodeObj *pVnode = calloc(sizeof(SVnodeObj), 1);
@@ -321,7 +340,21 @@ int32_t vnodeOpen(int32_t vgId) {
     }
   }
 
-  sprintf(temp, "%s/wal", rootDir);
+  // walRootDir for wal & syncInfo.path (not empty dir of /vnode/vnode{pVnode->vgId}/wal)
+  vnodeFindWalRootDir(pVnode->vgId, walRootDir);
+  if (walRootDir[0] == 0) {
+    int level = -1, id = -1;
+
+    tfsAllocDisk(TFS_PRIMARY_LEVEL, &level, &id);
+    if (level < 0 || id < 0) {
+      vnodeCleanUp(pVnode);
+      return terrno;
+    }
+
+    sprintf(walRootDir, "%s/vnode/vnode%d", TFS_DISK_PATH(level, id), vgId);
+  }
+
+  sprintf(temp, "%s/wal", walRootDir);
   pVnode->walCfg.vgId = pVnode->vgId;
   pVnode->wal = walOpen(temp, &pVnode->walCfg);
   if (pVnode->wal == NULL) { 
@@ -353,7 +386,7 @@ int32_t vnodeOpen(int32_t vgId) {
 
   pVnode->events = NULL;
 
-  vDebug("vgId:%d, vnode is opened in %s, pVnode:%p", pVnode->vgId, rootDir, pVnode);
+  vDebug("vgId:%d, vnode is opened in %s - %s, pVnode:%p", pVnode->vgId, rootDir, walRootDir, pVnode);
 
   vnodeAddIntoHash(pVnode);
   
@@ -361,7 +394,7 @@ int32_t vnodeOpen(int32_t vgId) {
   syncInfo.vgId = pVnode->vgId;
   syncInfo.version = pVnode->version;
   syncInfo.syncCfg = pVnode->syncCfg;
-  tstrncpy(syncInfo.path, rootDir, TSDB_FILENAME_LEN);
+  tstrncpy(syncInfo.path, walRootDir, TSDB_FILENAME_LEN);
   syncInfo.getWalInfoFp = vnodeGetWalInfo;
   syncInfo.writeToCacheFp = vnodeWriteToCache;
   syncInfo.confirmForward = vnodeConfirmForard;