refactor: refactor the lock type.

include/libs/stream/tstream.h

@@ -497,7 +497,7 @@ typedef struct SStreamMeta {
   int32_t         role;
   bool            sendMsgBeforeClosing; // send hb to mnode before close all tasks when switch to follower.
   STaskStartInfo  startInfo;
-  SRWLatch        lock;
+  TdThreadRwlock  lock;
   int32_t         walScanCounter;
   void*           streamBackend;
   int64_t         streamBackendRid;

source/dnode/vnode/src/sma/smaRollup.c

@@ -240,23 +240,23 @@ int32_t tdFetchTbUidList(SSma *pSma, STbUidStore **ppStore, tb_uid_t suid, tb_ui
 
 static void tdRSmaTaskInit(SStreamMeta *pMeta, SRSmaInfoItem *pItem, SStreamTaskId *pId) {
   STaskId id = {.streamId = pId->streamId, .taskId = pId->taskId};
-  taosRLockLatch(&pMeta->lock);
+  streamMetaRLock(pMeta);
   SStreamTask **ppTask = (SStreamTask **)taosHashGet(pMeta->pTasksMap, &id, sizeof(id));
   if (ppTask && *ppTask) {
     pItem->submitReqVer = (*ppTask)->chkInfo.checkpointVer;
     pItem->fetchResultVer = (*ppTask)->info.triggerParam;
   }
-  taosRUnLockLatch(&pMeta->lock);
+  streamMetaRUnLock(pMeta);
 }
 
 static void tdRSmaTaskRemove(SStreamMeta *pMeta, int64_t streamId, int32_t taskId) {
   streamMetaUnregisterTask(pMeta, streamId, taskId);
-  taosWLockLatch(&pMeta->lock);
+  streamMetaWLock(pMeta);
   int32_t numOfTasks = streamMetaGetNumOfTasks(pMeta);
   if (streamMetaCommit(pMeta) < 0) {
     // persist to disk
   }
-  taosWUnLockLatch(&pMeta->lock);
+  streamMetaWUnLock(pMeta);
   smaDebug("vgId:%d, rsma task:%" PRIi64 ",%d dropped, remain tasks:%d", pMeta->vgId, streamId, taskId, numOfTasks);
 }
 
@@ -1301,14 +1301,14 @@ _checkpoint:
             checkpointBuilt = true;
           }
 
-          taosWLockLatch(&pMeta->lock);
+          streamMetaWLock(pMeta);
           if (streamMetaSaveTask(pMeta, pTask)) {
-            taosWUnLockLatch(&pMeta->lock);
+            streamMetaWUnLock(pMeta);
             code = terrno ? terrno : TSDB_CODE_OUT_OF_MEMORY;
             taosHashCancelIterate(pInfoHash, infoHash);
             TSDB_CHECK_CODE(code, lino, _exit);
           }
-          taosWUnLockLatch(&pMeta->lock);
+          streamMetaWUnLock(pMeta);
           smaDebug("vgId:%d, rsma commit, succeed to commit task:%p, submitReqVer:%" PRIi64 ", fetchResultVer:%" PRIi64
                    ", table:%" PRIi64 ", level:%d",
                    TD_VID(pVnode), pTask, pItem->submitReqVer, pItem->fetchResultVer, pRSmaInfo->suid, i + 1);
@@ -1316,13 +1316,13 @@ _checkpoint:
       }
     }
     if (pMeta) {
-      taosWLockLatch(&pMeta->lock);
+      streamMetaWLock(pMeta);
       if (streamMetaCommit(pMeta)) {
-        taosWUnLockLatch(&pMeta->lock);
+        streamMetaWUnLock(pMeta);
         code = terrno ? terrno : TSDB_CODE_OUT_OF_MEMORY;
         TSDB_CHECK_CODE(code, lino, _exit);
       }
-      taosWUnLockLatch(&pMeta->lock);
+      streamMetaWUnLock(pMeta);
     }
     if (checkpointBuilt) {
       smaInfo("vgId:%d, rsma commit, succeed to commit checkpoint:%" PRIi64, TD_VID(pVnode), checkpointId);

source/dnode/vnode/src/tq/tqStreamTaskSnap.c

@@ -196,7 +196,7 @@ int32_t streamTaskSnapWriterClose(SStreamTaskWriter* pWriter, int8_t rollback) {
   int32_t code = 0;
   STQ*    pTq = pWriter->pTq;
 
-  taosWLockLatch(&pTq->pStreamMeta->lock);
+  streamMetaWLock(pTq->pStreamMeta);
   tqDebug("vgId:%d, vnode stream-task snapshot writer closed", TD_VID(pTq->pVnode));
   if (rollback) {
     tdbAbort(pTq->pStreamMeta->db, pTq->pStreamMeta->txn);
@@ -212,14 +212,14 @@ int32_t streamTaskSnapWriterClose(SStreamTaskWriter* pWriter, int8_t rollback) {
     taosMemoryFree(pWriter);
     goto _err;
   }
-  taosWUnLockLatch(&pTq->pStreamMeta->lock);
+  streamMetaWUnLock(pTq->pStreamMeta);
   taosMemoryFree(pWriter);
   return code;
 
 _err:
   tqError("vgId:%d, vnode stream-task snapshot writer failed to close since %s", TD_VID(pWriter->pTq->pVnode),
           tstrerror(code));
-  taosWUnLockLatch(&pTq->pStreamMeta->lock);
+  streamMetaWUnLock(pTq->pStreamMeta);
   return code;
 }
 
@@ -240,13 +240,13 @@ int32_t streamTaskSnapWrite(SStreamTaskWriter* pWriter, uint8_t* pData, uint32_t
     tDecoderClear(&decoder);
 
     int64_t key[2] = {taskId.streamId, taskId.taskId};
-    taosWLockLatch(&pTq->pStreamMeta->lock);
+    streamMetaWLock(pTq->pStreamMeta);
     if (tdbTbUpsert(pTq->pStreamMeta->pTaskDb, key, sizeof(int64_t) << 1, (uint8_t*)pData + sizeof(SSnapDataHdr),
                     nData - sizeof(SSnapDataHdr), pTq->pStreamMeta->txn) < 0) {
-      taosWUnLockLatch(&pTq->pStreamMeta->lock);
+      streamMetaWUnLock(pTq->pStreamMeta);
       return -1;
     }
-    taosWUnLockLatch(&pTq->pStreamMeta->lock);
+    streamMetaWUnLock(pTq->pStreamMeta);
   } else if (pHdr->type == SNAP_DATA_STREAM_TASK_CHECKPOINT) {
     // do nothing
   }

source/libs/stream/src/streamMeta.c

@@ -1293,24 +1293,23 @@ void streamMetaResetStartInfo(STaskStartInfo* pStartInfo) {
 
 void streamMetaRLock(SStreamMeta* pMeta) {
   stTrace("vgId:%d meta-rlock", pMeta->vgId);
-  taosRLockLatch(&pMeta->lock);
+  taosThreadRwlockRdlock(&pMeta->lock);
 }
 
 void streamMetaRUnLock(SStreamMeta* pMeta) {
   stTrace("vgId:%d meta-runlock", pMeta->vgId);
-  taosRUnLockLatch(&pMeta->lock);
+  taosThreadRwlockUnlock(&pMeta->lock);
 }
 
 void streamMetaWLock(SStreamMeta* pMeta) {
   stTrace("vgId:%d meta-wlock", pMeta->vgId);
-  taosWLockLatch(&pMeta->lock);
+  taosThreadRwlockWrlock(&pMeta->lock);
-  ASSERT(pMeta->lock != 0x40000001);
   stTrace("vgId:%d meta-wlock completed", pMeta->vgId);
 }
 
 void streamMetaWUnLock(SStreamMeta* pMeta) {
   stTrace("vgId:%d meta-wunlock", pMeta->vgId);
-  taosWUnLockLatch(&pMeta->lock);
+  taosThreadRwlockUnlock(&pMeta->lock);
 }
 
 static void execHelper(struct SSchedMsg* pSchedMsg) {