Merge pull request #19493 from taosdata/fix/TD-21871-simon

fix: allow mnode start even dnode in dropping state

source/dnode/mnode/impl/src/mndMnode.c

@@ -15,13 +15,13 @@
 
 #define _DEFAULT_SOURCE
 #include "mndMnode.h"
+#include "mndCluster.h"
 #include "mndDnode.h"
 #include "mndPrivilege.h"
 #include "mndShow.h"
 #include "mndSync.h"
 #include "mndTrans.h"
 #include "tmisce.h"
-#include "mndCluster.h"
 
 #define MNODE_VER_NUMBER   1
 #define MNODE_RESERVE_SIZE 64
@@ -181,9 +181,8 @@ _OVER:
 
 static int32_t mndMnodeActionInsert(SSdb *pSdb, SMnodeObj *pObj) {
   mTrace("mnode:%d, perform insert action, row:%p", pObj->id, pObj);
-  pObj->pDnode = sdbAcquire(pSdb, SDB_DNODE, &pObj->id);
+  pObj->pDnode = sdbAcquireNotReadyObj(pSdb, SDB_DNODE, &pObj->id);
   if (pObj->pDnode == NULL) {
-    terrno = TSDB_CODE_MND_DNODE_NOT_EXIST;
     mError("mnode:%d, failed to perform insert action since %s", pObj->id, terrstr());
     return -1;
   }

source/dnode/mnode/sdb/inc/sdb.h

@@ -291,6 +291,7 @@ int32_t sdbWriteWithoutFree(SSdb *pSdb, SSdbRaw *pRaw);
  * @return void* The object of the row.
  */
 void *sdbAcquire(SSdb *pSdb, ESdbType type, const void *pKey);
+void *sdbAcquireNotReadyObj(SSdb *pSdb, ESdbType type, const void *pKey);
 
 /**
  * @brief Release a row from sdb.

source/dnode/mnode/sdb/src/sdbHash.c

@@ -270,7 +270,7 @@ int32_t sdbWrite(SSdb *pSdb, SSdbRaw *pRaw) {
   return code;
 }
 
-void *sdbAcquire(SSdb *pSdb, ESdbType type, const void *pKey) {
+void *sdbAcquireAll(SSdb *pSdb, ESdbType type, const void *pKey, bool onlyReady) {
   terrno = 0;
 
   SHashObj *hash = sdbGetHash(pSdb, type);
@@ -306,10 +306,24 @@ void *sdbAcquire(SSdb *pSdb, ESdbType type, const void *pKey) {
       break;
   }
 
+  if (pRet == NULL) {
+    if (!onlyReady) {
+      terrno = 0;
+      atomic_add_fetch_32(&pRow->refCount, 1);
+      pRet = pRow->pObj;
+      sdbPrintOper(pSdb, pRow, "acquire");
+    }
+  }
+
   sdbUnLock(pSdb, type);
   return pRet;
 }
 
+void *sdbAcquire(SSdb *pSdb, ESdbType type, const void *pKey) { return sdbAcquireAll(pSdb, type, pKey, true); }
+void *sdbAcquireNotReadyObj(SSdb *pSdb, ESdbType type, const void *pKey) {
+  return sdbAcquireAll(pSdb, type, pKey, false);
+}
+
 static void sdbCheckRow(SSdb *pSdb, SSdbRow *pRow) {
   int32_t type = pRow->type;
   sdbWriteLock(pSdb, type);