Merge pull request #12354 from taosdata/fix/hzcheng_3.0

fix: concurrent r/w meta

source/dnode/vnode/src/inc/meta.h

@@ -37,6 +37,9 @@ typedef struct SMSmaCursor SMSmaCursor;
 // clang-format on
 
 // metaOpen ==================
+int32_t metaRLock(SMeta* pMeta);
+int32_t metaWLock(SMeta* pMeta);
+int32_t metaULock(SMeta* pMeta);
 
 // metaEntry ==================
 int metaEncodeEntry(SEncoder* pCoder, const SMetaEntry* pME);
@@ -57,6 +60,8 @@ int  metaRemoveTableFromIdx(SMeta* pMeta, tb_uid_t uid);
 static FORCE_INLINE tb_uid_t metaGenerateUid(SMeta* pMeta) { return tGenIdPI64(); }
 
 struct SMeta {
+  TdThreadRwlock lock;
+
   char*     path;
   SVnode*   pVnode;
   TENV*     pEnv;

source/dnode/vnode/src/meta/metaOpen.c

@@ -22,6 +22,9 @@ static int tagIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kL
 static int ttlIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2);
 static int uidIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2);
 
+static int32_t metaInitLock(SMeta *pMeta) { return taosThreadRwlockInit(&pMeta->lock, NULL); }
+static int32_t metaDestroyLock(SMeta *pMeta) { return taosThreadRwlockDestroy(&pMeta->lock); }
+
 int metaOpen(SVnode *pVnode, SMeta **ppMeta) {
   SMeta *pMeta = NULL;
   int    ret;
@@ -36,6 +39,7 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta) {
     return -1;
   }
 
+  metaInitLock(pMeta);
   pMeta->path = (char *)&pMeta[1];
   sprintf(pMeta->path, "%s%s%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path, TD_DIRSEP,
           VNODE_META_DIR);
@@ -121,6 +125,7 @@ _err:
   if (pMeta->pSkmDb) tdbDbClose(pMeta->pSkmDb);
   if (pMeta->pTbDb) tdbDbClose(pMeta->pTbDb);
   if (pMeta->pEnv) tdbEnvClose(pMeta->pEnv);
+  metaDestroyLock(pMeta);
   taosMemoryFree(pMeta);
   return -1;
 }
@@ -136,12 +141,19 @@ int metaClose(SMeta *pMeta) {
     if (pMeta->pSkmDb) tdbDbClose(pMeta->pSkmDb);
     if (pMeta->pTbDb) tdbDbClose(pMeta->pTbDb);
     if (pMeta->pEnv) tdbEnvClose(pMeta->pEnv);
+    metaDestroyLock(pMeta);
     taosMemoryFree(pMeta);
   }
 
   return 0;
 }
 
+int32_t metaRLock(SMeta *pMeta) { return taosThreadRwlockRdlock(&pMeta->lock); }
+
+int32_t metaWLock(SMeta *pMeta) { return taosThreadRwlockWrlock(&pMeta->lock); }
+
+int32_t metaULock(SMeta *pMeta) { return taosThreadRwlockUnlock(&pMeta->lock); }
+
 static int tbDbKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2) {
   STbDbKey *pTbDbKey1 = (STbDbKey *)pKey1;
   STbDbKey *pTbDbKey2 = (STbDbKey *)pKey2;

source/dnode/vnode/src/meta/metaQuery.c

@@ -19,9 +19,13 @@ void metaReaderInit(SMetaReader *pReader, SMeta *pMeta, int32_t flags) {
   memset(pReader, 0, sizeof(*pReader));
   pReader->flags = flags;
   pReader->pMeta = pMeta;
+  metaRLock(pMeta);
 }
 
 void metaReaderClear(SMetaReader *pReader) {
+  if (pReader->pMeta) {
+    metaULock(pReader->pMeta);
+  }
   tDecoderClear(&pReader->coder);
   tdbFree(pReader->pBuf);
 }

source/dnode/vnode/src/meta/metaTable.c

@@ -439,29 +439,36 @@ _exit:
 }
 
 static int metaHandleEntry(SMeta *pMeta, const SMetaEntry *pME) {
+  metaWLock(pMeta);
+
   // save to table.db
-  if (metaSaveToTbDb(pMeta, pME) < 0) return -1;
+  if (metaSaveToTbDb(pMeta, pME) < 0) goto _err;
 
   // update uid.idx
-  if (metaUpdateUidIdx(pMeta, pME) < 0) return -1;
+  if (metaUpdateUidIdx(pMeta, pME) < 0) goto _err;
 
   // update name.idx
-  if (metaUpdateNameIdx(pMeta, pME) < 0) return -1;
+  if (metaUpdateNameIdx(pMeta, pME) < 0) goto _err;
 
   if (pME->type == TSDB_CHILD_TABLE) {
     // update ctb.idx
-    if (metaUpdateCtbIdx(pMeta, pME) < 0) return -1;
+    if (metaUpdateCtbIdx(pMeta, pME) < 0) goto _err;
 
     // update tag.idx
-    if (metaUpdateTagIdx(pMeta, pME) < 0) return -1;
+    if (metaUpdateTagIdx(pMeta, pME) < 0) goto _err;
   } else {
     // update schema.db
-    if (metaSaveToSkmDb(pMeta, pME) < 0) return -1;
+    if (metaSaveToSkmDb(pMeta, pME) < 0) goto _err;
   }
 
   if (pME->type != TSDB_SUPER_TABLE) {
-    if (metaUpdateTtlIdx(pMeta, pME) < 0) return -1;
+    if (metaUpdateTtlIdx(pMeta, pME) < 0) goto _err;
   }
 
+  metaULock(pMeta);
   return 0;
+
+_err:
+  metaULock(pMeta);
+  return -1;
 }

source/dnode/vnode/src/tsdb/tsdbRead.c

@@ -3870,6 +3870,7 @@ int32_t tsdbQuerySTableByTagCond(void* pMeta, uint64_t uid, TSKEY skey, const ch
 
   if (metaGetTableEntryByUid(&mr, uid) < 0) {
     tsdbError("%p failed to get stable, uid:%" PRIu64 ", TID:0x%" PRIx64 " QID:0x%" PRIx64, pMeta, uid, taskId, reqId);
+    metaReaderClear(&mr);
     terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;
     goto _error;
   } else {
@@ -3880,6 +3881,7 @@ int32_t tsdbQuerySTableByTagCond(void* pMeta, uint64_t uid, TSKEY skey, const ch
     tsdbError("%p query normal tag not allowed, uid:%" PRIu64 ", TID:0x%" PRIx64 " QID:0x%" PRIx64, pMeta, uid, taskId,
               reqId);
     terrno = TSDB_CODE_OPS_NOT_SUPPORT;  // basically, this error is caused by invalid sql issued by client
+    metaReaderClear(&mr);
     goto _error;
   }