Merge pull request #4266 from taosdata/feature/wal

[TD-2014]<fix>: add flow ctrl to vnode 

src/dnode/inc/dnodeVWrite.h

@@ -24,8 +24,8 @@ int32_t dnodeInitVWrite();
 void    dnodeCleanupVWrite();
 void    dnodeDispatchToVWriteQueue(SRpcMsg *pMsg);
 void *  dnodeAllocVWriteQueue(void *pVnode);
-void    dnodeFreeVWriteQueue(void *wqueue);
-void    dnodeSendRpcVWriteRsp(void *pVnode, void *param, int32_t code);
+void    dnodeFreeVWriteQueue(void *pWqueue);
+void    dnodeSendRpcVWriteRsp(void *pVnode, void *pWrite, int32_t code);
 
 #ifdef __cplusplus
 }

src/dnode/src/dnodeVWrite.c

@@ -38,11 +38,11 @@ typedef struct {
 } SVWriteWorkerPool;
 
 static SVWriteWorkerPool tsVWriteWP;
-static void *dnodeProcessVWriteQueue(void *param);
+static void *dnodeProcessVWriteQueue(void *pWorker);
 
 int32_t dnodeInitVWrite() {
   tsVWriteWP.max = tsNumOfCores;
-  tsVWriteWP.worker = (SVWriteWorker *)tcalloc(sizeof(SVWriteWorker), tsVWriteWP.max);
+  tsVWriteWP.worker = tcalloc(sizeof(SVWriteWorker), tsVWriteWP.max);
   if (tsVWriteWP.worker == NULL) return -1;
   pthread_mutex_init(&tsVWriteWP.mutex, NULL);
 
@@ -162,13 +162,13 @@ void *dnodeAllocVWriteQueue(void *pVnode) {
   return queue;
 }
 
-void dnodeFreeVWriteQueue(void *wqueue) {
-  taosCloseQueue(wqueue);
+void dnodeFreeVWriteQueue(void *pWqueue) {
+  taosCloseQueue(pWqueue);
 }
 
-void dnodeSendRpcVWriteRsp(void *pVnode, void *param, int32_t code) {
-  if (param == NULL) return;
-  SVWriteMsg *pWrite = param;
+void dnodeSendRpcVWriteRsp(void *pVnode, void *wparam, int32_t code) {
+  if (wparam == NULL) return;
+  SVWriteMsg *pWrite = wparam;
 
   if (code < 0) pWrite->code = code;
   int32_t count = atomic_add_fetch_32(&pWrite->processedCount, 1);
@@ -183,13 +183,11 @@ void dnodeSendRpcVWriteRsp(void *pVnode, void *param, int32_t code) {
   };
 
   rpcSendResponse(&rpcRsp);
-  taosFreeQitem(pWrite);
-
-  vnodeRelease(pVnode);
+  vnodeFreeFromWQueue(pVnode, pWrite);
 }
 
-static void *dnodeProcessVWriteQueue(void *param) {
-  SVWriteWorker *pWorker = param;
+static void *dnodeProcessVWriteQueue(void *wparam) {
+  SVWriteWorker *pWorker = wparam;
   SVWriteMsg *   pWrite;
   void *         pVnode;
   int32_t        numOfMsgs;
@@ -232,8 +230,7 @@ static void *dnodeProcessVWriteQueue(void *param) {
         if (pWrite->rspRet.rsp) {
           rpcFreeCont(pWrite->rspRet.rsp);
         }
-        taosFreeQitem(pWrite);
-        vnodeRelease(pVnode);
+        vnodeFreeFromWQueue(pVnode, pWrite);
       }
     }
   }

src/inc/dnode.h

@@ -54,8 +54,8 @@ void  dnodeSendMsgToDnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp, SRpcEpSet *epSet
 void *dnodeSendCfgTableToRecv(int32_t vgId, int32_t tid);
 
 void *dnodeAllocVWriteQueue(void *pVnode);
-void  dnodeFreeVWriteQueue(void *wqueue);
-void  dnodeSendRpcVWriteRsp(void *pVnode, void *param, int32_t code);
+void  dnodeFreeVWriteQueue(void *pWqueue);
+void  dnodeSendRpcVWriteRsp(void *pVnode, void *pWrite, int32_t code);
 void *dnodeAllocVReadQueue(void *pVnode);
 void  dnodeFreeVReadQueue(void *rqueue);
 

src/inc/vnode.h

@@ -70,11 +70,12 @@ void*   vnodeAcquire(int32_t vgId);        // add refcount
 void    vnodeRelease(void *pVnode);        // dec refCount
 void*   vnodeGetWal(void *pVnode);
 
-int32_t vnodeWriteToWQueue(void *vparam, void *wparam, int32_t qtype, void *rparam);
-int32_t vnodeProcessWrite(void *vparam, void *wparam, int32_t qtype, void *rparam);
+int32_t vnodeWriteToWQueue(void *pVnode, void *pHead, int32_t qtype, void *pRpcMsg);
+void    vnodeFreeFromWQueue(void *pVnode, SVWriteMsg *pWrite);
+int32_t vnodeProcessWrite(void *pVnode, void *pHead, int32_t qtype, void *pRspRet);
 int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes);
-void    vnodeBuildStatusMsg(void *param);
-void    vnodeConfirmForward(void *param, uint64_t version, int32_t code);
+void    vnodeBuildStatusMsg(void *pStatus);
+void    vnodeConfirmForward(void *pVnode, uint64_t version, int32_t code);
 void    vnodeSetAccess(SVgroupAccess *pAccess, int32_t numOfVnodes);
 
 int32_t vnodeInitResources();

src/os/src/detail/osTimer.c

@@ -111,6 +111,9 @@ void taosUninitTimer() {
   pthread_sigmask(SIG_BLOCK, &set, NULL);
 */
 void taosMsleep(int mseconds) {
+#if 1
+  usleep(mseconds * 1000);
+#else
   struct timeval timeout;
   int            seconds, useconds;
 
@@ -126,7 +129,8 @@ void taosMsleep(int mseconds) {
 
   select(0, NULL, NULL, NULL, &timeout);
 
-  /* pthread_sigmask(SIG_UNBLOCK, &set, NULL); */
+/* pthread_sigmask(SIG_UNBLOCK, &set, NULL); */
+#endif
 }
 
 #endif
\ No newline at end of file

src/plugins/http/src/httpServer.c

@@ -85,7 +85,7 @@ static void httpProcessHttpData(void *param) {
   while (1) {
     struct epoll_event events[HTTP_MAX_EVENTS];
     //-1 means uncertainty, 0-nowait, 1-wait 1 ms, set it from -1 to 1
-    fdNum = epoll_wait(pThread->pollFd, events, HTTP_MAX_EVENTS, 1);
+    fdNum = epoll_wait(pThread->pollFd, events, HTTP_MAX_EVENTS, TAOS_EPOLL_WAIT_TIME);
     if (pThread->stop) {
       httpDebug("%p, http thread get stop event, exiting...", pThread);
       break;

src/vnode/inc/vnodeInt.h

@@ -37,6 +37,7 @@ extern int32_t vDebugFlag;
 typedef struct {
   int32_t      vgId;      // global vnode group ID
   int32_t      refCount;  // reference count
+  int32_t      queuedMsg;
   int32_t      delay;
   int8_t       status; 
   int8_t       role;   

src/vnode/src/vnodeWrite.c

@@ -28,13 +28,15 @@
 #include "syncInt.h"
 #include "tcq.h"
 
-static int32_t (*vnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MAX])(SVnodeObj *, void *, SRspRet *);
-static int32_t vnodeProcessSubmitMsg(SVnodeObj *pVnode, void *pMsg, SRspRet *);
-static int32_t vnodeProcessCreateTableMsg(SVnodeObj *pVnode, void *pMsg, SRspRet *);
-static int32_t vnodeProcessDropTableMsg(SVnodeObj *pVnode, void *pMsg, SRspRet *);
-static int32_t vnodeProcessAlterTableMsg(SVnodeObj *pVnode, void *pMsg, SRspRet *);
-static int32_t vnodeProcessDropStableMsg(SVnodeObj *pVnode, void *pMsg, SRspRet *);
-static int32_t vnodeProcessUpdateTagValMsg(SVnodeObj *pVnode, void *pCont, SRspRet *pRet);
+#define MAX_QUEUED_MSG_NUM 10000
+
+static int32_t (*vnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MAX])(SVnodeObj *, void *pCont, SRspRet *);
+static int32_t vnodeProcessSubmitMsg(SVnodeObj *pVnode, void *pCont, SRspRet *);
+static int32_t vnodeProcessCreateTableMsg(SVnodeObj *pVnode, void *pCont, SRspRet *);
+static int32_t vnodeProcessDropTableMsg(SVnodeObj *pVnode, void *pCont, SRspRet *);
+static int32_t vnodeProcessAlterTableMsg(SVnodeObj *pVnode, void *pCont, SRspRet *);
+static int32_t vnodeProcessDropStableMsg(SVnodeObj *pVnode, void *pCont, SRspRet *);
+static int32_t vnodeProcessUpdateTagValMsg(SVnodeObj *pVnode, void *pCont, SRspRet *);
 
 void vnodeInitWriteFp(void) {
   vnodeProcessWriteMsgFp[TSDB_MSG_TYPE_SUBMIT]          = vnodeProcessSubmitMsg;
@@ -100,8 +102,8 @@ int32_t vnodeProcessWrite(void *vparam, void *wparam, int32_t qtype, void *rpara
   return syncCode;
 }
 
-static int32_t vnodeCheckWrite(void *param) {
-  SVnodeObj *pVnode = param;
+static int32_t vnodeCheckWrite(void *vparam) {
+  SVnodeObj *pVnode = vparam;
   if (!(pVnode->accessState & TSDB_VN_WRITE_ACCCESS)) {
     vDebug("vgId:%d, no write auth, refCount:%d pVnode:%p", pVnode->vgId, pVnode->refCount, pVnode);
     return TSDB_CODE_VND_NO_WRITE_AUTH;
@@ -127,8 +129,8 @@ static int32_t vnodeCheckWrite(void *param) {
   return TSDB_CODE_SUCCESS;
 }
 
-void vnodeConfirmForward(void *param, uint64_t version, int32_t code) {
-  SVnodeObj *pVnode = (SVnodeObj *)param;
+void vnodeConfirmForward(void *vparam, uint64_t version, int32_t code) {
+  SVnodeObj *pVnode = vparam;
   syncConfirmForward(pVnode->sync, version, code);
 }
 
@@ -242,8 +244,25 @@ int32_t vnodeWriteToWQueue(void *vparam, void *wparam, int32_t qtype, void *rpar
   memcpy(pWrite->pHead, pHead, sizeof(SWalHead) + pHead->len);
 
   atomic_add_fetch_32(&pVnode->refCount, 1);
-  vTrace("vgId:%d, get vnode wqueue, refCount:%d pVnode:%p", pVnode->vgId, pVnode->refCount, pVnode);
+
+  int32_t queued = atomic_add_fetch_32(&pVnode->queuedMsg, 1);
+  if (queued > MAX_QUEUED_MSG_NUM) {
+    vDebug("vgId:%d, too many msg:%d in vwqueue, flow control", pVnode->vgId, queued);
+    taosMsleep(1);
+  }
+
+  vTrace("vgId:%d, write into vwqueue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount, pVnode->queuedMsg);
 
   taosWriteQitem(pVnode->wqueue, qtype, pWrite);
   return TSDB_CODE_SUCCESS;
 }
+
+void vnodeFreeFromWQueue(void *vparam, SVWriteMsg *pWrite) {
+  SVnodeObj *pVnode = vparam;
+
+  atomic_sub_fetch_32(&pVnode->queuedMsg, 1);
+  vTrace("vgId:%d, free from vwqueue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount, pVnode->queuedMsg);
+
+  taosFreeQitem(pWrite);
+  vnodeRelease(pVnode);
+}