Merge pull request #4257 from taosdata/feature/wal

Feature/wal

src/dnode/src/dnodePeer.c

@@ -151,6 +151,13 @@ void dnodeCleanupClient() {
 }
 
 static void dnodeProcessRspFromDnode(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
+  if (dnodeGetRunStatus() != TSDB_RUN_STATUS_RUNING) {
+    if (pMsg == NULL || pMsg->pCont == NULL) return;
+    dDebug("msg:%p is ignored since dnode not running", pMsg);
+    rpcFreeCont(pMsg->pCont);
+    return;
+  }
+
   if (pMsg->msgType == TSDB_MSG_TYPE_DM_STATUS_RSP && pEpSet) {
     dnodeUpdateEpSetForPeer(pEpSet);
   }

src/inc/taoserror.h

@@ -201,6 +201,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_VND_NO_SUCH_FILE_OR_DIR,      0, 0x0507, "Missing da
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_OUT_OF_MEMORY,            0, 0x0508, "Out of memory")
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_APP_ERROR,                0, 0x0509, "Unexpected generic error in vnode")
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_INVALID_VRESION_FILE,     0, 0x050A, "Invalid version file")
+TAOS_DEFINE_ERROR(TSDB_CODE_VND_IS_FULL,                  0, 0x050B, "Vnode memory is full for commit is failed")
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_NOT_SYNCED,               0, 0x0511, "Database suspended")
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_NO_WRITE_AUTH,            0, 0x0512, "Write operation denied")
 

src/vnode/inc/vnodeInt.h

@@ -41,6 +41,7 @@ typedef struct {
   int8_t       status; 
   int8_t       role;   
   int8_t       accessState;
+  int8_t       isFull;
   uint64_t     version;   // current version 
   uint64_t     fversion;  // version on saved data file
   void        *wqueue;

src/vnode/src/vnodeMain.c

@@ -381,6 +381,7 @@ int32_t vnodeClose(int32_t vgId) {
 void vnodeRelease(void *pVnodeRaw) {
   if (pVnodeRaw == NULL) return;
   SVnodeObj *pVnode = pVnodeRaw;
+  int32_t    code = 0;
   int32_t    vgId = pVnode->vgId;
 
   int32_t refCount = atomic_sub_fetch_32(&pVnode->refCount, 1);
@@ -406,7 +407,7 @@ void vnodeRelease(void *pVnodeRaw) {
   }
 
   if (pVnode->tsdb) {
-    tsdbCloseRepo(pVnode->tsdb, 1);
+    code = tsdbCloseRepo(pVnode->tsdb, 1);
     pVnode->tsdb = NULL;
   }
 
@@ -418,7 +419,7 @@ void vnodeRelease(void *pVnodeRaw) {
   }
 
   if (pVnode->wal) {
-    walRemoveAllOldFiles(pVnode->wal);
+    if (code == 0) walRemoveAllOldFiles(pVnode->wal);
     walClose(pVnode->wal);
     pVnode->wal = NULL;
   }
@@ -594,7 +595,10 @@ static int vnodeProcessTsdbStatus(void *arg, int status, int eno) {
   SVnodeObj *pVnode = arg;
 
   if (eno != TSDB_CODE_SUCCESS) {
-    // TODO: deal with the error here
+    vError("vgId:%d, failed to commit since %s, fver:%" PRIu64 " vver:%" PRIu64, pVnode->vgId, tstrerror(eno),
+           pVnode->fversion, pVnode->version);
+    pVnode->isFull = 1;
+    return 0;
   }
 
   if (status == TSDB_STATUS_COMMIT_START) {
@@ -609,6 +613,7 @@ static int vnodeProcessTsdbStatus(void *arg, int status, int eno) {
 
   if (status == TSDB_STATUS_COMMIT_OVER) {
     vDebug("vgId:%d, commit over, fver:%" PRIu64 " vver:%" PRIu64, pVnode->vgId, pVnode->fversion, pVnode->version);
+    pVnode->isFull = 0;
     walRemoveOneOldFile(pVnode->wal);
     return vnodeSaveVersion(pVnode);
   }

src/vnode/src/vnodeWrite.c

@@ -119,6 +119,11 @@ static int32_t vnodeCheckWrite(void *param) {
     return TSDB_CODE_APP_NOT_READY;
   }
 
+  if (pVnode->isFull) {
+    vDebug("vgId:%d, vnode is full, refCount:%d", pVnode->vgId, pVnode->refCount);
+    return TSDB_CODE_VND_IS_FULL;
+  }
+
   return TSDB_CODE_SUCCESS;
 }