Merge pull request #9737 from taosdata/feature/tq

add tmemmem

source/libs/wal/src/walMeta.c

@@ -13,7 +13,6 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#define _DEFAULT_SOURCE
 #include "cJSON.h"
 #include "os.h"
 #include "taoserror.h"
@@ -33,6 +32,24 @@ static inline int walBuildMetaName(SWal* pWal, int metaVer, char* buf) {
   return sprintf(buf, "%s/meta-ver%d", pWal->path, metaVer);
 }
 
+void* tmemmem(char* haystack, int hlen, char* needle, int nlen) {
+  char* limit;
+
+  if (nlen == 0 || hlen < nlen) {
+    return false;
+  }
+
+  limit = haystack + hlen - nlen + 1;
+  while ((haystack = (char*)memchr(
+              haystack, needle[0], limit - haystack)) != NULL) {
+    if (memcmp(haystack, needle, nlen) == 0) {
+      return haystack;
+    }
+    haystack++;
+  }
+  return NULL;
+}
+
 static inline int64_t walScanLogGetLastVer(SWal* pWal) {
   ASSERT(pWal->fileInfoSet != NULL);
   int sz = taosArrayGetSize(pWal->fileInfoSet);
@@ -47,7 +64,8 @@ static inline int64_t walScanLogGetLastVer(SWal* pWal) {
 
   struct stat statbuf;
   stat(fnameStr, &statbuf);
-  int readSize = MIN(WAL_MAX_SIZE, statbuf.st_size);
+  int readSize = MIN(WAL_MAX_SIZE + 2, statbuf.st_size);
+  pLastFileInfo->fileSize = statbuf.st_size;
 
   FileFd fd = taosOpenFileRead(fnameStr);
   if (fd < 0) {
@@ -64,6 +82,7 @@ static inline int64_t walScanLogGetLastVer(SWal* pWal) {
     return -1;
   }
 
+  taosLSeekFile(fd, -readSize, SEEK_END);
   if (readSize != taosReadFile(fd, buf, readSize)) {
     free(buf);
     taosCloseFile(fd);
@@ -71,21 +90,25 @@ static inline int64_t walScanLogGetLastVer(SWal* pWal) {
     return -1;
   }
   
-  char* found = strstr(buf, (const char*)&magic);
-  if (found == NULL) {
-    ASSERT(false);
-    // file has to be deleted
-    free(buf);
-    taosCloseFile(fd);
-    terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
-    return -1;
-  }
-  char *another;
-  while((another = strstr(found + 1, (const char*)&magic)) != NULL) {
+  char* haystack = buf;
+  char* found = NULL;
+  char *candidate;
+  while((candidate = tmemmem(haystack, readSize - (haystack - buf), (char*)&magic, sizeof(uint64_t))) != NULL) {
     // read and validate
-    SWalHead *logContent = (SWalHead*)another;
+    SWalHead *logContent = (SWalHead*)candidate;
     if (walValidHeadCksum(logContent) == 0 && walValidBodyCksum(logContent) == 0) {
-      found = another;
+      found = candidate;
+    }
+    haystack = candidate + 1;
+  }
+  if (found == buf) {
+    SWalHead *logContent = (SWalHead*)found;
+    if (walValidHeadCksum(logContent) != 0 || walValidBodyCksum(logContent) != 0) {
+      // file has to be deleted
+      free(buf);
+      taosCloseFile(fd);
+      terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
+      return -1;
     }
   }
   taosCloseFile(fd);
@@ -120,7 +143,6 @@ int walCheckAndRepairMeta(SWal* pWal) {
       SWalFileInfo fileInfo;
       memset(&fileInfo, -1, sizeof(SWalFileInfo));
       sscanf(name, "%" PRId64 ".log", &fileInfo.firstVer);
-      FileFd fd = taosOpenFileRead(ent->d_name);
       //get lastVer
       //get size
       taosArrayPush(pLogInfoArray, &fileInfo);
@@ -137,28 +159,25 @@ int walCheckAndRepairMeta(SWal* pWal) {
   }
   int newSz = taosArrayGetSize(pLogInfoArray);
   // case 1. meta file not exist / cannot be parsed
-  if (pWal->fileInfoSet == NULL && newSz != 0) {
-    // recover fileInfo set
-    pWal->fileInfoSet = pLogInfoArray;
-    if (newSz != 0) {
-    // recover meta version
-      pWal->vers.firstVer = ((SWalFileInfo*)taosArrayGet(pLogInfoArray, 0))->firstVer;
-      pWal->writeCur = newSz - 1;
-    }
-    // recover file size
-  } else if (oldSz < newSz) {
+  if (oldSz < newSz) {
     for (int i = oldSz; i < newSz; i++) {
       SWalFileInfo *pFileInfo = taosArrayGet(pLogInfoArray, i);
       taosArrayPush(pWal->fileInfoSet, pFileInfo);
     }
+
     pWal->writeCur = newSz - 1;
-  }
-  
-  if (pWal->fileInfoSet && taosArrayGetSize(pWal->fileInfoSet) != 0) {
+    pWal->vers.firstVer = ((SWalFileInfo*)taosArrayGet(pLogInfoArray, 0))->firstVer;
     pWal->vers.lastVer = walScanLogGetLastVer(pWal);
+    ((SWalFileInfo*)taosArrayGetLast(pWal->fileInfoSet))->lastVer = pWal->vers.lastVer;
     ASSERT(pWal->vers.lastVer != -1);
-  }
 
+    int code = walSaveMeta(pWal);
+    if (code < 0) {
+      taosArrayDestroy(pLogInfoArray);
+      return -1;
+    }
+  }
+  
   // case 2. versions in meta not match log 
   //         or some log not included in meta
   // (e.g. program killed)
@@ -182,14 +201,11 @@ int walCheckAndRepairMeta(SWal* pWal) {
   }
 #endif
 
-  int code = walSaveMeta(pWal);
-  if (code < 0) {
-    return -1;
-  }
 
   // get last version of this file
   //
   // rebuild meta
+  taosArrayDestroy(pLogInfoArray);
   return 0;
 }
 
@@ -397,6 +413,10 @@ int walLoadMeta(SWal* pWal) {
   }
   memset(buf, 0, size + 5);
   FileFd fd = taosOpenFileRead(fnameStr);
+  if (fd < 0) {
+    terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
+    return -1;
+  }
   if (taosReadFile(fd, buf, size) != size) {
     terrno = TAOS_SYSTEM_ERROR(errno);
     taosCloseFile(fd);

source/libs/wal/src/walMgmt.c

@@ -106,6 +106,7 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
   // init write buffer
   memset(&pWal->writeHead, 0, sizeof(SWalHead));
   pWal->writeHead.head.headVer = WAL_HEAD_VER;
+  pWal->writeHead.magic = WAL_MAGIC;
 
   if (pthread_mutex_init(&pWal->mutex, NULL) < 0) {
     taosArrayDestroy(pWal->fileInfoSet);
@@ -121,7 +122,9 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
     return NULL;
   }
 
-  if (walLoadMeta(pWal) < 0 && walCheckAndRepairMeta(pWal) < 0) {
+  walLoadMeta(pWal);
+
+  if (walCheckAndRepairMeta(pWal) < 0) {
     taosRemoveRef(tsWal.refSetId, pWal->refId);
     pthread_mutex_destroy(&pWal->mutex);
     taosArrayDestroy(pWal->fileInfoSet);
@@ -130,6 +133,7 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
   }
 
   if (walCheckAndRepairIdx(pWal) < 0) {
+
   }
 
   wDebug("vgId:%d, wal:%p is opened, level:%d fsyncPeriod:%d", pWal->cfg.vgId, pWal, pWal->cfg.level,

source/libs/wal/test/walMetaTest.cpp

@@ -340,7 +340,10 @@ TEST_F(WalRetentionEnv, repairMeta1) {
   char buf[100];
   sprintf(buf, "%s/meta-ver%d", pathName, 0);
   remove(buf);
+  sprintf(buf, "%s/meta-ver%d", pathName, 1);
+  remove(buf);
   SetUp();
+  //getchar();
 
   ASSERT_EQ(pWal->vers.lastVer, 99);
 
@@ -377,4 +380,26 @@ TEST_F(WalRetentionEnv, repairMeta1) {
     ASSERT_EQ(code, 0);
   }
 
+  for (int i = 0; i < 1000; i++) {
+    int ver = rand() % 200;
+    code = walReadWithHandle(pRead, ver);
+    ASSERT_EQ(code, 0);
+
+    // printf("rrbody: \n");
+    // for(int i = 0; i < pRead->pHead->head.len; i++) {
+    // printf("%d ", pRead->pHead->head.body[i]);
+    //}
+    // printf("\n");
+
+    ASSERT_EQ(pRead->pHead->head.version, ver);
+    ASSERT_EQ(pRead->curVersion, ver + 1);
+    char newStr[100];
+    sprintf(newStr, "%s-%d", ranStr, ver);
+    int len = strlen(newStr);
+    ASSERT_EQ(pRead->pHead->head.len, len);
+    for (int j = 0; j < len; j++) {
+      EXPECT_EQ(newStr[j], pRead->pHead->head.body[j]);
+    }
+  }
+
 }

tests/script/sim/db/basic1.sim

@@ -85,7 +85,6 @@ if $data02 != 2 then
   return -1
 endi
 
-return
 system sh/exec.sh -n dnode1 -s stop -x SIGKILL
 system sh/exec.sh -n dnode1 -s start
 
@@ -104,4 +103,4 @@ if $rows != 2 then
   return -1
 endi
 
-system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file
+system sh/exec.sh -n dnode1 -s stop -x SIGINT