Merge branch '3.0' into feature/vnode

include/common/taosmsg.h

@@ -75,9 +75,9 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_FUNCTION, "create-function" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_FUNCTION, "alter-function" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_FUNCTION, "drop-function" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_STABLE, "create-stable" )
-TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_STABLE_VGROUP, "stable-vgroup" )
-TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_STABLE, "drop-stable" )	
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_STABLE, "alter-stable" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_STABLE, "drop-stable" )	
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_STABLE_VGROUP, "stable-vgroup" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_QUERY, "kill-query" )	
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_CONN, "kill-conn" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_HEARTBEAT, "heartbeat" )
@@ -108,6 +108,7 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_AUTH_VNODE_IN, "auth-vnode" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_SYNC_VNODE_IN, "sync-vnode" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_COMPACT_VNODE_IN, "compact-vnode" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_MNODE_IN, "create-mnode" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_MNODE_IN, "alter-mnode" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_MNODE_IN, "drop-mnode" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CONFIG_DNODE_IN, "config-dnode" )
 
@@ -696,32 +697,33 @@ typedef struct {
 } SStatusRsp;
 
 typedef struct {
-  uint16_t port;
-  char     fqdn[TSDB_FQDN_LEN];
-} SVnodeDesc;
-
-typedef struct {
-  char       db[TSDB_FULL_DB_NAME_LEN];
-  uint32_t   vgId;
-  int32_t    cacheBlockSize;
-  int32_t    totalBlocks;
-  int32_t    daysPerFile;
-  int32_t    daysToKeep0;
-  int32_t    daysToKeep1;
-  int32_t    daysToKeep2;
-  int32_t    minRowsPerFileBlock;
-  int32_t    maxRowsPerFileBlock;
-  int32_t    fsyncPeriod;
-  int8_t     reserved[16];
-  int8_t     precision;
-  int8_t     compression;
-  int8_t     cacheLastRow;
-  int8_t     update;
-  int8_t     walLevel;
-  int8_t     replica;
-  int8_t     quorum;
-  int8_t     selfIndex;
-  SVnodeDesc replicas[TSDB_MAX_REPLICA];
+  int32_t  id;
+  uint16_t port;                 // node sync Port
+  char     fqdn[TSDB_FQDN_LEN];  // node FQDN
+} SReplica;
+
+typedef struct {
+  char     db[TSDB_FULL_DB_NAME_LEN];
+  uint32_t vgId;
+  int32_t  cacheBlockSize;
+  int32_t  totalBlocks;
+  int32_t  daysPerFile;
+  int32_t  daysToKeep0;
+  int32_t  daysToKeep1;
+  int32_t  daysToKeep2;
+  int32_t  minRowsPerFileBlock;
+  int32_t  maxRowsPerFileBlock;
+  int32_t  fsyncPeriod;
+  int8_t   reserved[16];
+  int8_t   precision;
+  int8_t   compression;
+  int8_t   cacheLastRow;
+  int8_t   update;
+  int8_t   walLevel;
+  int8_t   quorum;
+  int8_t   replica;
+  int8_t   selfIndex;
+  SReplica replicas[TSDB_MAX_REPLICA];
 } SCreateVnodeMsg, SAlterVnodeMsg;
 
 typedef struct {
@@ -825,8 +827,11 @@ typedef struct {
 } SCreateDnodeMsg, SDropDnodeMsg;
 
 typedef struct {
-  int32_t   dnodeId;
-} SCreateMnodeMsg, SDropMnodeMsg;
+  int32_t    dnodeId;
+  int8_t     replica;
+  int8_t     reserved[3];
+  SReplica replicas[TSDB_MAX_REPLICA];
+} SCreateMnodeMsg, SAlterMnodeMsg, SDropMnodeMsg;
 
 typedef struct {
   int32_t dnodeId;

include/server/mnode/mnode.h

@@ -20,6 +20,16 @@
 extern "C" {
 #endif
 
+typedef enum { MN_MSG_TYPE_WRITE = 1, MN_MSG_TYPE_APPLY, MN_MSG_TYPE_SYNC, MN_MSG_TYPE_READ } EMnMsgType;
+
+typedef struct SMnodeMsg SMnodeMsg;
+
+typedef struct {
+  int8_t   replica;
+  int8_t   selfIndex;
+  SReplica replicas[TSDB_MAX_REPLICA];
+} SMnodeCfg;
+
 typedef struct {
   int64_t numOfDnode;
   int64_t numOfMnode;
@@ -31,32 +41,33 @@ typedef struct {
   int64_t totalPoints;
   int64_t totalStorage;
   int64_t compStorage;
-} SMnodeStat;
+} SMnodeLoad;
 
 typedef struct {
+  int32_t dnodeId;
+  int64_t clusterId;
   void (*SendMsgToDnode)(struct SEpSet *epSet, struct SRpcMsg *rpcMsg);
   void (*SendMsgToMnode)(struct SRpcMsg *rpcMsg);
   void (*SendRedirectMsg)(struct SRpcMsg *rpcMsg, bool forShell);
-  void (*GetDnodeEp)(int32_t dnodeId, char *ep, char *fqdn, uint16_t *port);
-} SMnodeFp;
-
-typedef struct {
-  SMnodeFp fp;
-  int64_t  clusterId;
-  int32_t  dnodeId;
+  int32_t (*PutMsgIntoApplyQueue)(SMnodeMsg *pMsg);
 } SMnodePara;
 
 int32_t mnodeInit(SMnodePara para);
 void    mnodeCleanup();
-int32_t mnodeDeploy();
-void    mnodeUnDeploy();
-int32_t mnodeStart();
+
+int32_t mnodeDeploy(char *path, SMnodeCfg *pCfg);
+void    mnodeUnDeploy(char *path);
+int32_t mnodeStart(char *path, SMnodeCfg *pCfg);
+int32_t mnodeAlter(SMnodeCfg *pCfg);
 void    mnodeStop();
 
-int32_t mnodeGetStatistics(SMnodeStat *stat);
+int32_t mnodeGetLoad(SMnodeLoad *pLoad);
 int32_t mnodeRetriveAuth(char *user, char *spi, char *encrypt, char *secret, char *ckey);
 
-void mnodeProcessMsg(SRpcMsg *rpcMsg);
+SMnodeMsg *mnodeInitMsg(int32_t msgNum);
+int32_t    mnodeAppendMsg(SMnodeMsg *pMsg, SRpcMsg *pRpcMsg);
+void       mnodeCleanupMsg(SMnodeMsg *pMsg);
+void       mnodeProcessMsg(SMnodeMsg *pMsg, EMnMsgType msgType);
 
 #ifdef __cplusplus
 }

include/server/vnode/tq/tq.h

@@ -19,9 +19,6 @@
 #include "os.h"
 #include "tutil.h"
 
-#define TQ_ACTION_INSERT 0x7f7f7f7fULL
-#define TQ_ACTION_DELETE 0x80808080ULL
-
 #ifdef __cplusplus
 extern "C" {
 #endif

include/server/vnode/vnode.h

@@ -41,24 +41,25 @@ void vnodeOptionsClear(SVnodeOptions *);
 #include "trpc.h"
 
 typedef struct {
-  char       db[TSDB_FULL_DB_NAME_LEN];
-  int32_t    cacheBlockSize;  // MB
-  int32_t    totalBlocks;
-  int32_t    daysPerFile;
-  int32_t    daysToKeep0;
-  int32_t    daysToKeep1;
-  int32_t    daysToKeep2;
-  int32_t    minRowsPerFileBlock;
-  int32_t    maxRowsPerFileBlock;
-  int8_t     precision;  // time resolution
-  int8_t     compression;
-  int8_t     cacheLastRow;
-  int8_t     update;
-  int8_t     quorum;
-  int8_t     replica;
-  int8_t     walLevel;
-  int32_t    fsyncPeriod;  // millisecond
-  SVnodeDesc replicas[TSDB_MAX_REPLICA];
+  char     db[TSDB_FULL_DB_NAME_LEN];
+  int32_t  cacheBlockSize;  // MB
+  int32_t  totalBlocks;
+  int32_t  daysPerFile;
+  int32_t  daysToKeep0;
+  int32_t  daysToKeep1;
+  int32_t  daysToKeep2;
+  int32_t  minRowsPerFileBlock;
+  int32_t  maxRowsPerFileBlock;
+  int8_t   precision;  // time resolution
+  int8_t   compression;
+  int8_t   cacheLastRow;
+  int8_t   update;
+  int8_t   quorum;
+  int8_t   replica;
+  int8_t   selfIndex;
+  int8_t   walLevel;
+  int32_t  fsyncPeriod;  // millisecond
+  SReplica replicas[TSDB_MAX_REPLICA];
 } SVnodeCfg;
 
 typedef enum {

include/util/taoserror.h

@@ -216,14 +216,18 @@ int32_t* taosGetErrno();
 // dnode
 #define TSDB_CODE_DND_MSG_NOT_PROCESSED         TAOS_DEF_ERROR_CODE(0, 0x0400)  //"Message not processed")
 #define TSDB_CODE_DND_OUT_OF_MEMORY             TAOS_DEF_ERROR_CODE(0, 0x0401)  //"Dnode out of memory")
-#define TSDB_CODE_DND_DNODE_ID_NOT_MATCHED      TAOS_DEF_ERROR_CODE(0, 0x0402)  //"Dnode Id not matched")
-#define TSDB_CODE_DND_MNODE_ALREADY_DROPPED     TAOS_DEF_ERROR_CODE(0, 0x0403)  //"Mnode already deployed")
-#define TSDB_CODE_DND_NO_WRITE_ACCESS           TAOS_DEF_ERROR_CODE(0, 0x0404)  //"No permission for disk files in dnode")
-#define TSDB_CODE_DND_INVALID_MSG_LEN           TAOS_DEF_ERROR_CODE(0, 0x0405)  //"Invalid message length")
-#define TSDB_CODE_DND_ACTION_IN_PROGRESS        TAOS_DEF_ERROR_CODE(0, 0x0406)  //"Action in progress")
-#define TSDB_CODE_DND_TOO_MANY_VNODES           TAOS_DEF_ERROR_CODE(0, 0x0407)  //"Too many vnode directories")
-#define TSDB_CODE_DND_EXITING                   TAOS_DEF_ERROR_CODE(0, 0x0408)  //"Dnode is exiting"
-#define TSDB_CODE_DND_PARSE_VNODE_FILE_ERROR    TAOS_DEF_ERROR_CODE(0, 0x0409)  //"Parse vnodes.json error")
+#define TSDB_CODE_DND_MNODE_ID_NOT_MATCH_DNODE  TAOS_DEF_ERROR_CODE(0, 0x0402)  //"Mnode Id not match Dnode")
+#define TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED    TAOS_DEF_ERROR_CODE(0, 0x0403)  //"Mnode already deployed")
+#define TSDB_CODE_DND_MNODE_NOT_DEPLOYED        TAOS_DEF_ERROR_CODE(0, 0x0404)  //"Mnode not deployed")
+#define TSDB_CODE_DND_READ_MNODE_FILE_ERROR     TAOS_DEF_ERROR_CODE(0, 0x0405)  //"Read mnode.json error")
+#define TSDB_CODE_DND_WRITE_MNODE_FILE_ERROR    TAOS_DEF_ERROR_CODE(0, 0x0406)  //"Write mnode.json error")
+#define TSDB_CODE_DND_NO_WRITE_ACCESS           TAOS_DEF_ERROR_CODE(0, 0x0407)  //"No permission for disk files in dnode")
+#define TSDB_CODE_DND_INVALID_MSG_LEN           TAOS_DEF_ERROR_CODE(0, 0x0408)  //"Invalid message length")
+#define TSDB_CODE_DND_ACTION_IN_PROGRESS        TAOS_DEF_ERROR_CODE(0, 0x0409)  //"Action in progress")
+#define TSDB_CODE_DND_TOO_MANY_VNODES           TAOS_DEF_ERROR_CODE(0, 0x040A)  //"Too many vnode directories")
+#define TSDB_CODE_DND_EXITING                   TAOS_DEF_ERROR_CODE(0, 0x040B)  //"Dnode is exiting"
+#define TSDB_CODE_DND_PARSE_VNODE_FILE_ERROR    TAOS_DEF_ERROR_CODE(0, 0x040C)  //"Parse vnodes.json error")
+#define TSDB_CODE_DND_PARSE_DNODE_FILE_ERROR    TAOS_DEF_ERROR_CODE(0, 0x040D)  //"Parse dnodes.json error")
 
 // vnode
 #define TSDB_CODE_VND_ACTION_IN_PROGRESS        TAOS_DEF_ERROR_CODE(0, 0x0500)  //"Action in progress")

source/dnode/mgmt/inc/dnodeMnode.h

@@ -23,9 +23,13 @@ extern "C" {
 
 int32_t dnodeInitMnode();
 void    dnodeCleanupMnode();
-void    dnodeProcessMnodeMsg(SRpcMsg *pMsg, SEpSet *pEpSet);
 int32_t dnodeGetUserAuthFromMnode(char *user, char *spi, char *encrypt, char *secret, char *ckey);
 
+void dnodeProcessMnodeMgmtMsg(SRpcMsg *pMsg, SEpSet *pEpSet);
+void dnodeProcessMnodeReadMsg(SRpcMsg *pMsg, SEpSet *pEpSet);
+void dnodeProcessMnodeWriteMsg(SRpcMsg *pMsg, SEpSet *pEpSet);
+void dnodeProcessMnodeSyncMsg(SRpcMsg *pMsg, SEpSet *pEpSet);
+
 #ifdef __cplusplus
 }
 #endif

source/dnode/mgmt/src/dnodeTransport.c

@@ -51,76 +51,78 @@ static void dnodeInitMsgFp() {
   tsTrans.msgFp[TSDB_MSG_TYPE_MQ_RESET] = dnodeProcessVnodeWriteMsg;
   
   // msg from client to mnode
-  tsTrans.msgFp[TSDB_MSG_TYPE_CONNECT] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_ACCT] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_ACCT] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_ACCT] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_USER] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_USER] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_USER] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_DNODE] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_CONFIG_DNODE] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_DNODE] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_DB] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_DB] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_USE_DB] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_DB] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_SYNC_DB] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_TOPIC] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_TOPIC] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_TOPIC] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_FUNCTION] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_FUNCTION] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_FUNCTION] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_STABLE] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_STABLE] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_STABLE] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_STABLE_VGROUP] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_KILL_QUERY] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_KILL_CONN] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_HEARTBEAT] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_SHOW] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_SHOW_RETRIEVE] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_SHOW_RETRIEVE_FUNC] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_COMPACT_VNODE] = dnodeProcessMnodeMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_CONNECT] = dnodeProcessMnodeReadMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_ACCT] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_ACCT] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_ACCT] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_USER] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_USER] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_USER] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_DNODE] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_CONFIG_DNODE] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_DNODE] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_DB] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_DB] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_USE_DB] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_DB] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_SYNC_DB] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_TOPIC] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_TOPIC] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_TOPIC] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_FUNCTION] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_FUNCTION] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_FUNCTION] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_STABLE] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_STABLE] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_STABLE] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_STABLE_VGROUP] = dnodeProcessMnodeReadMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_KILL_QUERY] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_KILL_CONN] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_HEARTBEAT] = dnodeProcessMnodeReadMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_SHOW] = dnodeProcessMnodeReadMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_SHOW_RETRIEVE] = dnodeProcessMnodeReadMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_SHOW_RETRIEVE_FUNC] = dnodeProcessMnodeReadMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_COMPACT_VNODE] = dnodeProcessMnodeWriteMsg;
 
   // message from client to dnode
   tsTrans.msgFp[TSDB_MSG_TYPE_NETWORK_TEST] = dnodeProcessDnodeMsg;
 
   // message from mnode to vnode
   tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_STABLE_IN] = dnodeProcessVnodeWriteMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_STABLE_IN_RSP] = dnodeProcessMnodeMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_STABLE_IN_RSP] = dnodeProcessMnodeWriteMsg;
   tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_STABLE_IN] = dnodeProcessVnodeWriteMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_STABLE_IN_RSP] = dnodeProcessMnodeMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_STABLE_IN_RSP] = dnodeProcessMnodeWriteMsg;
   tsTrans.msgFp[TSDB_MSG_TYPE_DROP_STABLE_IN] = dnodeProcessVnodeWriteMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_STABLE_IN] = dnodeProcessMnodeMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_STABLE_IN] = dnodeProcessMnodeWriteMsg;
 
   // message from mnode to dnode
   tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_VNODE_IN] = dnodeProcessVnodeMgmtMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_VNODE_IN_RSP] = dnodeProcessMnodeMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_VNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
   tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_VNODE_IN] = dnodeProcessVnodeMgmtMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_VNODE_IN_RSP] = dnodeProcessMnodeMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_VNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
   tsTrans.msgFp[TSDB_MSG_TYPE_DROP_VNODE_IN] = dnodeProcessVnodeMgmtMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_VNODE_IN_RSP] = dnodeProcessMnodeMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_VNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
   tsTrans.msgFp[TSDB_MSG_TYPE_SYNC_VNODE_IN] = dnodeProcessVnodeMgmtMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_SYNC_VNODE_IN_RSP] = dnodeProcessMnodeMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_SYNC_VNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
   tsTrans.msgFp[TSDB_MSG_TYPE_AUTH_VNODE_IN] = dnodeProcessVnodeMgmtMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_AUTH_VNODE_IN_RSP] = dnodeProcessMnodeMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_AUTH_VNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
   tsTrans.msgFp[TSDB_MSG_TYPE_COMPACT_VNODE_IN] = dnodeProcessVnodeMgmtMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_COMPACT_VNODE_IN_RSP] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_MNODE_IN] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_MNODE_IN_RSP] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_MNODE_IN] = dnodeProcessMnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_MNODE_IN_RSP] = dnodeProcessMnodeMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_COMPACT_VNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_MNODE_IN] = dnodeProcessMnodeMgmtMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_MNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_MNODE_IN] = dnodeProcessMnodeMgmtMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_MNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_MNODE_IN] = dnodeProcessMnodeMgmtMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_DROP_MNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
   tsTrans.msgFp[TSDB_MSG_TYPE_CONFIG_DNODE_IN] = dnodeProcessDnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_CONFIG_DNODE_IN_RSP] = dnodeProcessMnodeMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_CONFIG_DNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
 
   // message from dnode to mnode
-  tsTrans.msgFp[TSDB_MSG_TYPE_AUTH] = dnodeProcessMnodeMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_AUTH] = dnodeProcessMnodeReadMsg;
   tsTrans.msgFp[TSDB_MSG_TYPE_AUTH_RSP] = dnodeProcessDnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_GRANT] = dnodeProcessMnodeMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_GRANT] = dnodeProcessMnodeWriteMsg;
   tsTrans.msgFp[TSDB_MSG_TYPE_GRANT_RSP] = dnodeProcessDnodeMsg;
-  tsTrans.msgFp[TSDB_MSG_TYPE_STATUS] = dnodeProcessMnodeMsg;
+  tsTrans.msgFp[TSDB_MSG_TYPE_STATUS] = dnodeProcessMnodeWriteMsg;
   tsTrans.msgFp[TSDB_MSG_TYPE_STATUS_RSP] = dnodeProcessDnodeMsg;
 }
 

source/dnode/mgmt/src/dnodeVnodes.c

@@ -983,7 +983,7 @@ static int32_t dnodeInitVnodeModule() {
 int32_t dnodeInitVnodes() {
   dInfo("dnode-vnodes start to init");
 
-  SSteps *pSteps = taosStepInit(3, dnodeReportStartup);
+  SSteps *pSteps = taosStepInit(6, dnodeReportStartup);
   taosStepAdd(pSteps, "dnode-vnode-env", dnodeInitVnodeModule, vnodeCleanup);
   taosStepAdd(pSteps, "dnode-vnode-mgmt", dnodeInitVnodeMgmtWorker, dnodeCleanupVnodeMgmtWorker);
   taosStepAdd(pSteps, "dnode-vnode-read", dnodeInitVnodeReadWorker, dnodeCleanupVnodeReadWorker);

source/dnode/mnode/inc/mnodeInt.h

@@ -32,7 +32,6 @@ EMnStatus mnodeGetStatus();
 void mnodeSendMsgToDnode(struct SEpSet *epSet, struct SRpcMsg *rpcMsg);
 void mnodeSendMsgToMnode(struct SRpcMsg *rpcMsg);
 void mnodeSendRedirectMsg(struct SRpcMsg *rpcMsg, bool forShell);
-void mnodeGetDnodeEp(int32_t dnodeId, char *ep, char *fqdn, uint16_t *port);
 
 #ifdef __cplusplus
 }

source/dnode/mnode/inc/mnodeWorker.h

@@ -24,7 +24,6 @@ extern "C" {
 
 int32_t mnodeInitWorker();
 void    mnodeCleanupWorker();
-void    mnodeProcessMsg(SRpcMsg *rpcMsg);
 void    mnodeSendRsp(SMnMsg *pMsg, int32_t code);
 void    mnodeReDispatchToWriteQueue(SMnMsg *pMsg);
 

source/dnode/mnode/src/mnodeTelem.c

@@ -172,21 +172,21 @@ static void mnodeAddVersionInfo(SBufferWriter* bw) {
 }
 
 static void mnodeAddRuntimeInfo(SBufferWriter* bw) {
-  SMnodeStat stat = {0};
-  if (mnodeGetStatistics(&stat) != 0) {
+  SMnodeLoad load = {0};
+  if (mnodeGetLoad(&load) != 0) {
     return;
   }
 
-  mnodeAddIntField(bw, "numOfDnode", stat.numOfDnode);
-  mnodeAddIntField(bw, "numOfMnode", stat.numOfMnode);
-  mnodeAddIntField(bw, "numOfVgroup", stat.numOfVgroup);
-  mnodeAddIntField(bw, "numOfDatabase", stat.numOfDatabase);
-  mnodeAddIntField(bw, "numOfSuperTable", stat.numOfSuperTable);
-  mnodeAddIntField(bw, "numOfChildTable", stat.numOfChildTable);
-  mnodeAddIntField(bw, "numOfColumn", stat.numOfColumn);
-  mnodeAddIntField(bw, "numOfPoint", stat.totalPoints);
-  mnodeAddIntField(bw, "totalStorage", stat.totalStorage);
-  mnodeAddIntField(bw, "compStorage", stat.compStorage);
+  mnodeAddIntField(bw, "numOfDnode", load.numOfDnode);
+  mnodeAddIntField(bw, "numOfMnode", load.numOfMnode);
+  mnodeAddIntField(bw, "numOfVgroup", load.numOfVgroup);
+  mnodeAddIntField(bw, "numOfDatabase", load.numOfDatabase);
+  mnodeAddIntField(bw, "numOfSuperTable", load.numOfSuperTable);
+  mnodeAddIntField(bw, "numOfChildTable", load.numOfChildTable);
+  mnodeAddIntField(bw, "numOfColumn", load.numOfColumn);
+  mnodeAddIntField(bw, "numOfPoint", load.totalPoints);
+  mnodeAddIntField(bw, "totalStorage", load.totalStorage);
+  mnodeAddIntField(bw, "compStorage", load.compStorage);
 }
 
 static void mnodeSendTelemetryReport() {

source/dnode/mnode/src/mnodeWorker.c

@@ -39,7 +39,7 @@ static struct {
   void (*msgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *pMsg);
 } tsMworker = {0};
 
-static SMnMsg *mnodeInitMsg(SRpcMsg *pRpcMsg) {
+static SMnMsg *mnodeInitMsg2(SRpcMsg *pRpcMsg) {
   int32_t    size = sizeof(SMnMsg) + pRpcMsg->contLen;
   SMnMsg *pMsg = taosAllocateQitem(size);
 
@@ -62,7 +62,7 @@ static SMnMsg *mnodeInitMsg(SRpcMsg *pRpcMsg) {
   return pMsg;
 }
 
-static void mnodeCleanupMsg(SMnMsg *pMsg) {
+static void mnodeCleanupMsg2(SMnMsg *pMsg) {
   if (pMsg == NULL) return;
   if (pMsg->rpcMsg.pCont != pMsg->pCont) {
     tfree(pMsg->rpcMsg.pCont);
@@ -75,7 +75,7 @@ static void mnodeDispatchToWriteQueue(SRpcMsg *pRpcMsg) {
   if (mnodeGetStatus() != MN_STATUS_READY || tsMworker.writeQ == NULL) {
     mnodeSendRedirectMsg(pRpcMsg, true);
   } else {
-    SMnMsg *pMsg = mnodeInitMsg(pRpcMsg);
+    SMnMsg *pMsg = mnodeInitMsg2(pRpcMsg);
     if (pMsg == NULL) {
       SRpcMsg rpcRsp = {.handle = pRpcMsg->handle, .code = TSDB_CODE_MND_INVALID_USER};
       rpcSendResponse(&rpcRsp);
@@ -91,7 +91,7 @@ static void mnodeDispatchToWriteQueue(SRpcMsg *pRpcMsg) {
 void mnodeReDispatchToWriteQueue(SMnMsg *pMsg) {
   if (mnodeGetStatus() != MN_STATUS_READY || tsMworker.writeQ == NULL) {
     mnodeSendRedirectMsg(&pMsg->rpcMsg, true);
-    mnodeCleanupMsg(pMsg);
+    mnodeCleanupMsg2(pMsg);
   } else {
     taosWriteQitem(tsMworker.writeQ, pMsg);
   }
@@ -101,7 +101,7 @@ static void mnodeDispatchToReadQueue(SRpcMsg *pRpcMsg) {
   if (mnodeGetStatus() != MN_STATUS_READY || tsMworker.readQ == NULL) {
     mnodeSendRedirectMsg(pRpcMsg, true);
   } else {
-    SMnMsg *pMsg = mnodeInitMsg(pRpcMsg);
+    SMnMsg *pMsg = mnodeInitMsg2(pRpcMsg);
     if (pMsg == NULL) {
       SRpcMsg rpcRsp = {.handle = pRpcMsg->handle, .code = TSDB_CODE_MND_INVALID_USER};
       rpcSendResponse(&rpcRsp);
@@ -118,7 +118,7 @@ static void mnodeDispatchToPeerQueue(SRpcMsg *pRpcMsg) {
   if (mnodeGetStatus() != MN_STATUS_READY || tsMworker.peerReqQ == NULL) {
     mnodeSendRedirectMsg(pRpcMsg, false);
   } else {
-    SMnMsg *pMsg = mnodeInitMsg(pRpcMsg);
+    SMnMsg *pMsg = mnodeInitMsg2(pRpcMsg);
     if (pMsg == NULL) {
       SRpcMsg rpcRsp = {.handle = pRpcMsg->handle, .code = TSDB_CODE_MND_INVALID_USER};
       rpcSendResponse(&rpcRsp);
@@ -133,7 +133,7 @@ static void mnodeDispatchToPeerQueue(SRpcMsg *pRpcMsg) {
 }
 
 void mnodeDispatchToPeerRspQueue(SRpcMsg *pRpcMsg) {
-  SMnMsg *pMsg = mnodeInitMsg(pRpcMsg);
+  SMnMsg *pMsg = mnodeInitMsg2(pRpcMsg);
   if (pMsg == NULL) {
     SRpcMsg rpcRsp = {.handle = pRpcMsg->handle, .code = TSDB_CODE_MND_INVALID_USER};
     rpcSendResponse(&rpcRsp);
@@ -162,7 +162,7 @@ void mnodeSendRsp(SMnMsg *pMsg, int32_t code) {
   };
 
   rpcSendResponse(&rpcRsp);
-  mnodeCleanupMsg(pMsg);
+  mnodeCleanupMsg2(pMsg);
 }
 
 static void mnodeInitMsgFp() {
@@ -405,7 +405,7 @@ static void mnodeProcessPeerRsp(SMnMsg *pMsg, void *unused) {
 
   if (!mnodeIsMaster()) {
     mError("msg:%p, ahandle:%p type:%s not processed for not master", pRpcMsg, pRpcMsg->ahandle, taosMsg[msgType]);
-    mnodeCleanupMsg(pMsg);
+    mnodeCleanupMsg2(pMsg);
   }
 
   if (tsMworker.peerRspFp[msgType]) {
@@ -414,7 +414,7 @@ static void mnodeProcessPeerRsp(SMnMsg *pMsg, void *unused) {
     mError("msg:%p, ahandle:%p type:%s is not processed", pRpcMsg, pRpcMsg->ahandle, taosMsg[msgType]);
   }
 
-  mnodeCleanupMsg(pMsg);
+  mnodeCleanupMsg2(pMsg);
 }
 
 int32_t mnodeInitWorker() {
@@ -486,10 +486,9 @@ void mnodeCleanupWorker() {
   mInfo("mnode worker is closed");
 }
 
-void mnodeProcessMsg(SRpcMsg *pMsg) {
-  if (tsMworker.msgFp[pMsg->msgType]) {
-    (*tsMworker.msgFp[pMsg->msgType])(pMsg);
-  } else {
-    assert(0);
-  }
-}
+SMnodeMsg *mnodeInitMsg(int32_t msgNum) { return NULL; }
+
+int32_t mnodeAppendMsg(SMnodeMsg *pMsg, SRpcMsg *pRpcMsg) { return 0; }
+
+void mnodeCleanupMsg(SMnodeMsg *pMsg) {}
+void mnodeProcessMsg(SMnodeMsg *pMsg, EMnMsgType msgType) {}

source/dnode/mnode/src/mondeInt.c

@@ -37,48 +37,42 @@
 #include "mnodeTelem.h"
 
 static struct {
-  int32_t  state;
-  int32_t  dnodeId;
-  int64_t  clusterId;
-  tmr_h    timer;
-  SMnodeFp fp;
-  SSteps * steps1;
-  SSteps * steps2;
+  int32_t    state;
+  int32_t    dnodeId;
+  int64_t    clusterId;
+  tmr_h      timer;
+  SSteps    *steps1;
+  SSteps    *steps2;
+  SMnodePara para;
 } tsMint;
 
 tmr_h mnodeGetTimer() { return tsMint.timer; }
 
-int32_t mnodeGetDnodeId() { return tsMint.dnodeId; }
+int32_t mnodeGetDnodeId() { return tsMint.para.dnodeId; }
 
-int64_t mnodeGetClusterId() { return tsMint.clusterId; }
+int64_t mnodeGetClusterId() { return tsMint.para.clusterId; }
 
 EMnStatus mnodeGetStatus() { return tsMint.state; }
 
 void mnodeSendMsgToDnode(struct SEpSet *epSet, struct SRpcMsg *rpcMsg) {
-  (*tsMint.fp.SendMsgToDnode)(epSet, rpcMsg);
+  (*tsMint.para.SendMsgToDnode)(epSet, rpcMsg);
 }
 
-void mnodeSendMsgToMnode(struct SRpcMsg *rpcMsg) { return (*tsMint.fp.SendMsgToMnode)(rpcMsg); }
+void mnodeSendMsgToMnode(struct SRpcMsg *rpcMsg) { return (*tsMint.para.SendMsgToMnode)(rpcMsg); }
 
-void mnodeSendRedirectMsg(struct SRpcMsg *rpcMsg, bool forShell) { (*tsMint.fp.SendRedirectMsg)(rpcMsg, forShell); }
+void mnodeSendRedirectMsg(struct SRpcMsg *rpcMsg, bool forShell) { (*tsMint.para.SendRedirectMsg)(rpcMsg, forShell); }
 
-void mnodeGetDnodeEp(int32_t dnodeId, char *ep, char *fqdn, uint16_t *port) {
-  (*tsMint.fp.GetDnodeEp)(dnodeId, ep, fqdn, port);
-}
-
-int32_t mnodeGetStatistics(SMnodeStat *stat) { return 0; }
+int32_t mnodeGetLoad(SMnodeLoad *pLoad) { return 0; }
 
 static int32_t mnodeSetPara(SMnodePara para) {
-  tsMint.fp = para.fp;
-  tsMint.dnodeId = para.dnodeId;
-  tsMint.clusterId = para.clusterId;
+  tsMint.para = para;
 
-  if (tsMint.fp.SendMsgToDnode == NULL) return -1;
-  if (tsMint.fp.SendMsgToMnode == NULL) return -1;
-  if (tsMint.fp.SendRedirectMsg == NULL) return -1;
-  if (tsMint.fp.GetDnodeEp == NULL) return -1;
-  if (tsMint.dnodeId < 0) return -1;
-  if (tsMint.clusterId < 0) return -1;
+  if (tsMint.para.SendMsgToDnode == NULL) return -1;
+  if (tsMint.para.SendMsgToMnode == NULL) return -1;
+  if (tsMint.para.SendRedirectMsg == NULL) return -1;
+  if (tsMint.para.PutMsgIntoApplyQueue == NULL) return -1;
+  if (tsMint.para.dnodeId < 0) return -1;
+  if (tsMint.para.clusterId < 0) return -1;
 
   return 0;
 }
@@ -142,13 +136,13 @@ static void mnodeCleanupStep1() { taosStepCleanup(tsMint.steps1); }
 static void mnodeCleanupStep2() { taosStepCleanup(tsMint.steps2); }
 
 static bool mnodeNeedDeploy() {
-  if (tsMint.dnodeId > 0) return false;
-  if (tsMint.clusterId > 0) return false;
+  if (tsMint.para.dnodeId > 0) return false;
+  if (tsMint.para.clusterId > 0) return false;
   if (strcmp(tsFirst, tsLocalEp) != 0) return false;
   return true;
 }
 
-int32_t mnodeDeploy() {
+int32_t mnodeDeploy(char *path, SMnodeCfg *pCfg) {
   if (tsMint.state != MN_STATUS_UNINIT) {
     mError("failed to deploy mnode since its deployed");
     return 0;
@@ -156,7 +150,7 @@ int32_t mnodeDeploy() {
     tsMint.state = MN_STATUS_INIT;
   }
 
-  if (tsMint.dnodeId <= 0 || tsMint.clusterId <= 0) {
+  if (tsMint.para.dnodeId <= 0 || tsMint.para.clusterId <= 0) {
     mError("failed to deploy mnode since cluster not ready");
     return TSDB_CODE_MND_NOT_READY;
   }
@@ -183,7 +177,7 @@ int32_t mnodeDeploy() {
   return 0;
 }
 
-void mnodeUnDeploy() {
+void mnodeUnDeploy(char *path) {
   sdbUnDeploy();
   mnodeCleanup();
 }
@@ -251,5 +245,6 @@ void mnodeCleanup() {
   }
 }
 
-int32_t mnodeStart() { return 0; }
+int32_t mnodeStart(char *path, SMnodeCfg *pCfg) { return 0; }
+int32_t mnodeAlter(SMnodeCfg *pCfg) { return 0; }
 void    mnodeStop() {}
\ No newline at end of file

source/dnode/vnode/tq/CMakeLists.txt

@@ -12,3 +12,7 @@ target_link_libraries(
     PUBLIC os
     PUBLIC util
 )
+
+if(${BUILD_TEST})
+    add_subdirectory(test)
+endif(${BUILD_TEST})

source/dnode/vnode/tq/inc/tqMetaStore.h

@@ -19,6 +19,11 @@
 #include "os.h"
 #include "tq.h"
 
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
 #define TQ_BUCKET_SIZE 0xFF
 #define TQ_PAGE_SIZE 4096
 //key + offset + size
@@ -32,9 +37,23 @@ inline static int TqEmptyTail() { //16
   return TQ_PAGE_SIZE - TqMaxEntryOnePage();
 }
 
-#ifdef __cplusplus
-extern "C" {
-#endif
+#define TQ_ACTION_CONST      0
+#define TQ_ACTION_INUSE      1
+#define TQ_ACTION_INUSE_CONT 2
+#define TQ_ACTION_INTXN      3
+
+#define TQ_SVER              0
+
+static const int8_t TQ_CONST_DELETE = TQ_ACTION_CONST;
+#define TQ_DELETE_TOKEN  (void*)&TQ_CONST_DELETE
+
+typedef struct TqSerializedHead {
+  int16_t ver;
+  int16_t action;
+  int32_t checksum;
+  int64_t ssize;
+  char    content[];
+} TqSerializedHead;
 
 typedef struct TqMetaHandle {
   int64_t key;
@@ -59,30 +78,33 @@ typedef struct TqMetaStore {
   TqMetaList* unpersistHead;
   int fileFd; //TODO:temporaral use, to be replaced by unified tfile
   int idxFd;  //TODO:temporaral use, to be replaced by unified tfile
-  int (*serializer)(TqGroupHandle*, void**);
-  const void* (*deserializer)(const void*, TqGroupHandle*);
+  char* dirPath;
+  int (*serializer)(const void* pObj, TqSerializedHead** ppHead);
+  const void* (*deserializer)(const TqSerializedHead* pHead, void** ppObj);
   void  (*deleter)(void*);
 } TqMetaStore;
 
 TqMetaStore*  tqStoreOpen(const char* path,
-    int serializer(TqGroupHandle*, void**),
-    const void* deserializer(const void*, TqGroupHandle*),
-    void deleter(void*));
+    int serializer(const void* pObj, TqSerializedHead** ppHead),
+    const void* deserializer(const TqSerializedHead* pHead, void** ppObj),
+    void deleter(void* pObj));
 int32_t       tqStoreClose(TqMetaStore*);
 //int32_t       tqStoreDelete(TqMetaStore*);
 //int32_t       TqStoreCommitAll(TqMetaStore*);
 int32_t       tqStorePersist(TqMetaStore*);
 
-TqMetaHandle* tqHandleGet(TqMetaStore*, int64_t key);
-int32_t       tqHandlePut(TqMetaStore*, int64_t key, void* value);
+void*   tqHandleGet(TqMetaStore*, int64_t key);
+int32_t tqHandleMovePut(TqMetaStore*, int64_t key, void* value);
+int32_t tqHandleCopyPut(TqMetaStore*, int64_t key, void* value, size_t vsize);
 //do commit
-int32_t       tqHandleCommit(TqMetaStore*, int64_t key);
+int32_t tqHandleCommit(TqMetaStore*, int64_t key);
 //delete uncommitted
-int32_t       tqHandleAbort(TqMetaStore*, int64_t key);
-//delete committed
-int32_t       tqHandleDel(TqMetaStore*, int64_t key);
+int32_t tqHandleAbort(TqMetaStore*, int64_t key);
+//delete committed kv pair
+//notice that a delete action still needs to be committed
+int32_t tqHandleDel(TqMetaStore*, int64_t key);
 //delete both committed and uncommitted
-int32_t       tqHandleClear(TqMetaStore*, int64_t key);
+int32_t tqHandleClear(TqMetaStore*, int64_t key);
 
 #ifdef __cplusplus
 }

source/dnode/vnode/tq/test/CMakeLists.txt

+add_executable(tqTest "")
+target_sources(tqTest
+    PRIVATE
+    "tqMetaTest.cpp"
+)
+target_include_directories(tqTest
+    PUBLIC
+    "${CMAKE_SOURCE_DIR}/include/server/vnode/tq"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+)
+
+target_link_libraries(tqTest
+    tq
+    gtest_main
+)
+enable_testing()
+add_test(
+    NAME tq_test
+    COMMAND tqTest
+)

source/dnode/vnode/tq/test/tqMetaTest.cpp

+#include <gtest/gtest.h>
+#include <cstring>
+#include <iostream>
+#include <queue>
+
+#include "tqMetaStore.h"
+
+struct Foo {
+  int32_t a;
+};
+
+int FooSerializer(const void* pObj, TqSerializedHead** ppHead) {
+  Foo* foo = (Foo*) pObj;
+  if((*ppHead) == NULL || (*ppHead)->ssize < sizeof(TqSerializedHead) + sizeof(int32_t)) {
+    *ppHead = (TqSerializedHead*)realloc(*ppHead, sizeof(TqSerializedHead) + sizeof(int32_t));
+    (*ppHead)->ssize = sizeof(TqSerializedHead) + sizeof(int32_t);
+  }
+  *(int32_t*)(*ppHead)->content = foo->a;
+  return (*ppHead)->ssize;
+}
+
+const void* FooDeserializer(const TqSerializedHead* pHead, void** ppObj) {
+  if(*ppObj == NULL) {
+    *ppObj = realloc(*ppObj, sizeof(int32_t));
+  }
+  Foo* pFoo = *(Foo**)ppObj;
+  pFoo->a = *(int32_t*)pHead->content; 
+  return NULL;
+}
+
+void FooDeleter(void* pObj) {
+  free(pObj); 
+}
+
+class TqMetaTest : public ::testing::Test {
+  protected:
+
+    void SetUp() override {
+      taosRemoveDir(pathName);
+      pMeta = tqStoreOpen(pathName,
+          FooSerializer, FooDeserializer, FooDeleter);
+      ASSERT(pMeta);
+    }
+
+    void TearDown() override {
+      tqStoreClose(pMeta);
+    }
+
+    TqMetaStore* pMeta;
+    const char* pathName = "/tmp/tq_test";
+};
+
+TEST_F(TqMetaTest, copyPutTest) {
+  Foo foo;
+  foo.a = 3;
+  tqHandleCopyPut(pMeta, 1, &foo, sizeof(Foo));
+
+  Foo* pFoo = (Foo*) tqHandleGet(pMeta, 1);
+  EXPECT_EQ(pFoo == NULL, true);
+}
+
+TEST_F(TqMetaTest, persistTest) {
+  Foo* pFoo = (Foo*)malloc(sizeof(Foo));
+  pFoo->a = 2;
+  tqHandleMovePut(pMeta, 1, pFoo);
+  Foo* pBar = (Foo*)tqHandleGet(pMeta, 1);
+  EXPECT_EQ(pBar == NULL, true);
+  tqHandleCommit(pMeta, 1);
+  pBar = (Foo*)tqHandleGet(pMeta, 1);
+  EXPECT_EQ(pBar->a, pFoo->a);
+  pBar = (Foo*)tqHandleGet(pMeta, 2);
+  EXPECT_EQ(pBar == NULL, true);
+
+  tqStoreClose(pMeta);
+  pMeta = tqStoreOpen(pathName,
+      FooSerializer, FooDeserializer, FooDeleter);
+  ASSERT(pMeta);
+
+  pBar = (Foo*)tqHandleGet(pMeta, 1);
+  ASSERT_EQ(pBar != NULL, true);
+  EXPECT_EQ(pBar->a, 2);
+
+  pBar = (Foo*)tqHandleGet(pMeta, 2);
+  EXPECT_EQ(pBar == NULL, true);
+
+  //taosRemoveDir(pathName);
+}
+
+TEST_F(TqMetaTest, uncommittedTest) {
+  Foo* pFoo = (Foo*)malloc(sizeof(Foo));
+  pFoo->a = 3;
+  tqHandleMovePut(pMeta, 1, pFoo);
+
+  pFoo = (Foo*) tqHandleGet(pMeta, 1);
+  EXPECT_EQ(pFoo == NULL, true);
+}
+
+TEST_F(TqMetaTest, abortTest) {
+  Foo* pFoo = (Foo*)malloc(sizeof(Foo));
+  pFoo->a = 3;
+  tqHandleMovePut(pMeta, 1, pFoo);
+
+  pFoo = (Foo*) tqHandleGet(pMeta, 1);
+  EXPECT_EQ(pFoo == NULL, true);
+
+  tqHandleAbort(pMeta, 1);
+  pFoo = (Foo*) tqHandleGet(pMeta, 1);
+  EXPECT_EQ(pFoo == NULL, true);
+}
+
+TEST_F(TqMetaTest, deleteTest) {
+  Foo* pFoo = (Foo*)malloc(sizeof(Foo));
+  pFoo->a = 3;
+  tqHandleMovePut(pMeta, 1, pFoo);
+
+  pFoo = (Foo*) tqHandleGet(pMeta, 1);
+  EXPECT_EQ(pFoo == NULL, true);
+
+  tqHandleCommit(pMeta, 1);
+
+  pFoo = (Foo*) tqHandleGet(pMeta, 1);
+  ASSERT_EQ(pFoo != NULL, true);
+  EXPECT_EQ(pFoo->a, 3);
+
+  tqHandleDel(pMeta, 1);
+  pFoo = (Foo*) tqHandleGet(pMeta, 1);
+  ASSERT_EQ(pFoo != NULL, true);
+  EXPECT_EQ(pFoo->a, 3);
+
+  tqHandleCommit(pMeta, 1);
+  pFoo = (Foo*) tqHandleGet(pMeta, 1);
+  EXPECT_EQ(pFoo == NULL, true);
+}

source/os/src/osDir.c

@@ -55,7 +55,7 @@ void taosRemoveDir(const char *dirname) {
   closedir(dir);
   rmdir(dirname);
 
-  printf("dir:%s is removed", dirname);
+  printf("dir:%s is removed\n", dirname);
 }
 
 bool taosDirExist(char *dirname) { return access(dirname, F_OK) == 0; }
@@ -138,4 +138,4 @@ bool taosRealPath(char *dirname, int32_t maxlen) {
   return true;
 }
 
-#endif
\ No newline at end of file
+#endif

source/util/src/terror.c

@@ -228,14 +228,18 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_ALREADY_EXIST,      "Topic already exists"
 // dnode
 TAOS_DEFINE_ERROR(TSDB_CODE_DND_MSG_NOT_PROCESSED,        "Message not processed")
 TAOS_DEFINE_ERROR(TSDB_CODE_DND_OUT_OF_MEMORY,            "Dnode out of memory")
-TAOS_DEFINE_ERROR(TSDB_CODE_DND_DNODE_ID_NOT_MATCHED,     "Dnode Id not matched")
-TAOS_DEFINE_ERROR(TSDB_CODE_DND_MNODE_ALREADY_DROPPED,    "Mnode already deployed")
+TAOS_DEFINE_ERROR(TSDB_CODE_DND_MNODE_ID_NOT_MATCH_DNODE, "Mnode Id not match Dnode")
+TAOS_DEFINE_ERROR(TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED,   "Mnode already deployed")
+TAOS_DEFINE_ERROR(TSDB_CODE_DND_MNODE_NOT_DEPLOYED,       "Mnode not deployed")
+TAOS_DEFINE_ERROR(TSDB_CODE_DND_READ_MNODE_FILE_ERROR,    "Read mnode.json error")
+TAOS_DEFINE_ERROR(TSDB_CODE_DND_WRITE_MNODE_FILE_ERROR,   "Write mnode.json error")
 TAOS_DEFINE_ERROR(TSDB_CODE_DND_NO_WRITE_ACCESS,          "No permission for disk files in dnode")
 TAOS_DEFINE_ERROR(TSDB_CODE_DND_INVALID_MSG_LEN,          "Invalid message length")
 TAOS_DEFINE_ERROR(TSDB_CODE_DND_ACTION_IN_PROGRESS,       "Action in progress")
 TAOS_DEFINE_ERROR(TSDB_CODE_DND_TOO_MANY_VNODES,          "Too many vnode directories")
 TAOS_DEFINE_ERROR(TSDB_CODE_DND_EXITING,                  "Dnode is exiting")
 TAOS_DEFINE_ERROR(TSDB_CODE_DND_PARSE_VNODE_FILE_ERROR,   "Parse vnodes.json error")
+TAOS_DEFINE_ERROR(TSDB_CODE_DND_PARSE_DNODE_FILE_ERROR,   "Parse dnodes.json error")
 
 // vnode
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_ACTION_IN_PROGRESS,       "Action in progress")