TD-1918

src/inc/vnode.h

@@ -63,11 +63,9 @@ int32_t vnodeClose(int32_t vgId);
 
 void*   vnodeAcquire(int32_t vgId);        // add refcount
 void*   vnodeAcquireRqueue(int32_t vgId);  // add refCount, get read queue 
-void*   vnodeAcquireWqueue(int32_t vgId);  // add recCount, get write queue
 void    vnodeRelease(void *pVnode);        // dec refCount
 void*   vnodeGetWal(void *pVnode);
 
-
 int32_t vnodeWriteToQueue(void *vparam, void *wparam, int32_t qtype, void *pMsg);
 int32_t vnodeProcessWrite(void *param, int32_t qtype, SVWriteMsg *pWrite);
 int32_t vnodeCheckWrite(void *pVnode);

src/mnode/src/mnodeSdb.c

@@ -98,8 +98,8 @@ static taos_qall  tsSdbWriteQall;
 static taos_queue tsSdbWriteQueue;
 static SSdbWriteWorkerPool tsSdbPool;
 
-static int     sdbWrite(void *param, void *data, int type);
-static int     sdbWriteToQueue(void *param, void *data, int type);
+static int32_t sdbWrite(void *param, void *data, int32_t type, void *pMsg);
+static int32_t sdbWriteToQueue(void *param, void *data, int32_t type, void *pMsg);
 static void *  sdbWorkerFp(void *param);
 static int32_t sdbInitWriteWorker();
 static void    sdbCleanupWriteWorker();
@@ -575,7 +575,7 @@ static int32_t sdbUpdateHash(SSdbTable *pTable, SSdbOper *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int sdbWrite(void *param, void *data, int type) {
+static int sdbWrite(void *param, void *data, int32_t type, void *pMsg) {
   SSdbOper *pOper = param;
   SWalHead *pHead = data;
   int32_t tableId = pHead->msgType / 10;
@@ -1040,13 +1040,13 @@ void sdbFreeWritequeue() {
   tsSdbWriteQueue = NULL;
 }
 
-int sdbWriteToQueue(void *param, void *data, int type) {
+int32_t sdbWriteToQueue(void *param, void *data, int32_t qtype, void *pMsg) {
   SWalHead *pHead = data;
-  int size = sizeof(SWalHead) + pHead->len;
+  int32_t   size = sizeof(SWalHead) + pHead->len;
   SWalHead *pWal = (SWalHead *)taosAllocateQitem(size);
   memcpy(pWal, pHead, size);
 
-  taosWriteQitem(tsSdbWriteQueue, type, pWal);
+  taosWriteQitem(tsSdbWriteQueue, qtype, pWal);
   return 0;
 }
 
@@ -1081,7 +1081,7 @@ static void *sdbWorkerFp(void *param) {
         pOper = NULL;
       }
 
-      int32_t code = sdbWrite(pOper, pHead, type);
+      int32_t code = sdbWrite(pOper, pHead, type, NULL);
       if (code > 0) code = 0;
       if (pOper) {
         pOper->retCode = code;

src/vnode/src/vnodeMain.c

@@ -483,21 +483,6 @@ void *vnodeAcquireRqueue(int32_t vgId) {
   return pVnode->rqueue;
 }
 
-void *vnodeAcquireWqueue(int32_t vgId) {
-  SVnodeObj *pVnode = vnodeAcquire(vgId);
-  if (pVnode == NULL) return NULL;
-
-  int32_t code = vnodeCheckWrite(pVnode);
-  if (code != TSDB_CODE_SUCCESS) {
-    terrno = code;
-    vInfo("vgId:%d, can not provide write service, status is %s", vgId, vnodeStatus[pVnode->status]);
-    vnodeRelease(pVnode);
-    return NULL;
-  }
-
-  return pVnode->wqueue;
-}
-
 void *vnodeGetWal(void *pVnode) {
   return ((SVnodeObj *)pVnode)->wal;
 }

src/vnode/src/vnodeWrite.c

@@ -208,8 +208,9 @@ int32_t vnodeWriteToQueue(void *vparam, void *wparam, int32_t qtype, void *pMsg)
   SVnodeObj *pVnode = vparam;
   SWalHead * pHead = wparam;
 
-  if (qtype == TAOS_QTYPE_RPC && vnodeCheckWrite(pVnode) != TSDB_CODE_SUCCESS) {
-    return TSDB_CODE_VND_INVALID_VGROUP_ID;
+  if (qtype == TAOS_QTYPE_RPC) {
+    int32_t code = vnodeCheckWrite(pVnode);
+    if (code != TSDB_CODE_SUCCESS) return code;
   }
 
   int32_t size = sizeof(SVWriteMsg) + sizeof(SWalHead) + pHead->len;