flag 'hasTtlTask' for single task

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

@@ -77,7 +77,7 @@ int32_t metaUidFilterCacheGet(SMeta* pMeta, uint64_t suid, const void* pKey, int
 struct SMeta {
   TdThreadRwlock lock;
 
-  tsem_t txnReady; // if false we should not write journal
+  tsem_t txnReady; // if false, we should not write journal
 
   char*   path;
   SVnode* pVnode;

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

@@ -31,7 +31,7 @@ typedef enum DirtyEntryType {
 } DirtyEntryType;
 
 typedef struct STtlManger {
-  TdThreadRwlock lock;
+  //TdThreadRwlock lock; // use meta lock, no self lock needed
 
   TTB* pOldTtlIdx;  // btree<{deleteTime, tuid}, NULL>
 

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

@@ -153,7 +153,7 @@ int             metaCreateTable(SMeta* pMeta, int64_t version, SVCreateTbReq* pR
 int             metaDropTable(SMeta* pMeta, int64_t version, SVDropTbReq* pReq, SArray* tbUids, int64_t* tbUid);
 int32_t         metaTrimTables(SMeta* pMeta);
 int32_t         metaTtlSetExpireTime(SMeta* pMeta, int64_t timePointMs);
-int             metaTtlDropTable(SMeta* pMeta, SArray* tbUids);
+int             metaTtlDropTables(SMeta* pMeta, SArray* tbUids);
 int             metaAlterTable(SMeta* pMeta, int64_t version, SVAlterTbReq* pReq, STableMetaRsp* pMetaRsp);
 int             metaUpdateChangeTimeWithLock(SMeta* pMeta, tb_uid_t uid, int64_t changeTimeMs);
 SSchemaWrapper* metaGetTableSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver, int lock);
@@ -415,6 +415,7 @@ struct SVnode {
   TdThreadMutex lock;
   bool          blocked;
   bool          restored;
+  bool          hasTtlTask;
   tsem_t        syncSem;
   int32_t       blockSec;
   int64_t       blockSeq;

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

@@ -932,14 +932,19 @@ int metaTtlSetExpireTime(SMeta *pMeta, int64_t timePointMs) {
   return 0;
 }
 
-int metaTtlDropTable(SMeta *pMeta, SArray *tbUids) {
+int metaTtlDropTables(SMeta *pMeta, SArray *tbUids) {
+  int     ret = 0;
   int64_t startMs = taosGetTimestampMs();
 
   tsem_wait(&pMeta->txnReady);
+  metaWLock(pMeta);
   ttlMgrFlush(pMeta->pTtlMgr, pMeta->txn);
+  metaULock(pMeta);
   tsem_post(&pMeta->txnReady);
 
-  int ret = ttlMgrFindExpired(pMeta->pTtlMgr, tbUids);
+  metaRLock(pMeta);
+  ret = ttlMgrFindExpired(pMeta->pTtlMgr, tbUids);
+  metaULock(pMeta);
   if (ret != 0) {
     metaError("ttl failed to find expired table, ret:%d", ret);
     return ret;
@@ -959,6 +964,7 @@ int metaTtlDropTable(SMeta *pMeta, SArray *tbUids) {
     metaDropTableByUid(pMeta, uid, NULL);
     metaULock(pMeta);
     tsem_post(&pMeta->txnReady);
+    taosUsleep(1);
   }
 
   int64_t endMs = taosGetTimestampMs();

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

@@ -73,7 +73,7 @@ int ttlMgrOpen(STtlManger **ppTtlMgr, TDB *pEnv, int8_t rollback, const char *lo
   pTtlMgr->pTtlCache = taosHashInit(8192, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
   pTtlMgr->pDirtyUids = taosHashInit(8192, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
 
-  taosThreadRwlockInit(&pTtlMgr->lock, NULL);
+  //taosThreadRwlockInit(&pTtlMgr->lock, NULL);
 
   ret = ttlMgrFillCache(pTtlMgr);
   if (ret < 0) {
@@ -150,7 +150,7 @@ static void ttlMgrCleanup(STtlManger *pTtlMgr) {
   taosHashCleanup(pTtlMgr->pTtlCache);
   taosHashCleanup(pTtlMgr->pDirtyUids);
   tdbTbClose(pTtlMgr->pTtlIdx);
-  taosThreadRwlockDestroy(&pTtlMgr->lock);
+  //taosThreadRwlockDestroy(&pTtlMgr->lock);
   taosMemoryFree(pTtlMgr);
 }
 
@@ -508,9 +508,7 @@ int ttlMgrSetExpireTime(STtlManger *pTtlMgr, int64_t timePointMs) {
 static int32_t ttlMgrRLock(STtlManger *pTtlMgr) {
   int32_t ret = 0;
 
-  metaTrace("%s, ttl mgr rlock %p", pTtlMgr->logPrefix, &pTtlMgr->lock);
-
-  ret = taosThreadRwlockRdlock(&pTtlMgr->lock);
+  //ret = taosThreadRwlockRdlock(&pTtlMgr->lock);
 
   return ret;
 }
@@ -518,9 +516,7 @@ static int32_t ttlMgrRLock(STtlManger *pTtlMgr) {
 static int32_t ttlMgrWLock(STtlManger *pTtlMgr) {
   int32_t ret = 0;
 
-  metaTrace("%s, ttl mgr wlock %p", pTtlMgr->logPrefix, &pTtlMgr->lock);
-
-  ret = taosThreadRwlockWrlock(&pTtlMgr->lock);
+  //ret = taosThreadRwlockWrlock(&pTtlMgr->lock);
 
   return ret;
 }
@@ -528,9 +524,7 @@ static int32_t ttlMgrWLock(STtlManger *pTtlMgr) {
 static int32_t ttlMgrULock(STtlManger *pTtlMgr) {
   int32_t ret = 0;
 
-  metaTrace("%s, ttl mgr ulock %p", pTtlMgr->logPrefix, &pTtlMgr->lock);
-
-  ret = taosThreadRwlockUnlock(&pTtlMgr->lock);
+  //ret = taosThreadRwlockUnlock(&pTtlMgr->lock);
 
   return ret;
 }

source/dnode/vnode/src/vnd/vnodeCommit.c

@@ -362,7 +362,7 @@ static int32_t vnodeTtlTask(void *arg) {
   SVnode   *pVnode = pInfo->pVnode;
   SArray   *tbUids = taosArrayInit(8, sizeof(int64_t));
 
-  code = metaTtlDropTable(pVnode->pMeta, tbUids);
+  code = metaTtlDropTables(pVnode->pMeta, tbUids);
   if (code) {
     vFatal("vgId:%d, meta failed to drop table by ttl since %s", TD_VID(pVnode), terrstr());
     goto _exit;
@@ -373,7 +373,7 @@ static int32_t vnodeTtlTask(void *arg) {
   }
 
 _exit:
-  // end commit
+  pVnode->hasTtlTask = false;
   taosArrayDestroy(tbUids);
   taosMemoryFree(pInfo);
   return code;

source/dnode/vnode/src/vnd/vnodeOpen.c

@@ -371,6 +371,7 @@ SVnode *vnodeOpen(const char *path, int32_t diskPrimary, STfs *pTfs, SMsgCb msgC
   pVnode->msgCb = msgCb;
   taosThreadMutexInit(&pVnode->lock, NULL);
   pVnode->blocked = false;
+  pVnode->hasTtlTask = false;
 
   tsem_init(&pVnode->syncSem, 0, 0);
   tsem_init(&(pVnode->canCommit), 0, 1);

source/dnode/vnode/src/vnd/vnodeSvr.c

@@ -725,8 +725,14 @@ static int32_t vnodeProcessDropTtlTbReq(SVnode *pVnode, int64_t ver, void *pReq,
   code = metaTtlSetExpireTime(pVnode->pMeta, ttlExpireTimeMs);
   if (code) goto end;
 
-  code = vnodeAsyncTtlDropTable(pVnode);
-  if (code) goto end;
+  if (!pVnode->hasTtlTask) {
+    pVnode->hasTtlTask = true;
+    code = vnodeAsyncTtlDropTable(pVnode);
+    if (code) {
+      pVnode->hasTtlTask = false;
+      goto end;
+    }
+  }
 
   code = vnodeDoRetention(pVnode, ttlReq.timestampSec);