Merge pull request #4220 from taosdata/feature/wal

TD-2087

src/vnode/src/vnodeMain.c

@@ -315,7 +315,13 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
     pVnode->version = walGetVersion(pVnode->wal);
   }
 
-  tsdbSyncCommit(pVnode->tsdb);
+  code = tsdbSyncCommit(pVnode->tsdb);
+  if (code != 0) {
+    vError("vgId:%d, failed to commit after restore from wal since %s", pVnode->vgId, tstrerror(code));
+    vnodeCleanUp(pVnode);
+    return code;
+  }
+
   walRemoveAllOldFiles(pVnode->wal);
   walRenew(pVnode->wal);
 
@@ -412,6 +418,7 @@ void vnodeRelease(void *pVnodeRaw) {
   }
 
   if (pVnode->wal) {
+    walRemoveAllOldFiles(pVnode->wal);
     walClose(pVnode->wal);
     pVnode->wal = NULL;
   }
@@ -589,8 +596,12 @@ static int vnodeProcessTsdbStatus(void *arg, int status) {
   if (status == TSDB_STATUS_COMMIT_START) {
     pVnode->fversion = pVnode->version;
     vDebug("vgId:%d, start commit, fver:%" PRIu64 " vver:%" PRIu64, pVnode->vgId, pVnode->fversion, pVnode->version);
+    if (pVnode->status == TAOS_VN_STATUS_INIT) {
+      return 0;
+    } else {
       return walRenew(pVnode->wal);
     }
+  }
 
   if (status == TSDB_STATUS_COMMIT_OVER) {
     vDebug("vgId:%d, commit over, fver:%" PRIu64 " vver:%" PRIu64, pVnode->vgId, pVnode->fversion, pVnode->version);

src/wal/src/walMgmt.c

@@ -124,15 +124,7 @@ void walClose(void *handle) {
 
   SWal *pWal = handle;
   pthread_mutex_lock(&pWal->mutex);
-
   taosClose(pWal->fd);
-
-  if (pWal->keep != TAOS_WAL_KEEP) {
-    walRemoveAllOldFiles(pWal);
-  } else {
-    wDebug("vgId:%d, wal:%p file:%s, it is closed and kept", pWal->vgId, pWal, pWal->name);
-  }
-
   pthread_mutex_unlock(&pWal->mutex);
   taosRemoveRef(tsWal.refId, pWal->rid);
 }

src/wal/src/walWrite.c

@@ -67,6 +67,7 @@ void walRemoveOneOldFile(void *handle) {
   SWal *pWal = handle;
   if (pWal == NULL) return;
   if (pWal->keep == TAOS_WAL_KEEP) return;
+  if (pWal->fd <= 0) return;
 
   pthread_mutex_lock(&pWal->mutex);
 
@@ -91,6 +92,8 @@ void walRemoveAllOldFiles(void *handle) {
 
   SWal *  pWal = handle;
   int64_t fileId = -1;
+
+  pthread_mutex_lock(&pWal->mutex);
   while (walGetNextFile(pWal, &fileId) >= 0) {
     snprintf(pWal->name, sizeof(pWal->name), "%s/%s%" PRId64, pWal->path, WAL_PREFIX, fileId);
 
@@ -100,6 +103,7 @@ void walRemoveAllOldFiles(void *handle) {
       wInfo("vgId:%d, wal:%p file:%s, it is removed", pWal->vgId, pWal, pWal->name);
     }
   }
+  pthread_mutex_unlock(&pWal->mutex);
 }
 
 int32_t walWrite(void *handle, SWalHead *pHead) {