Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
未验证 提交 ef5ca96d 编写于 作者: S Shengliang Guan 提交者： GitHub
浏览文件
操作

Merge pull request #11224 from taosdata/feature/shm 

refact files
上级 aaf35118 1839793b
隐藏空白更改
内联 并排
Showing with 1076 addition and 815 deletion
include/libs/transport/trpc.h
浏览文件 @ ef5ca96d
...@@ -58,6 +58,9 @@ typedef struct { ...@@ -58,6 +58,9 @@ typedef struct {
void *pNode; void *pNode;
} SNodeMsg; } SNodeMsg;
typedef void (*RpcCfp)(void *parent, SRpcMsg *, SEpSet *);
typedef int (*RpcAfp)(void *parent, char *tableId, char *spi, char *encrypt, char *secret, char *ckey);
typedef struct SRpcInit { typedef struct SRpcInit {
uint16_t localPort; // local port uint16_t localPort; // local port
char * label; // for debug purpose char * label; // for debug purpose
...@@ -74,10 +77,10 @@ typedef struct SRpcInit { ...@@ -74,10 +77,10 @@ typedef struct SRpcInit {
char *ckey; // ciphering key char *ckey; // ciphering key
// call back to process incoming msg, code shall be ignored by server app // call back to process incoming msg, code shall be ignored by server app
void (*cfp)(void *parent, SRpcMsg *, SEpSet *); RpcCfp cfp;
// call back to retrieve the client auth info, for server app only // call back to retrieve the client auth info, for server app only
int (*afp)(void *parent, char *tableId, char *spi, char *encrypt, char *secret, char *ckey); RpcAfp afp;;
void *parent; void *parent;
} SRpcInit; } SRpcInit;
......
include/util/tdef.h
浏览文件 @ ef5ca96d
...@@ -480,9 +480,9 @@ enum { ...@@ -480,9 +480,9 @@ enum {
SND_WORKER_TYPE__UNIQUE, SND_WORKER_TYPE__UNIQUE,
}; };
#define MND_VGID -1 #define MNODE_HANDLE -1
#define QND_VGID 1 #define QNODE_HANDLE 1
#define VND_VGID 0 #define DEFAULT_HANDLE 0
#define MAX_NUM_STR_SIZE 40 #define MAX_NUM_STR_SIZE 40
......
source/dnode/mgmt/CMakeLists.txt
浏览文件 @ ef5ca96d
aux_source_directory(dm/src DNODE_SRC) aux_source_directory(dm DNODE_SRC)
aux_source_directory(qm/src DNODE_SRC) aux_source_directory(qm DNODE_SRC)
aux_source_directory(bm/src DNODE_SRC) aux_source_directory(bm DNODE_SRC)
aux_source_directory(sm/src DNODE_SRC) aux_source_directory(sm DNODE_SRC)
aux_source_directory(vm/src DNODE_SRC) aux_source_directory(vm DNODE_SRC)
aux_source_directory(mm/src DNODE_SRC) aux_source_directory(mm DNODE_SRC)
aux_source_directory(main/src DNODE_SRC) aux_source_directory(main DNODE_SRC)
add_library(dnode STATIC ${DNODE_SRC}) add_library(dnode STATIC ${DNODE_SRC})
target_link_libraries( target_link_libraries(
dnode cjson mnode vnode qnode snode bnode wal sync taos tfs monitor dnode cjson mnode vnode qnode snode bnode wal sync taos tfs monitor
...@@ -12,20 +12,14 @@ target_link_libraries( ...@@ -12,20 +12,14 @@ target_link_libraries(
target_include_directories( target_include_directories(
dnode dnode
PUBLIC "${CMAKE_SOURCE_DIR}/include/dnode/mgmt" PUBLIC "${CMAKE_SOURCE_DIR}/include/dnode/mgmt"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/dm/inc" PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/qm/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/bm/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/sm/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/vm/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/mm/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/main/inc"
) )
aux_source_directory(main/exe EXEC_SRC) aux_source_directory(exe EXEC_SRC)
add_executable(taosd ${EXEC_SRC}) add_executable(taosd ${EXEC_SRC})
target_include_directories( target_include_directories(
taosd taosd
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/main/inc" PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
) )
target_link_libraries(taosd dnode) target_link_libraries(taosd dnode)
......
source/dnode/mgmt/bm/src/bmInt.c → source/dnode/mgmt/bm/bmInt.c
浏览文件 @ ef5ca96d
...@@ -109,7 +109,7 @@ int32_t bmOpen(SMgmtWrapper *pWrapper) { ...@@ -109,7 +109,7 @@ int32_t bmOpen(SMgmtWrapper *pWrapper) {
return code; return code;
} }
void bmGetMgmtFp(SMgmtWrapper *pWrapper) { void bmSetMgmtFp(SMgmtWrapper *pWrapper) {
SMgmtFp mgmtFp = {0}; SMgmtFp mgmtFp = {0};
mgmtFp.openFp = bmOpen; mgmtFp.openFp = bmOpen;
mgmtFp.closeFp = bmClose; mgmtFp.closeFp = bmClose;
......
source/dnode/mgmt/bm/src/bmWorker.c → source/dnode/mgmt/bm/bmWorker.c
浏览文件 @ ef5ca96d
...@@ -18,7 +18,7 @@ ...@@ -18,7 +18,7 @@
static void bmSendErrorRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) { static void bmSendErrorRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) {
SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle, .code = code}; SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle, .code = code};
dndSendRsp(pWrapper, &rpcRsp); tmsgSendRsp(&rpcRsp);
dTrace("msg:%p, is freed", pMsg); dTrace("msg:%p, is freed", pMsg);
rpcFreeCont(pMsg->rpcMsg.pCont); rpcFreeCont(pMsg->rpcMsg.pCont);
......
source/dnode/mgmt/bm/inc/bm.h 已删除 100644 → 0
浏览文件 @ aaf35118
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_DND_BNODE_H_
#define _TD_DND_BNODE_H_
#include "dnd.h"
#ifdef __cplusplus
extern "C" {
#endif
void bmGetMgmtFp(SMgmtWrapper *pWrapper);
#ifdef __cplusplus
}
#endif
#endif /*_TD_DND_BNODE_H_*/
\ No newline at end of file
source/dnode/mgmt/dm/src/dmMsg.c → source/dnode/mgmt/dm/dmHandle.c
浏览文件 @ ef5ca96d
...@@ -15,7 +15,6 @@ ...@@ -15,7 +15,6 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "dmInt.h" #include "dmInt.h"
#include "vm.h"
void dmSendStatusReq(SDnodeMgmt *pMgmt) { void dmSendStatusReq(SDnodeMgmt *pMgmt) {
SDnode *pDnode = pMgmt->pDnode; SDnode *pDnode = pMgmt->pDnode;
...@@ -57,7 +56,9 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) { ...@@ -57,7 +56,9 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
pMgmt->statusSent = 1; pMgmt->statusSent = 1;
dTrace("send req:%s to mnode, app:%p", TMSG_INFO(rpcMsg.msgType), rpcMsg.ahandle); dTrace("send req:%s to mnode, app:%p", TMSG_INFO(rpcMsg.msgType), rpcMsg.ahandle);
dndSendReqToMnode(pMgmt->pWrapper, &rpcMsg); SEpSet epSet = {0};
dmGetMnodeEpSet(pMgmt, &epSet);
tmsgSendReq(&pMgmt->msgCb, &epSet, &rpcMsg);
} }
static void dmUpdateDnodeCfg(SDnodeMgmt *pMgmt, SDnodeCfg *pCfg) { static void dmUpdateDnodeCfg(SDnodeMgmt *pMgmt, SDnodeCfg *pCfg) {
...@@ -116,21 +117,115 @@ int32_t dmProcessConfigReq(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) { ...@@ -116,21 +117,115 @@ int32_t dmProcessConfigReq(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) {
return TSDB_CODE_OPS_NOT_SUPPORT; return TSDB_CODE_OPS_NOT_SUPPORT;
} }
void dmInitMsgHandles(SMgmtWrapper *pWrapper) {
static int32_t dmProcessCreateNodeMsg(SDnode *pDnode, ENodeType ntype, SNodeMsg *pMsg) {
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, ntype);
if (pWrapper != NULL) {
dndReleaseWrapper(pWrapper);
terrno = TSDB_CODE_NODE_ALREADY_DEPLOYED;
dError("failed to create node since %s", terrstr());
return -1;
}
pWrapper = &pDnode->wrappers[ntype];
if (taosMkDir(pWrapper->path) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to create dir:%s since %s", pWrapper->path, terrstr());
return -1;
}
int32_t code = (*pWrapper->fp.createMsgFp)(pWrapper, pMsg);
if (code != 0) {
dError("node:%s, failed to open since %s", pWrapper->name, terrstr());
} else {
dDebug("node:%s, has been opened", pWrapper->name);
pWrapper->deployed = true;
}
return code;
}
static int32_t dmProcessDropNodeMsg(SDnode *pDnode, ENodeType ntype, SNodeMsg *pMsg) {
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, ntype);
if (pWrapper == NULL) {
terrno = TSDB_CODE_NODE_NOT_DEPLOYED;
dError("failed to drop node since %s", terrstr());
return -1;
}
taosWLockLatch(&pWrapper->latch);
pWrapper->deployed = false;
int32_t code = (*pWrapper->fp.dropMsgFp)(pWrapper, pMsg);
if (code != 0) {
pWrapper->deployed = true;
dError("node:%s, failed to drop since %s", pWrapper->name, terrstr());
} else {
pWrapper->deployed = false;
dDebug("node:%s, has been dropped", pWrapper->name);
}
taosWUnLockLatch(&pWrapper->latch);
dndReleaseWrapper(pWrapper);
return code;
}
int32_t dmProcessCDnodeMsg(SDnode *pDnode, SNodeMsg *pMsg) {
switch (pMsg->rpcMsg.msgType) {
case TDMT_DND_CREATE_MNODE:
return dmProcessCreateNodeMsg(pDnode, MNODE, pMsg);
case TDMT_DND_DROP_MNODE:
return dmProcessDropNodeMsg(pDnode, MNODE, pMsg);
case TDMT_DND_CREATE_QNODE:
return dmProcessCreateNodeMsg(pDnode, QNODE, pMsg);
case TDMT_DND_DROP_QNODE:
return dmProcessDropNodeMsg(pDnode, QNODE, pMsg);
case TDMT_DND_CREATE_SNODE:
return dmProcessCreateNodeMsg(pDnode, SNODE, pMsg);
case TDMT_DND_DROP_SNODE:
return dmProcessDropNodeMsg(pDnode, SNODE, pMsg);
case TDMT_DND_CREATE_BNODE:
return dmProcessCreateNodeMsg(pDnode, BNODE, pMsg);
case TDMT_DND_DROP_BNODE:
return dmProcessDropNodeMsg(pDnode, BNODE, pMsg);
default:
terrno = TSDB_CODE_MSG_NOT_PROCESSED;
return -1;
}
}
static void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup) {
memcpy(pStartup, &pDnode->startup, sizeof(SStartupReq));
pStartup->finished = (dndGetStatus(pDnode) == DND_STAT_RUNNING);
}
void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) {
dDebug("startup req is received");
SStartupReq *pStartup = rpcMallocCont(sizeof(SStartupReq));
dndGetStartup(pDnode, pStartup);
dDebug("startup req is sent, step:%s desc:%s finished:%d", pStartup->name, pStartup->desc, pStartup->finished);
SRpcMsg rpcRsp = {
.handle = pReq->handle, .pCont = pStartup, .contLen = sizeof(SStartupReq), .ahandle = pReq->ahandle};
rpcSendResponse(&rpcRsp);
}
void dmInitMsgHandle(SMgmtWrapper *pWrapper) {
// Requests handled by DNODE // Requests handled by DNODE
dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_MNODE, dmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_MNODE, dmPutMsgToMgmtWorker, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_DROP_MNODE, dmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_DROP_MNODE, dmPutMsgToMgmtWorker, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_QNODE, dmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_QNODE, dmPutMsgToMgmtWorker, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_DROP_QNODE, dmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_DROP_QNODE, dmPutMsgToMgmtWorker, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_SNODE, dmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_SNODE, dmPutMsgToMgmtWorker, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_DROP_SNODE, dmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_DROP_SNODE, dmPutMsgToMgmtWorker, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_BNODE, dmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_BNODE, dmPutMsgToMgmtWorker, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_DROP_BNODE, dmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_DROP_BNODE, dmPutMsgToMgmtWorker, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_CONFIG_DNODE, dmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_CONFIG_DNODE, dmPutMsgToMgmtWorker, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_NETWORK_TEST, dmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_NETWORK_TEST, dmPutMsgToMgmtWorker, DEFAULT_HANDLE);
// Requests handled by MNODE // Requests handled by MNODE
dndSetMsgHandle(pWrapper, TDMT_MND_STATUS_RSP, dmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_STATUS_RSP, dmPutMsgToStatusWorker, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_GRANT_RSP, dmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_GRANT_RSP, dmPutMsgToMgmtWorker, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_AUTH_RSP, dmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_AUTH_RSP, dmPutMsgToMgmtWorker, DEFAULT_HANDLE);
} }
source/dnode/mgmt/dm/src/dmInt.c → source/dnode/mgmt/dm/dmInt.c
浏览文件 @ ef5ca96d
...@@ -78,7 +78,7 @@ static int32_t dmStart(SMgmtWrapper *pWrapper) { ...@@ -78,7 +78,7 @@ static int32_t dmStart(SMgmtWrapper *pWrapper) {
return dmStartThread(pWrapper->pMgmt); return dmStartThread(pWrapper->pMgmt);
} }
int32_t dmInit(SMgmtWrapper *pWrapper) { static int32_t dmInit(SMgmtWrapper *pWrapper) {
SDnode *pDnode = pWrapper->pDnode; SDnode *pDnode = pWrapper->pDnode;
SDnodeMgmt *pMgmt = taosMemoryCalloc(1, sizeof(SDnodeMgmt)); SDnodeMgmt *pMgmt = taosMemoryCalloc(1, sizeof(SDnodeMgmt));
dInfo("dnode-mgmt start to init"); dInfo("dnode-mgmt start to init");
...@@ -112,22 +112,19 @@ int32_t dmInit(SMgmtWrapper *pWrapper) { ...@@ -112,22 +112,19 @@ int32_t dmInit(SMgmtWrapper *pWrapper) {
return -1; return -1;
} }
if (dndInitServer(pDnode) != 0) { if (dndInitTrans(pDnode) != 0) {
dError("failed to init trans server since %s", terrstr()); dError("failed to init transport since %s", terrstr());
return -1;
}
if (dndInitClient(pDnode) != 0) {
dError("failed to init trans client since %s", terrstr());
return -1; return -1;
} }
pWrapper->pMgmt = pMgmt; pWrapper->pMgmt = pMgmt;
pMgmt->msgCb = dndCreateMsgcb(pWrapper);
dInfo("dnode-mgmt is initialized"); dInfo("dnode-mgmt is initialized");
return 0; return 0;
} }
void dmCleanup(SMgmtWrapper *pWrapper) { static void dmCleanup(SMgmtWrapper *pWrapper) {
SDnodeMgmt *pMgmt = pWrapper->pMgmt; SDnodeMgmt *pMgmt = pWrapper->pMgmt;
if (pMgmt == NULL) return; if (pMgmt == NULL) return;
...@@ -151,25 +148,24 @@ void dmCleanup(SMgmtWrapper *pWrapper) { ...@@ -151,25 +148,24 @@ void dmCleanup(SMgmtWrapper *pWrapper) {
taosMemoryFree(pMgmt); taosMemoryFree(pMgmt);
pWrapper->pMgmt = NULL; pWrapper->pMgmt = NULL;
dndCleanupServer(pDnode); dndCleanupTrans(pDnode);
dndCleanupClient(pDnode);
dInfo("dnode-mgmt is cleaned up"); dInfo("dnode-mgmt is cleaned up");
} }
int32_t dmRequire(SMgmtWrapper *pWrapper, bool *required) { static int32_t dmRequire(SMgmtWrapper *pWrapper, bool *required) {
*required = true; *required = true;
return 0; return 0;
} }
void dmGetMgmtFp(SMgmtWrapper *pWrapper) { void dmSetMgmtFp(SMgmtWrapper *pWrapper) {
SMgmtFp mgmtFp = {0}; SMgmtFp mgmtFp = {0};
mgmtFp.openFp = dmInit; mgmtFp.openFp = dmInit;
mgmtFp.closeFp = dmCleanup; mgmtFp.closeFp = dmCleanup;
mgmtFp.startFp = dmStart; mgmtFp.startFp = dmStart;
mgmtFp.requiredFp = dmRequire; mgmtFp.requiredFp = dmRequire;
dmInitMsgHandles(pWrapper); dmInitMsgHandle(pWrapper);
pWrapper->name = "dnode"; pWrapper->name = "dnode";
pWrapper->fp = mgmtFp; pWrapper->fp = mgmtFp;
} }
source/dnode/mgmt/dm/src/dmWorker.c → source/dnode/mgmt/dm/dmWorker.c
浏览文件 @ ef5ca96d
...@@ -14,16 +14,11 @@ ...@@ -14,16 +14,11 @@
*/ */
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "bm.h"
#include "dmInt.h" #include "dmInt.h"
#include "mm.h"
#include "qm.h"
#include "sm.h"
#include "vm.h"
static void *dmThreadRoutine(void *param) { static void *dmThreadRoutine(void *param) {
SDnodeMgmt *pMgmt = param; SDnodeMgmt *pMgmt = param;
SDnode * pDnode = pMgmt->pDnode; SDnode *pDnode = pMgmt->pDnode;
int64_t lastStatusTime = taosGetTimestampMs(); int64_t lastStatusTime = taosGetTimestampMs();
int64_t lastMonitorTime = lastStatusTime; int64_t lastMonitorTime = lastStatusTime;
...@@ -37,8 +32,7 @@ static void *dmThreadRoutine(void *param) { ...@@ -37,8 +32,7 @@ static void *dmThreadRoutine(void *param) {
} }
int64_t curTime = taosGetTimestampMs(); int64_t curTime = taosGetTimestampMs();
float statusInterval = (curTime - lastStatusTime) / 1000.0f;
float statusInterval = (curTime - lastStatusTime) / 1000.0f;
if (statusInterval >= tsStatusInterval && !pMgmt->statusSent) { if (statusInterval >= tsStatusInterval && !pMgmt->statusSent) {
dmSendStatusReq(pMgmt); dmSendStatusReq(pMgmt);
lastStatusTime = curTime; lastStatusTime = curTime;
...@@ -52,25 +46,26 @@ static void *dmThreadRoutine(void *param) { ...@@ -52,25 +46,26 @@ static void *dmThreadRoutine(void *param) {
} }
} }
int32_t dmStartThread(SDnodeMgmt *pMgmt) {
pMgmt->threadId = taosCreateThread(dmThreadRoutine, pMgmt);
if (pMgmt->threadId == NULL) {
dError("failed to init dnode thread");
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
return 0;
}
static void dmProcessQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) { static void dmProcessQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
SDnodeMgmt *pMgmt = pInfo->ahandle; SDnodeMgmt *pMgmt = pInfo->ahandle;
SDnode * pDnode = pMgmt->pDnode; SDnode *pDnode = pMgmt->pDnode;
SRpcMsg *pRpc = &pMsg->rpcMsg; SRpcMsg *pRpc = &pMsg->rpcMsg;
int32_t code = -1; int32_t code = -1;
dTrace("msg:%p, will be processed in dnode queue", pMsg); dTrace("msg:%p, will be processed in dnode queue", pMsg);
switch (pRpc->msgType) { switch (pRpc->msgType) {
case TDMT_DND_CREATE_MNODE:
case TDMT_DND_CREATE_QNODE:
case TDMT_DND_CREATE_SNODE:
case TDMT_DND_CREATE_BNODE:
case TDMT_DND_DROP_MNODE:
case TDMT_DND_DROP_QNODE:
case TDMT_DND_DROP_SNODE:
case TDMT_DND_DROP_BNODE:
code = dndProcessNodeMsg(pMgmt->pDnode, pMsg);
break;
case TDMT_DND_CONFIG_DNODE: case TDMT_DND_CONFIG_DNODE:
code = dmProcessConfigReq(pMgmt, pMsg); code = dmProcessConfigReq(pMgmt, pMsg);
break; break;
...@@ -84,8 +79,8 @@ static void dmProcessQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) { ...@@ -84,8 +79,8 @@ static void dmProcessQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
code = dmProcessGrantRsp(pMgmt, pMsg); code = dmProcessGrantRsp(pMgmt, pMsg);
break; break;
default: default:
terrno = TSDB_CODE_MSG_NOT_PROCESSED; code = dmProcessCDnodeMsg(pMgmt->pDnode, pMsg);
dError("msg:%p, type:%s not processed in dnode queue", pRpc->handle, TMSG_INFO(pRpc->msgType)); break;
} }
if (pRpc->msgType & 1u) { if (pRpc->msgType & 1u) {
...@@ -100,16 +95,14 @@ static void dmProcessQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) { ...@@ -100,16 +95,14 @@ static void dmProcessQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
} }
int32_t dmStartWorker(SDnodeMgmt *pMgmt) { int32_t dmStartWorker(SDnodeMgmt *pMgmt) {
SSingleWorkerCfg mgmtCfg = { SSingleWorkerCfg mcfg = {.min = 1, .max = 1, .name = "dnode-mgmt", .fp = (FItem)dmProcessQueue, .param = pMgmt};
.min = 1, .max = 1, .name = "dnode-mgmt", .fp = (FItem)dmProcessQueue, .param = pMgmt}; if (tSingleWorkerInit(&pMgmt->mgmtWorker, &mcfg) != 0) {
if (tSingleWorkerInit(&pMgmt->mgmtWorker, &mgmtCfg) != 0) {
dError("failed to start dnode mgmt worker since %s", terrstr()); dError("failed to start dnode mgmt worker since %s", terrstr());
return -1; return -1;
} }
SSingleWorkerCfg statusCfg = { SSingleWorkerCfg scfg = {.min = 1, .max = 1, .name = "dnode-status", .fp = (FItem)dmProcessQueue, .param = pMgmt};
.min = 1, .max = 1, .name = "dnode-status", .fp = (FItem)dmProcessQueue, .param = pMgmt}; if (tSingleWorkerInit(&pMgmt->statusWorker, &scfg) != 0) {
if (tSingleWorkerInit(&pMgmt->statusWorker, &statusCfg) != 0) {
dError("failed to start dnode status worker since %s", terrstr()); dError("failed to start dnode status worker since %s", terrstr());
return -1; return -1;
} }
...@@ -118,17 +111,6 @@ int32_t dmStartWorker(SDnodeMgmt *pMgmt) { ...@@ -118,17 +111,6 @@ int32_t dmStartWorker(SDnodeMgmt *pMgmt) {
return 0; return 0;
} }
int32_t dmStartThread(SDnodeMgmt *pMgmt) {
pMgmt->threadId = taosCreateThread(dmThreadRoutine, pMgmt);
if (pMgmt->threadId == NULL) {
dError("failed to init dnode thread");
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
return 0;
}
void dmStopWorker(SDnodeMgmt *pMgmt) { void dmStopWorker(SDnodeMgmt *pMgmt) {
tSingleWorkerCleanup(&pMgmt->mgmtWorker); tSingleWorkerCleanup(&pMgmt->mgmtWorker);
tSingleWorkerCleanup(&pMgmt->statusWorker); tSingleWorkerCleanup(&pMgmt->statusWorker);
...@@ -140,12 +122,18 @@ void dmStopWorker(SDnodeMgmt *pMgmt) { ...@@ -140,12 +122,18 @@ void dmStopWorker(SDnodeMgmt *pMgmt) {
dDebug("dnode workers are closed"); dDebug("dnode workers are closed");
} }
int32_t dmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { int32_t dmPutMsgToMgmtWorker(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SDnodeMgmt *pMgmt = pWrapper->pMgmt; SDnodeMgmt *pMgmt = pWrapper->pMgmt;
SSingleWorker *pWorker = &pMgmt->mgmtWorker; SSingleWorker *pWorker = &pMgmt->mgmtWorker;
if (pMsg->rpcMsg.msgType == TDMT_MND_STATUS_RSP) {
pWorker = &pMgmt->statusWorker; dTrace("msg:%p, put into worker %s", pMsg, pWorker->name);
} taosWriteQitem(pWorker->queue, pMsg);
return 0;
}
int32_t dmPutMsgToStatusWorker(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SDnodeMgmt *pMgmt = pWrapper->pMgmt;
SSingleWorker *pWorker = &pMgmt->statusWorker;
dTrace("msg:%p, put into worker %s", pMsg, pWorker->name); dTrace("msg:%p, put into worker %s", pMsg, pWorker->name);
taosWriteQitem(pWorker->queue, pMsg); taosWriteQitem(pWorker->queue, pMsg);
......
source/dnode/mgmt/dm/inc/dm.h 已删除 100644 → 0
浏览文件 @ aaf35118
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_DND_DNODE_H_
#define _TD_DND_DNODE_H_
#include "dnd.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct SDnodeMgmt SDnodeMgmt;
void dmGetMgmtFp(SMgmtWrapper *pWrapper);
void dmInitMsgHandles(SMgmtWrapper *pWrapper);
void dmGetMnodeEpSet(SDnodeMgmt *pMgmt, SEpSet *pEpSet);
void dmUpdateMnodeEpSet(SDnodeMgmt *pMgmt, SEpSet *pEpSet);
void dmSendRedirectRsp(SDnodeMgmt *pMgmt, const SRpcMsg *pMsg);
#ifdef __cplusplus
}
#endif
#endif /*_TD_DND_DNODE_H_*/
\ No newline at end of file
source/dnode/mgmt/bm/inc/bmInt.h → source/dnode/mgmt/inc/bmInt.h
浏览文件 @ ef5ca96d
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
#ifndef _TD_DND_BNODE_INT_H_ #ifndef _TD_DND_BNODE_INT_H_
#define _TD_DND_BNODE_INT_H_ #define _TD_DND_BNODE_INT_H_
#include "bm.h" #include "dnd.h"
#include "bnode.h" #include "bnode.h"
#ifdef __cplusplus #ifdef __cplusplus
......
source/dnode/mgmt/dm/inc/dmInt.h → source/dnode/mgmt/inc/dmInt.h
浏览文件 @ ef5ca96d
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
#ifndef _TD_DND_DNODE_INT_H_ #ifndef _TD_DND_DNODE_INT_H_
#define _TD_DND_DNODE_INT_H_ #define _TD_DND_DNODE_INT_H_
#include "dm.h" #include "dnd.h"
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
...@@ -33,28 +33,38 @@ typedef struct SDnodeMgmt { ...@@ -33,28 +33,38 @@ typedef struct SDnodeMgmt {
SRWLatch latch; SRWLatch latch;
SSingleWorker mgmtWorker; SSingleWorker mgmtWorker;
SSingleWorker statusWorker; SSingleWorker statusWorker;
SMsgCb msgCb;
const char *path; const char *path;
SDnode *pDnode; SDnode *pDnode;
SMgmtWrapper *pWrapper; SMgmtWrapper *pWrapper;
} SDnodeMgmt; } SDnodeMgmt;
// dmInt.c
void dmSetMgmtFp(SMgmtWrapper *pWrapper);
void dmGetMnodeEpSet(SDnodeMgmt *pMgmt, SEpSet *pEpSet);
void dmUpdateMnodeEpSet(SDnodeMgmt *pMgmt, SEpSet *pEpSet);
void dmSendRedirectRsp(SDnodeMgmt *pMgmt, const SRpcMsg *pMsg);
// dmFile.c // dmFile.c
int32_t dmReadFile(SDnodeMgmt *pMgmt); int32_t dmReadFile(SDnodeMgmt *pMgmt);
int32_t dmWriteFile(SDnodeMgmt *pMgmt); int32_t dmWriteFile(SDnodeMgmt *pMgmt);
void dmUpdateDnodeEps(SDnodeMgmt *pMgmt, SArray *pDnodeEps); void dmUpdateDnodeEps(SDnodeMgmt *pMgmt, SArray *pDnodeEps);
// dmMsg.c // dmHandle.c
void dmInitMsgHandle(SMgmtWrapper *pWrapper);
void dmSendStatusReq(SDnodeMgmt *pMgmt); void dmSendStatusReq(SDnodeMgmt *pMgmt);
int32_t dmProcessConfigReq(SDnodeMgmt *pMgmt, SNodeMsg *pMsg); int32_t dmProcessConfigReq(SDnodeMgmt *pMgmt, SNodeMsg *pMsg);
int32_t dmProcessStatusRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg); int32_t dmProcessStatusRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg);
int32_t dmProcessAuthRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg); int32_t dmProcessAuthRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg);
int32_t dmProcessGrantRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg); int32_t dmProcessGrantRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg);
int32_t dmProcessCDnodeMsg(SDnode *pDnode, SNodeMsg *pMsg);
// dmWorker.c // dmWorker.c
int32_t dmStartThread(SDnodeMgmt *pMgmt);
int32_t dmStartWorker(SDnodeMgmt *pMgmt); int32_t dmStartWorker(SDnodeMgmt *pMgmt);
void dmStopWorker(SDnodeMgmt *pMgmt); void dmStopWorker(SDnodeMgmt *pMgmt);
int32_t dmStartThread(SDnodeMgmt *pMgmt); int32_t dmPutMsgToMgmtWorker(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t dmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg); int32_t dmPutMsgToStatusWorker(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
source/dnode/mgmt/main/inc/dnd.h → source/dnode/mgmt/inc/dnd.h
浏览文件 @ ef5ca96d
...@@ -27,13 +27,13 @@ ...@@ -27,13 +27,13 @@
#include "tlockfree.h" #include "tlockfree.h"
#include "tlog.h" #include "tlog.h"
#include "tmsg.h" #include "tmsg.h"
#include "tmsgcb.h"
#include "tprocess.h" #include "tprocess.h"
#include "tqueue.h" #include "tqueue.h"
#include "trpc.h" #include "trpc.h"
#include "tthread.h" #include "tthread.h"
#include "ttime.h" #include "ttime.h"
#include "tworker.h" #include "tworker.h"
#include "tmsgcb.h"
#include "dnode.h" #include "dnode.h"
#include "monitor.h" #include "monitor.h"
...@@ -42,12 +42,42 @@ ...@@ -42,12 +42,42 @@
extern "C" { extern "C" {
#endif #endif
#define dFatal(...) { if (dDebugFlag & DEBUG_FATAL) { taosPrintLog("DND FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }} #define dFatal(...) \
#define dError(...) { if (dDebugFlag & DEBUG_ERROR) { taosPrintLog("DND ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }} { \
#define dWarn(...) { if (dDebugFlag & DEBUG_WARN) { taosPrintLog("DND WARN ", DEBUG_WARN, 255, __VA_ARGS__); }} if (dDebugFlag & DEBUG_FATAL) { \
#define dInfo(...) { if (dDebugFlag & DEBUG_INFO) { taosPrintLog("DND ", DEBUG_INFO, 255, __VA_ARGS__); }} taosPrintLog("DND FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); \
#define dDebug(...) { if (dDebugFlag & DEBUG_DEBUG) { taosPrintLog("DND ", DEBUG_DEBUG, dDebugFlag, __VA_ARGS__); }} } \
#define dTrace(...) { if (dDebugFlag & DEBUG_TRACE) { taosPrintLog("DND ", DEBUG_TRACE, dDebugFlag, __VA_ARGS__); }} }
#define dError(...) \
{ \
if (dDebugFlag & DEBUG_ERROR) { \
taosPrintLog("DND ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); \
} \
}
#define dWarn(...) \
{ \
if (dDebugFlag & DEBUG_WARN) { \
taosPrintLog("DND WARN ", DEBUG_WARN, 255, __VA_ARGS__); \
} \
}
#define dInfo(...) \
{ \
if (dDebugFlag & DEBUG_INFO) { \
taosPrintLog("DND ", DEBUG_INFO, 255, __VA_ARGS__); \
} \
}
#define dDebug(...) \
{ \
if (dDebugFlag & DEBUG_DEBUG) { \
taosPrintLog("DND ", DEBUG_DEBUG, dDebugFlag, __VA_ARGS__); \
} \
}
#define dTrace(...) \
{ \
if (dDebugFlag & DEBUG_TRACE) { \
taosPrintLog("DND ", DEBUG_TRACE, dDebugFlag, __VA_ARGS__); \
} \
}
typedef enum { DNODE, VNODES, QNODE, SNODE, MNODE, BNODE, NODE_MAX } ENodeType; typedef enum { DNODE, VNODES, QNODE, SNODE, MNODE, BNODE, NODE_MAX } ENodeType;
typedef enum { DND_STAT_INIT, DND_STAT_RUNNING, DND_STAT_STOPPED } EDndStatus; typedef enum { DND_STAT_INIT, DND_STAT_RUNNING, DND_STAT_STOPPED } EDndStatus;
...@@ -73,7 +103,7 @@ typedef int32_t (*DropNodeFp)(SMgmtWrapper *pWrapper, SNodeMsg *pMsg); ...@@ -73,7 +103,7 @@ typedef int32_t (*DropNodeFp)(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
typedef int32_t (*RequireNodeFp)(SMgmtWrapper *pWrapper, bool *required); typedef int32_t (*RequireNodeFp)(SMgmtWrapper *pWrapper, bool *required);
typedef struct SMsgHandle { typedef struct SMsgHandle {
SMgmtWrapper *pQndWrapper; SMgmtWrapper *pQndWrapper;
SMgmtWrapper *pMndWrapper; SMgmtWrapper *pMndWrapper;
SMgmtWrapper *pWrapper; SMgmtWrapper *pWrapper;
} SMsgHandle; } SMsgHandle;
...@@ -92,6 +122,7 @@ typedef struct SMgmtWrapper { ...@@ -92,6 +122,7 @@ typedef struct SMgmtWrapper {
char *path; char *path;
int32_t refCount; int32_t refCount;
SRWLatch latch; SRWLatch latch;
ENodeType ntype;
bool deployed; bool deployed;
bool required; bool required;
EProcType procType; EProcType procType;
...@@ -145,14 +176,11 @@ void dndSetMsgHandle(SMgmtWrapper *pWrapper, tmsg_t msgType, NodeMsgFp ...@@ -145,14 +176,11 @@ void dndSetMsgHandle(SMgmtWrapper *pWrapper, tmsg_t msgType, NodeMsgFp
SMgmtWrapper *dndAcquireWrapper(SDnode *pDnode, ENodeType nodeType); SMgmtWrapper *dndAcquireWrapper(SDnode *pDnode, ENodeType nodeType);
int32_t dndMarkWrapper(SMgmtWrapper *pWrapper); int32_t dndMarkWrapper(SMgmtWrapper *pWrapper);
void dndReleaseWrapper(SMgmtWrapper *pWrapper); void dndReleaseWrapper(SMgmtWrapper *pWrapper);
void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc);
// dndMonitor.h // dndMonitor.h
void dndSendMonitorReport(SDnode *pDnode); void dndSendMonitorReport(SDnode *pDnode);
// dndMsg.h
void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc);
int32_t dndProcessNodeMsg(SDnode *pDnode, SNodeMsg *pMsg);
// dndStr.h // dndStr.h
const char *dndStatStr(EDndStatus stat); const char *dndStatStr(EDndStatus stat);
const char *dndNodeLogStr(ENodeType ntype); const char *dndNodeLogStr(ENodeType ntype);
...@@ -160,15 +188,40 @@ const char *dndNodeProcStr(ENodeType ntype); ...@@ -160,15 +188,40 @@ const char *dndNodeProcStr(ENodeType ntype);
const char *dndEventStr(EDndEvent ev); const char *dndEventStr(EDndEvent ev);
// dndTransport.h // dndTransport.h
int32_t dndInitServer(SDnode *pDnode); int32_t dndInitTrans(SDnode *pDnode);
void dndCleanupServer(SDnode *pDnode); void dndCleanupTrans(SDnode *pDnode);
int32_t dndInitClient(SDnode *pDnode); SMsgCb dndCreateMsgcb(SMgmtWrapper *pWrapper);
void dndCleanupClient(SDnode *pDnode); SProcCfg dndGenProcCfg(SMgmtWrapper *pWrapper);
int32_t dndSendReqToMnode(SMgmtWrapper *pWrapper, SRpcMsg *pMsg); int32_t dndInitMsgHandle(SDnode *pDnode);
int32_t dndSendReq(SMgmtWrapper *pWrapper, const SEpSet *pEpSet, SRpcMsg *pMsg);
void dndSendRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp); // mgmt
void dndRegisterBrokenLinkArg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg); void dmSetMgmtFp(SMgmtWrapper *pWrapper);
SMsgCb dndCreateMsgcb(SMgmtWrapper *pWrapper); void bmSetMgmtFp(SMgmtWrapper *pWrapper);
void qmSetMgmtFp(SMgmtWrapper *pMgmt);
void smSetMgmtFp(SMgmtWrapper *pWrapper);
void vmSetMgmtFp(SMgmtWrapper *pWrapper);
void mmSetMgmtFp(SMgmtWrapper *pMgmt);
void dmGetMnodeEpSet(SDnodeMgmt *pMgmt, SEpSet *pEpSet);
void dmUpdateMnodeEpSet(SDnodeMgmt *pMgmt, SEpSet *pEpSet);
void dmSendRedirectRsp(SDnodeMgmt *pMgmt, const SRpcMsg *pMsg);
typedef struct {
int32_t openVnodes;
int32_t totalVnodes;
int32_t masterNum;
int64_t numOfSelectReqs;
int64_t numOfInsertReqs;
int64_t numOfInsertSuccessReqs;
int64_t numOfBatchInsertReqs;
int64_t numOfBatchInsertSuccessReqs;
} SVnodesStat;
void vmMonitorVnodeLoads(SMgmtWrapper *pWrapper, SArray *pLoads);
int32_t vmMonitorTfsInfo(SMgmtWrapper *pWrapper, SMonDiskInfo *pInfo);
void vmMonitorVnodeReqs(SMgmtWrapper *pWrapper, SMonDnodeInfo *pInfo);
int32_t mmMonitorMnodeInfo(SMgmtWrapper *pWrapper, SMonClusterInfo *pClusterInfo, SMonVgroupInfo *pVgroupInfo,
SMonGrantInfo *pGrantInfo);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
source/dnode/mgmt/main/inc/dndInt.h → source/dnode/mgmt/inc/dndInt.h
浏览文件 @ ef5ca96d
...@@ -18,13 +18,6 @@ ...@@ -18,13 +18,6 @@
#include "dnd.h" #include "dnd.h"
#include "bm.h"
#include "dm.h"
#include "mm.h"
#include "qm.h"
#include "sm.h"
#include "vm.h"
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
#endif #endif
...@@ -44,13 +37,8 @@ void dndClose(SDnode *pDnode); ...@@ -44,13 +37,8 @@ void dndClose(SDnode *pDnode);
void dndHandleEvent(SDnode *pDnode, EDndEvent event); void dndHandleEvent(SDnode *pDnode, EDndEvent event);
// dndMsg.c // dndMsg.c
void dndProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, SEpSet *pEpSet);
void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pMsg); void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pMsg);
// dndTransport.c
int32_t dndInitMsgHandle(SDnode *pDnode);
void dndSendRpcRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp);
// dndFile.c // dndFile.c
TdFilePtr dndCheckRunning(const char *dataDir); TdFilePtr dndCheckRunning(const char *dataDir);
int32_t dndReadShmFile(SDnode *pDnode); int32_t dndReadShmFile(SDnode *pDnode);
......
source/dnode/mgmt/mm/inc/mmInt.h → source/dnode/mgmt/inc/mmInt.h
浏览文件 @ ef5ca96d
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
#ifndef _TD_DND_MNODE_INT_H_ #ifndef _TD_DND_MNODE_INT_H_
#define _TD_DND_MNODE_INT_H_ #define _TD_DND_MNODE_INT_H_
#include "mm.h" #include "dnd.h"
#include "mnode.h" #include "mnode.h"
#ifdef __cplusplus #ifdef __cplusplus
......
source/dnode/mgmt/qm/inc/qmInt.h → source/dnode/mgmt/inc/qmInt.h
浏览文件 @ ef5ca96d
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
#ifndef _TD_DND_QNODE_INT_H_ #ifndef _TD_DND_QNODE_INT_H_
#define _TD_DND_QNODE_INT_H_ #define _TD_DND_QNODE_INT_H_
#include "qm.h" #include "dnd.h"
#include "qnode.h" #include "qnode.h"
#ifdef __cplusplus #ifdef __cplusplus
......
source/dnode/mgmt/sm/inc/smInt.h → source/dnode/mgmt/inc/smInt.h
浏览文件 @ ef5ca96d
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
#ifndef _TD_DND_SNODE_INT_H_ #ifndef _TD_DND_SNODE_INT_H_
#define _TD_DND_SNODE_INT_H_ #define _TD_DND_SNODE_INT_H_
#include "sm.h" #include "dnd.h"
#include "snode.h" #include "snode.h"
#ifdef __cplusplus #ifdef __cplusplus
......
source/dnode/mgmt/vm/inc/vmInt.h → source/dnode/mgmt/inc/vmInt.h
浏览文件 @ ef5ca96d
...@@ -17,7 +17,7 @@ ...@@ -17,7 +17,7 @@
#define _TD_DND_VNODES_INT_H_ #define _TD_DND_VNODES_INT_H_
#include "sync.h" #include "sync.h"
#include "vm.h" #include "dnd.h"
#include "vnode.h" #include "vnode.h"
#ifdef __cplusplus #ifdef __cplusplus
......
source/dnode/mgmt/main/src/dndStr.c → source/dnode/mgmt/main/dndEnv.c
浏览文件 @ ef5ca96d
...@@ -15,6 +15,48 @@ ...@@ -15,6 +15,48 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "dndInt.h" #include "dndInt.h"
#include "wal.h"
static int8_t once = DND_ENV_INIT;
int32_t dndInit() {
dDebug("start to init dnode env");
if (atomic_val_compare_exchange_8(&once, DND_ENV_INIT, DND_ENV_READY) != DND_ENV_INIT) {
terrno = TSDB_CODE_REPEAT_INIT;
dError("failed to init dnode env since %s", terrstr());
return -1;
}
taosIgnSIGPIPE();
taosBlockSIGPIPE();
taosResolveCRC();
SMonCfg monCfg = {0};
monCfg.maxLogs = tsMonitorMaxLogs;
monCfg.port = tsMonitorPort;
monCfg.server = tsMonitorFqdn;
monCfg.comp = tsMonitorComp;
if (monInit(&monCfg) != 0) {
dError("failed to init monitor since %s", terrstr());
return -1;
}
dInfo("dnode env is initialized");
return 0;
}
void dndCleanup() {
dDebug("start to cleanup dnode env");
if (atomic_val_compare_exchange_8(&once, DND_ENV_READY, DND_ENV_CLEANUP) != DND_ENV_READY) {
dError("dnode env is already cleaned up");
return;
}
monCleanup();
walCleanUp();
taosStopCacheRefreshWorker();
dInfo("dnode env is cleaned up");
}
const char *dndStatStr(EDndStatus status) { const char *dndStatStr(EDndStatus status) {
switch (status) { switch (status) {
...@@ -25,7 +67,7 @@ const char *dndStatStr(EDndStatus status) { ...@@ -25,7 +67,7 @@ const char *dndStatStr(EDndStatus status) {
case DND_STAT_STOPPED: case DND_STAT_STOPPED:
return "stopped"; return "stopped";
default: default:
return "unknown"; return "UNKNOWN";
} }
} }
......
source/dnode/mgmt/main/src/dndExec.c → source/dnode/mgmt/main/dndExec.c
浏览文件 @ ef5ca96d
...@@ -65,53 +65,6 @@ void dndCloseNode(SMgmtWrapper *pWrapper) { ...@@ -65,53 +65,6 @@ void dndCloseNode(SMgmtWrapper *pWrapper) {
dDebug("node:%s, mgmt has been closed", pWrapper->name); dDebug("node:%s, mgmt has been closed", pWrapper->name);
} }
static void dndConsumeChildQueue(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int16_t msgLen, void *pCont, int32_t contLen,
ProcFuncType ftype) {
SRpcMsg *pRpc = &pMsg->rpcMsg;
pRpc->pCont = pCont;
dTrace("msg:%p, get from child queue, handle:%p app:%p", pMsg, pRpc->handle, pRpc->ahandle);
NodeMsgFp msgFp = pWrapper->msgFps[TMSG_INDEX(pRpc->msgType)];
int32_t code = (*msgFp)(pWrapper, pMsg);
if (code != 0) {
dError("msg:%p, failed to process since code:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code));
if (pRpc->msgType & 1U) {
SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = terrno};
dndSendRsp(pWrapper, &rsp);
}
dTrace("msg:%p, is freed", pMsg);
taosFreeQitem(pMsg);
rpcFreeCont(pCont);
}
}
static void dndConsumeParentQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, int16_t msgLen, void *pCont, int32_t contLen,
ProcFuncType ftype) {
pMsg->pCont = pCont;
dTrace("msg:%p, get from parent queue, ftype:%d handle:%p, app:%p", pMsg, ftype, pMsg->handle, pMsg->ahandle);
switch (ftype) {
case PROC_REGIST:
rpcRegisterBrokenLinkArg(pMsg);
break;
case PROC_RELEASE:
rpcReleaseHandle(pMsg->handle, (int8_t)pMsg->code);
rpcFreeCont(pCont);
break;
case PROC_REQ:
dndSendReqToMnode(pWrapper, pMsg);
// dndSendReq(pWrapper, (const SEpSet *)((char *)pMsg + sizeof(SRpcMsg)), pMsg);
break;
case PROC_RSP:
dndSendRpcRsp(pWrapper, pMsg);
break;
default:
break;
}
taosMemoryFree(pMsg);
}
static int32_t dndNewProc(SMgmtWrapper *pWrapper, ENodeType n) { static int32_t dndNewProc(SMgmtWrapper *pWrapper, ENodeType n) {
char tstr[8] = {0}; char tstr[8] = {0};
...@@ -135,22 +88,6 @@ static int32_t dndNewProc(SMgmtWrapper *pWrapper, ENodeType n) { ...@@ -135,22 +88,6 @@ static int32_t dndNewProc(SMgmtWrapper *pWrapper, ENodeType n) {
return 0; return 0;
} }
static SProcCfg dndGenProcCfg(SMgmtWrapper *pWrapper) {
SProcCfg cfg = {.childConsumeFp = (ProcConsumeFp)dndConsumeChildQueue,
.childMallocHeadFp = (ProcMallocFp)taosAllocateQitem,
.childFreeHeadFp = (ProcFreeFp)taosFreeQitem,
.childMallocBodyFp = (ProcMallocFp)rpcMallocCont,
.childFreeBodyFp = (ProcFreeFp)rpcFreeCont,
.parentConsumeFp = (ProcConsumeFp)dndConsumeParentQueue,
.parentMallocHeadFp = (ProcMallocFp)taosMemoryMalloc,
.parentFreeHeadFp = (ProcFreeFp)taosMemoryFree,
.parentMallocBodyFp = (ProcMallocFp)rpcMallocCont,
.parentFreeBodyFp = (ProcFreeFp)rpcFreeCont,
.shm = pWrapper->shm,
.pParent = pWrapper,
.name = pWrapper->name};
return cfg;
}
static int32_t dndRunInSingleProcess(SDnode *pDnode) { static int32_t dndRunInSingleProcess(SDnode *pDnode) {
dInfo("dnode run in single process"); dInfo("dnode run in single process");
......
source/dnode/mgmt/main/src/dndInt.c → source/dnode/mgmt/main/dndInt.c
浏览文件 @ ef5ca96d
...@@ -82,12 +82,12 @@ SDnode *dndCreate(const SDnodeOpt *pOption) { ...@@ -82,12 +82,12 @@ SDnode *dndCreate(const SDnodeOpt *pOption) {
} }
dndSetStatus(pDnode, DND_STAT_INIT); dndSetStatus(pDnode, DND_STAT_INIT);
dmGetMgmtFp(&pDnode->wrappers[DNODE]); dmSetMgmtFp(&pDnode->wrappers[DNODE]);
mmGetMgmtFp(&pDnode->wrappers[MNODE]); mmSetMgmtFp(&pDnode->wrappers[MNODE]);
vmGetMgmtFp(&pDnode->wrappers[VNODES]); vmSetMgmtFp(&pDnode->wrappers[VNODES]);
qmGetMgmtFp(&pDnode->wrappers[QNODE]); qmSetMgmtFp(&pDnode->wrappers[QNODE]);
smGetMgmtFp(&pDnode->wrappers[SNODE]); smSetMgmtFp(&pDnode->wrappers[SNODE]);
bmGetMgmtFp(&pDnode->wrappers[BNODE]); bmSetMgmtFp(&pDnode->wrappers[BNODE]);
for (ENodeType n = 0; n < NODE_MAX; ++n) { for (ENodeType n = 0; n < NODE_MAX; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n]; SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
...@@ -95,6 +95,7 @@ SDnode *dndCreate(const SDnodeOpt *pOption) { ...@@ -95,6 +95,7 @@ SDnode *dndCreate(const SDnodeOpt *pOption) {
pWrapper->path = strdup(path); pWrapper->path = strdup(path);
pWrapper->shm.id = -1; pWrapper->shm.id = -1;
pWrapper->pDnode = pDnode; pWrapper->pDnode = pDnode;
pWrapper->ntype = n;
if (pWrapper->path == NULL) { if (pWrapper->path == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
goto _OVER; goto _OVER;
...@@ -203,3 +204,10 @@ void dndSetStatus(SDnode *pDnode, EDndStatus status) { ...@@ -203,3 +204,10 @@ void dndSetStatus(SDnode *pDnode, EDndStatus status) {
pDnode->status = status; pDnode->status = status;
} }
} }
void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc) {
SStartupReq *pStartup = &pDnode->startup;
tstrncpy(pStartup->name, pName, TSDB_STEP_NAME_LEN);
tstrncpy(pStartup->desc, pDesc, TSDB_STEP_DESC_LEN);
pStartup->finished = 0;
}
source/dnode/mgmt/main/src/dndTransport.c → source/dnode/mgmt/main/dndTransport.c
浏览文件 @ ef5ca96d
...@@ -20,56 +20,143 @@ ...@@ -20,56 +20,143 @@
#define INTERNAL_CKEY "_key" #define INTERNAL_CKEY "_key"
#define INTERNAL_SECRET "_pwd" #define INTERNAL_SECRET "_pwd"
static inline void dndProcessQMVnodeRpcMsg(SMsgHandle *pHandle, SRpcMsg *pMsg, SEpSet *pEpSet) { static void dndUpdateMnodeEpSet(SDnode *pDnode, SEpSet *pEpSet) {
SMsgHead *pHead = pMsg->pCont; SMgmtWrapper *pWrapper = &pDnode->wrappers[DNODE];
int32_t vgId = htonl(pHead->vgId); dmUpdateMnodeEpSet(pWrapper->pMgmt, pEpSet);
}
SMgmtWrapper *pWrapper = pHandle->pWrapper; static inline NodeMsgFp dndGetMsgFp(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) {
if (vgId == QND_VGID) { NodeMsgFp msgFp = pWrapper->msgFps[TMSG_INDEX(pRpc->msgType)];
pWrapper = pHandle->pQndWrapper; if (msgFp == NULL) {
} else if (vgId == MND_VGID) { terrno = TSDB_CODE_MSG_NOT_PROCESSED;
pWrapper = pHandle->pMndWrapper;
} }
dTrace("msg:%s will be processed by %s, handle:%p app:%p vgId:%d", TMSG_INFO(pMsg->msgType), pWrapper->name, return msgFp;
pMsg->handle, pMsg->ahandle, vgId);
dndProcessRpcMsg(pWrapper, pMsg, pEpSet);
} }
static void dndProcessResponse(void *parent, SRpcMsg *pRsp, SEpSet *pEpSet) { static inline int32_t dndBuildMsg(SNodeMsg *pMsg, SRpcMsg *pRpc) {
SDnode *pDnode = parent; SRpcConnInfo connInfo = {0};
STransMgmt *pMgmt = &pDnode->trans; if ((pRpc->msgType & 1U) && rpcGetConnInfo(pRpc->handle, &connInfo) != 0) {
tmsg_t msgType = pRsp->msgType; terrno = TSDB_CODE_MND_NO_USER_FROM_CONN;
dError("failed to build msg since %s, app:%p handle:%p", terrstr(), pRpc->ahandle, pRpc->handle);
return -1;
}
memcpy(pMsg->user, connInfo.user, TSDB_USER_LEN);
pMsg->clientIp = connInfo.clientIp;
pMsg->clientPort = connInfo.clientPort;
memcpy(&pMsg->rpcMsg, pRpc, sizeof(SRpcMsg));
return 0;
}
static void dndProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSet) {
int32_t code = -1;
SNodeMsg *pMsg = NULL;
NodeMsgFp msgFp = NULL;
if (pEpSet && pEpSet->numOfEps > 0 && pRpc->msgType == TDMT_MND_STATUS_RSP) {
dndUpdateMnodeEpSet(pWrapper->pDnode, pEpSet);
}
if (dndMarkWrapper(pWrapper) != 0) goto _OVER;
if ((msgFp = dndGetMsgFp(pWrapper, pRpc)) == NULL) goto _OVER;
if ((pMsg = taosAllocateQitem(sizeof(SNodeMsg))) == NULL) goto _OVER;
if (dndBuildMsg(pMsg, pRpc) != 0) goto _OVER;
if (pWrapper->procType == PROC_SINGLE) {
dTrace("msg:%p, is created, handle:%p user:%s", pMsg, pRpc->handle, pMsg->user);
code = (*msgFp)(pWrapper, pMsg);
} else if (pWrapper->procType == PROC_PARENT) {
dTrace("msg:%p, is created and put into child queue, handle:%p user:%s", pMsg, pRpc->handle, pMsg->user);
code = taosProcPutToChildQ(pWrapper->pProc, pMsg, sizeof(SNodeMsg), pRpc->pCont, pRpc->contLen, PROC_REQ);
} else {
dTrace("msg:%p, should not processed in child process, handle:%p user:%s", pMsg, pRpc->handle, pMsg->user);
ASSERT(1);
}
_OVER:
if (code == 0) {
if (pWrapper->procType == PROC_PARENT) {
dTrace("msg:%p, is freed in parent process", pMsg);
taosFreeQitem(pMsg);
rpcFreeCont(pRpc->pCont);
}
} else {
dError("msg:%p, failed to process since 0x%04x:%s", pMsg, code & 0XFFFF, terrstr());
if (pRpc->msgType & 1U) {
SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = terrno};
tmsgSendRsp(&rsp);
}
dTrace("msg:%p, is freed", pMsg);
taosFreeQitem(pMsg);
rpcFreeCont(pRpc->pCont);
}
dndReleaseWrapper(pWrapper);
}
static void dndProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
STransMgmt *pMgmt = &pDnode->trans;
tmsg_t msgType = pMsg->msgType;
bool isReq = msgType & 1u;
SMsgHandle *pHandle = &pMgmt->msgHandles[TMSG_INDEX(msgType)];
SMgmtWrapper *pWrapper = pHandle->pWrapper;
if (msgType == TDMT_DND_NETWORK_TEST) {
dTrace("network test req will be processed, handle:%p, app:%p", pMsg->handle, pMsg->ahandle);
dndProcessStartupReq(pDnode, pMsg);
return;
}
if (dndGetStatus(pDnode) != DND_STAT_RUNNING) { if (dndGetStatus(pDnode) != DND_STAT_RUNNING) {
dTrace("rsp:%s ignored since dnode not running, handle:%p app:%p", TMSG_INFO(msgType), pRsp->handle, pRsp->ahandle); dError("msg:%s ignored since dnode not running, handle:%p app:%p", TMSG_INFO(msgType), pMsg->handle, pMsg->ahandle);
rpcFreeCont(pRsp->pCont); if (isReq) {
SRpcMsg rspMsg = {.handle = pMsg->handle, .code = TSDB_CODE_APP_NOT_READY, .ahandle = pMsg->ahandle};
rpcSendResponse(&rspMsg);
}
rpcFreeCont(pMsg->pCont);
return; return;
} }
SMsgHandle *pHandle = &pMgmt->msgHandles[TMSG_INDEX(msgType)]; if (isReq && pMsg->pCont == NULL) {
if (pHandle->pWrapper != NULL) { dError("req:%s not processed since its empty, handle:%p app:%p", TMSG_INFO(msgType), pMsg->handle, pMsg->ahandle);
if (pHandle->pMndWrapper == NULL && pHandle->pQndWrapper == NULL) { SRpcMsg rspMsg = {.handle = pMsg->handle, .code = TSDB_CODE_DND_INVALID_MSG_LEN, .ahandle = pMsg->ahandle};
dTrace("rsp:%s will be processed by %s, handle:%p app:%p code:0x%04x:%s", TMSG_INFO(msgType), rpcSendResponse(&rspMsg);
pHandle->pWrapper->name, pRsp->handle, pRsp->ahandle, pRsp->code & 0XFFFF, tstrerror(pRsp->code)); return;
dndProcessRpcMsg(pHandle->pWrapper, pRsp, pEpSet); }
if (pWrapper == NULL) {
dError("msg:%s not processed since no handle, handle:%p app:%p", TMSG_INFO(msgType), pMsg->handle, pMsg->ahandle);
if (isReq) {
SRpcMsg rspMsg = {.handle = pMsg->handle, .code = TSDB_CODE_MSG_NOT_PROCESSED, .ahandle = pMsg->ahandle};
rpcSendResponse(&rspMsg);
}
rpcFreeCont(pMsg->pCont);
}
if (pHandle->pMndWrapper != NULL || pHandle->pQndWrapper != NULL) {
SMsgHead *pHead = pMsg->pCont;
int32_t vgId = ntohl(pHead->vgId);
if (vgId == QNODE_HANDLE) {
pWrapper = pHandle->pQndWrapper;
} else if (vgId == MNODE_HANDLE) {
pWrapper = pHandle->pMndWrapper;
} else { } else {
dndProcessQMVnodeRpcMsg(pHandle, pRsp, pEpSet);
} }
} else {
dError("rsp:%s not processed since no handle, handle:%p app:%p", TMSG_INFO(msgType), pRsp->handle, pRsp->ahandle);
rpcFreeCont(pRsp->pCont);
} }
dTrace("msg:%s will be processed by %s, app:%p", TMSG_INFO(msgType), pWrapper->name, pMsg->ahandle);
dndProcessRpcMsg(pWrapper, pMsg, pEpSet);
} }
int32_t dndInitClient(SDnode *pDnode) { static int32_t dndInitClient(SDnode *pDnode) {
STransMgmt *pMgmt = &pDnode->trans; STransMgmt *pMgmt = &pDnode->trans;
SRpcInit rpcInit; SRpcInit rpcInit;
memset(&rpcInit, 0, sizeof(rpcInit)); memset(&rpcInit, 0, sizeof(rpcInit));
rpcInit.label = "DND"; rpcInit.label = "DND";
rpcInit.numOfThreads = 1; rpcInit.numOfThreads = 1;
rpcInit.cfp = dndProcessResponse; rpcInit.cfp = (RpcCfp)dndProcessMsg;
rpcInit.sessions = 1024; rpcInit.sessions = 1024;
rpcInit.connType = TAOS_CONN_CLIENT; rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.idleTime = tsShellActivityTimer * 1000; rpcInit.idleTime = tsShellActivityTimer * 1000;
...@@ -92,7 +179,7 @@ int32_t dndInitClient(SDnode *pDnode) { ...@@ -92,7 +179,7 @@ int32_t dndInitClient(SDnode *pDnode) {
return 0; return 0;
} }
void dndCleanupClient(SDnode *pDnode) { static void dndCleanupClient(SDnode *pDnode) {
STransMgmt *pMgmt = &pDnode->trans; STransMgmt *pMgmt = &pDnode->trans;
if (pMgmt->clientRpc) { if (pMgmt->clientRpc) {
rpcClose(pMgmt->clientRpc); rpcClose(pMgmt->clientRpc);
...@@ -101,63 +188,15 @@ void dndCleanupClient(SDnode *pDnode) { ...@@ -101,63 +188,15 @@ void dndCleanupClient(SDnode *pDnode) {
} }
} }
static void dndProcessRequest(void *param, SRpcMsg *pReq, SEpSet *pEpSet) { static inline void dndSendMsgToMnodeRecv(SDnode *pDnode, SRpcMsg *pReq, SRpcMsg *pRsp) {
SDnode *pDnode = param; SEpSet epSet = {0};
STransMgmt *pMgmt = &pDnode->trans; SMgmtWrapper *pWrapper = &pDnode->wrappers[DNODE];
tmsg_t msgType = pReq->msgType; dmGetMnodeEpSet(pWrapper->pMgmt, &epSet);
rpcSendRecv(pDnode->trans.clientRpc, &epSet, pReq, pRsp);
if (msgType == TDMT_DND_NETWORK_TEST) {
dTrace("network test req will be processed, handle:%p, app:%p", pReq->handle, pReq->ahandle);
dndProcessStartupReq(pDnode, pReq);
return;
}
if (dndGetStatus(pDnode) != DND_STAT_RUNNING) {
dError("req:%s ignored since dnode not running, handle:%p app:%p", TMSG_INFO(msgType), pReq->handle, pReq->ahandle);
SRpcMsg rspMsg = {.handle = pReq->handle, .code = TSDB_CODE_APP_NOT_READY, .ahandle = pReq->ahandle};
rpcSendResponse(&rspMsg);
rpcFreeCont(pReq->pCont);
return;
}
if (pReq->pCont == NULL) {
dTrace("req:%s not processed since its empty, handle:%p app:%p", TMSG_INFO(msgType), pReq->handle, pReq->ahandle);
SRpcMsg rspMsg = {.handle = pReq->handle, .code = TSDB_CODE_DND_INVALID_MSG_LEN, .ahandle = pReq->ahandle};
rpcSendResponse(&rspMsg);
return;
}
SMsgHandle *pHandle = &pMgmt->msgHandles[TMSG_INDEX(msgType)];
if (pHandle->pWrapper != NULL) {
if (pHandle->pMndWrapper == NULL && pHandle->pQndWrapper == NULL) {
dTrace("req:%s will be processed by %s, handle:%p app:%p", TMSG_INFO(msgType), pHandle->pWrapper->name,
pReq->handle, pReq->ahandle);
dndProcessRpcMsg(pHandle->pWrapper, pReq, pEpSet);
} else {
dndProcessQMVnodeRpcMsg(pHandle, pReq, pEpSet);
}
} else {
dError("req:%s not processed since no handle, handle:%p app:%p", TMSG_INFO(msgType), pReq->handle, pReq->ahandle);
SRpcMsg rspMsg = {.handle = pReq->handle, .code = TSDB_CODE_MSG_NOT_PROCESSED, .ahandle = pReq->ahandle};
rpcSendResponse(&rspMsg);
rpcFreeCont(pReq->pCont);
}
} }
static void dndSendMsgToMnodeRecv(SDnode *pDnode, SRpcMsg *pRpcMsg, SRpcMsg *pRpcRsp) { static inline int32_t dndGetHideUserAuth(SDnode *pDnode, char *user, char *spi, char *encrypt, char *secret,
STransMgmt *pMgmt = &pDnode->trans; char *ckey) {
SEpSet epSet = {0};
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, DNODE);
if (pWrapper != NULL) {
dmGetMnodeEpSet(pWrapper->pMgmt, &epSet);
dndReleaseWrapper(pWrapper);
}
rpcSendRecv(pMgmt->clientRpc, &epSet, pRpcMsg, pRpcRsp);
}
static int32_t dndGetHideUserAuth(SDnode *pDnode, char *user, char *spi, char *encrypt, char *secret, char *ckey) {
int32_t code = 0; int32_t code = 0;
char pass[TSDB_PASSWORD_LEN + 1] = {0}; char pass[TSDB_PASSWORD_LEN + 1] = {0};
...@@ -179,29 +218,12 @@ static int32_t dndGetHideUserAuth(SDnode *pDnode, char *user, char *spi, char *e ...@@ -179,29 +218,12 @@ static int32_t dndGetHideUserAuth(SDnode *pDnode, char *user, char *spi, char *e
return code; return code;
} }
static int32_t dndRetrieveUserAuthInfo(void *parent, char *user, char *spi, char *encrypt, char *secret, char *ckey) { static int32_t dndRetrieveUserAuthInfo(SDnode *pDnode, char *user, char *spi, char *encrypt, char *secret, char *ckey) {
SDnode *pDnode = parent; if (dndGetHideUserAuth(pDnode, user, spi, encrypt, secret, ckey) == 0) {
if (dndGetHideUserAuth(parent, user, spi, encrypt, secret, ckey) == 0) {
dTrace("user:%s, get auth from mnode, spi:%d encrypt:%d", user, *spi, *encrypt); dTrace("user:%s, get auth from mnode, spi:%d encrypt:%d", user, *spi, *encrypt);
return 0; return 0;
} }
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, MNODE);
if (pWrapper != NULL) {
if (mmGetUserAuth(pWrapper, user, spi, encrypt, secret, ckey) == 0) {
dndReleaseWrapper(pWrapper);
dTrace("user:%s, get auth from mnode, spi:%d encrypt:%d", user, *spi, *encrypt);
return 0;
}
dndReleaseWrapper(pWrapper);
}
if (terrno != TSDB_CODE_APP_NOT_READY) {
dTrace("failed to get user auth from mnode since %s", terrstr());
return -1;
}
SAuthReq authReq = {0}; SAuthReq authReq = {0};
tstrncpy(authReq.user, user, TSDB_USER_LEN); tstrncpy(authReq.user, user, TSDB_USER_LEN);
int32_t contLen = tSerializeSAuthReq(NULL, 0, &authReq); int32_t contLen = tSerializeSAuthReq(NULL, 0, &authReq);
...@@ -231,7 +253,7 @@ static int32_t dndRetrieveUserAuthInfo(void *parent, char *user, char *spi, char ...@@ -231,7 +253,7 @@ static int32_t dndRetrieveUserAuthInfo(void *parent, char *user, char *spi, char
return rpcRsp.code; return rpcRsp.code;
} }
int32_t dndInitServer(SDnode *pDnode) { static int32_t dndInitServer(SDnode *pDnode) {
STransMgmt *pMgmt = &pDnode->trans; STransMgmt *pMgmt = &pDnode->trans;
int32_t numOfThreads = (int32_t)((tsNumOfCores * tsNumOfThreadsPerCore) / 2.0); int32_t numOfThreads = (int32_t)((tsNumOfCores * tsNumOfThreadsPerCore) / 2.0);
...@@ -244,11 +266,11 @@ int32_t dndInitServer(SDnode *pDnode) { ...@@ -244,11 +266,11 @@ int32_t dndInitServer(SDnode *pDnode) {
rpcInit.localPort = pDnode->serverPort; rpcInit.localPort = pDnode->serverPort;
rpcInit.label = "DND"; rpcInit.label = "DND";
rpcInit.numOfThreads = numOfThreads; rpcInit.numOfThreads = numOfThreads;
rpcInit.cfp = dndProcessRequest; rpcInit.cfp = (RpcCfp)dndProcessMsg;
rpcInit.sessions = tsMaxShellConns; rpcInit.sessions = tsMaxShellConns;
rpcInit.connType = TAOS_CONN_SERVER; rpcInit.connType = TAOS_CONN_SERVER;
rpcInit.idleTime = tsShellActivityTimer * 1000; rpcInit.idleTime = tsShellActivityTimer * 1000;
rpcInit.afp = dndRetrieveUserAuthInfo; rpcInit.afp = (RpcAfp)dndRetrieveUserAuthInfo;
rpcInit.parent = pDnode; rpcInit.parent = pDnode;
pMgmt->serverRpc = rpcOpen(&rpcInit); pMgmt->serverRpc = rpcOpen(&rpcInit);
...@@ -261,7 +283,7 @@ int32_t dndInitServer(SDnode *pDnode) { ...@@ -261,7 +283,7 @@ int32_t dndInitServer(SDnode *pDnode) {
return 0; return 0;
} }
void dndCleanupServer(SDnode *pDnode) { static void dndCleanupServer(SDnode *pDnode) {
STransMgmt *pMgmt = &pDnode->trans; STransMgmt *pMgmt = &pDnode->trans;
if (pMgmt->serverRpc) { if (pMgmt->serverRpc) {
rpcClose(pMgmt->serverRpc); rpcClose(pMgmt->serverRpc);
...@@ -270,6 +292,17 @@ void dndCleanupServer(SDnode *pDnode) { ...@@ -270,6 +292,17 @@ void dndCleanupServer(SDnode *pDnode) {
} }
} }
int32_t dndInitTrans(SDnode *pDnode) {
if (dndInitServer(pDnode) != 0) return -1;
if (dndInitClient(pDnode) != 0) return -1;
return 0;
}
void dndCleanupTrans(SDnode *pDnode) {
dndCleanupServer(pDnode);
dndCleanupClient(pDnode);
}
int32_t dndInitMsgHandle(SDnode *pDnode) { int32_t dndInitMsgHandle(SDnode *pDnode) {
STransMgmt *pMgmt = &pDnode->trans; STransMgmt *pMgmt = &pDnode->trans;
...@@ -278,19 +311,17 @@ int32_t dndInitMsgHandle(SDnode *pDnode) { ...@@ -278,19 +311,17 @@ int32_t dndInitMsgHandle(SDnode *pDnode) {
for (int32_t msgIndex = 0; msgIndex < TDMT_MAX; ++msgIndex) { for (int32_t msgIndex = 0; msgIndex < TDMT_MAX; ++msgIndex) {
NodeMsgFp msgFp = pWrapper->msgFps[msgIndex]; NodeMsgFp msgFp = pWrapper->msgFps[msgIndex];
int32_t vgId = pWrapper->msgVgIds[msgIndex]; int8_t vgId = pWrapper->msgVgIds[msgIndex];
if (msgFp == NULL) continue; if (msgFp == NULL) continue;
// dTrace("msg:%s will be processed by %s, vgId:%d", tMsgInfo[msgIndex], pWrapper->name, vgId);
SMsgHandle *pHandle = &pMgmt->msgHandles[msgIndex]; SMsgHandle *pHandle = &pMgmt->msgHandles[msgIndex];
if (vgId == QND_VGID) { if (vgId == QNODE_HANDLE) {
if (pHandle->pQndWrapper != NULL) { if (pHandle->pQndWrapper != NULL) {
dError("msg:%s has multiple process nodes", tMsgInfo[msgIndex]); dError("msg:%s has multiple process nodes", tMsgInfo[msgIndex]);
return -1; return -1;
} }
pHandle->pQndWrapper = pWrapper; pHandle->pQndWrapper = pWrapper;
} else if (vgId == MND_VGID) { } else if (vgId == MNODE_HANDLE) {
if (pHandle->pMndWrapper != NULL) { if (pHandle->pMndWrapper != NULL) {
dError("msg:%s has multiple process nodes", tMsgInfo[msgIndex]); dError("msg:%s has multiple process nodes", tMsgInfo[msgIndex]);
return -1; return -1;
...@@ -319,60 +350,43 @@ static int32_t dndSendRpcReq(STransMgmt *pMgmt, const SEpSet *pEpSet, SRpcMsg *p ...@@ -319,60 +350,43 @@ static int32_t dndSendRpcReq(STransMgmt *pMgmt, const SEpSet *pEpSet, SRpcMsg *p
return 0; return 0;
} }
int32_t dndSendReqToMnode(SMgmtWrapper *pWrapper, SRpcMsg *pReq) { static void dndSendRpcRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp) {
SDnode *pDnode = pWrapper->pDnode;
STransMgmt *pTrans = &pDnode->trans;
SEpSet epSet = {0};
SMgmtWrapper *pWrapper2 = dndAcquireWrapper(pDnode, DNODE);
if (pWrapper2 != NULL) {
dmGetMnodeEpSet(pWrapper2->pMgmt, &epSet);
dndReleaseWrapper(pWrapper2);
}
return dndSendRpcReq(pTrans, &epSet, pReq);
}
void dndSendRpcRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp) {
if (pRsp->code == TSDB_CODE_APP_NOT_READY) { if (pRsp->code == TSDB_CODE_APP_NOT_READY) {
SMgmtWrapper *pDnodeWrapper = dndAcquireWrapper(pWrapper->pDnode, DNODE); if (pWrapper->ntype == MNODE) {
if (pDnodeWrapper != NULL) { dmSendRedirectRsp(pWrapper->pMgmt, pRsp);
dmSendRedirectRsp(pDnodeWrapper->pMgmt, pRsp); return;
dndReleaseWrapper(pDnodeWrapper);
} else {
rpcSendResponse(pRsp);
} }
} else {
rpcSendResponse(pRsp);
} }
rpcSendResponse(pRsp);
} }
int32_t dndSendReq(SMgmtWrapper *pWrapper, const SEpSet *pEpSet, SRpcMsg *pReq) { static int32_t dndSendReq(SMgmtWrapper *pWrapper, const SEpSet *pEpSet, SRpcMsg *pReq) {
SDnode *pDnode = pWrapper->pDnode; if (dndGetStatus(pWrapper->pDnode) != DND_STAT_RUNNING) {
if (dndGetStatus(pDnode) != DND_STAT_RUNNING) {
terrno = TSDB_CODE_DND_OFFLINE; terrno = TSDB_CODE_DND_OFFLINE;
dError("failed to send rpc msg since %s, handle:%p", terrstr(), pReq->handle); dError("failed to send rpc msg since %s, handle:%p", terrstr(), pReq->handle);
return -1; return -1;
} }
if (pWrapper->procType != PROC_CHILD) { if (pWrapper->procType != PROC_CHILD) {
return dndSendRpcReq(&pDnode->trans, pEpSet, pReq); return dndSendRpcReq(&pWrapper->pDnode->trans, pEpSet, pReq);
} else { } else {
int32_t headLen = sizeof(SRpcMsg) + sizeof(SEpSet); char *pHead = taosMemoryMalloc(sizeof(SRpcMsg) + sizeof(SEpSet));
char *pHead = taosMemoryMalloc(headLen);
if (pHead == NULL) { if (pHead == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1; return -1;
} }
memcpy(pHead, pReq, sizeof(SRpcMsg)); memcpy(pHead, pReq, sizeof(SRpcMsg));
memcpy(pHead + sizeof(SRpcMsg), pEpSet, sizeof(SEpSet)); memcpy(pHead + sizeof(SRpcMsg), pEpSet, sizeof(SEpSet));
taosProcPutToParentQ(pWrapper->pProc, pHead, sizeof(SRpcMsg) + sizeof(SEpSet), pReq->pCont, pReq->contLen,
taosProcPutToParentQ(pWrapper->pProc, pReq, headLen, pReq->pCont, pReq->contLen, PROC_REQ); PROC_REQ);
taosMemoryFree(pHead); taosMemoryFree(pHead);
return 0; return 0;
} }
} }
void dndSendRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp) { static void dndSendRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp) {
if (pWrapper->procType != PROC_CHILD) { if (pWrapper->procType != PROC_CHILD) {
dndSendRpcRsp(pWrapper, pRsp); dndSendRpcRsp(pWrapper, pRsp);
} else { } else {
...@@ -380,7 +394,7 @@ void dndSendRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp) { ...@@ -380,7 +394,7 @@ void dndSendRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp) {
} }
} }
void dndRegisterBrokenLinkArg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg) { static void dndRegisterBrokenLinkArg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg) {
if (pWrapper->procType != PROC_CHILD) { if (pWrapper->procType != PROC_CHILD) {
rpcRegisterBrokenLinkArg(pMsg); rpcRegisterBrokenLinkArg(pMsg);
} else { } else {
...@@ -406,4 +420,68 @@ SMsgCb dndCreateMsgcb(SMgmtWrapper *pWrapper) { ...@@ -406,4 +420,68 @@ SMsgCb dndCreateMsgcb(SMgmtWrapper *pWrapper) {
.releaseHandleFp = dndReleaseHandle, .releaseHandleFp = dndReleaseHandle,
}; };
return msgCb; return msgCb;
}
static void dndConsumeChildQueue(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int16_t msgLen, void *pCont, int32_t contLen,
ProcFuncType ftype) {
SRpcMsg *pRpc = &pMsg->rpcMsg;
pRpc->pCont = pCont;
dTrace("msg:%p, get from child queue, handle:%p app:%p", pMsg, pRpc->handle, pRpc->ahandle);
NodeMsgFp msgFp = pWrapper->msgFps[TMSG_INDEX(pRpc->msgType)];
int32_t code = (*msgFp)(pWrapper, pMsg);
if (code != 0) {
dError("msg:%p, failed to process since code:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code));
if (pRpc->msgType & 1U) {
SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = terrno};
dndSendRsp(pWrapper, &rsp);
}
dTrace("msg:%p, is freed", pMsg);
taosFreeQitem(pMsg);
rpcFreeCont(pCont);
}
}
static void dndConsumeParentQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, int16_t msgLen, void *pCont, int32_t contLen,
ProcFuncType ftype) {
pMsg->pCont = pCont;
dTrace("msg:%p, get from parent queue, ftype:%d handle:%p, app:%p", pMsg, ftype, pMsg->handle, pMsg->ahandle);
switch (ftype) {
case PROC_REGIST:
rpcRegisterBrokenLinkArg(pMsg);
break;
case PROC_RELEASE:
rpcReleaseHandle(pMsg->handle, (int8_t)pMsg->code);
rpcFreeCont(pCont);
break;
case PROC_REQ:
dndSendRpcReq(&pWrapper->pDnode->trans, (SEpSet *)((char *)pMsg + sizeof(SRpcMsg)), pMsg);
break;
case PROC_RSP:
dndSendRpcRsp(pWrapper, pMsg);
break;
default:
break;
}
taosMemoryFree(pMsg);
}
SProcCfg dndGenProcCfg(SMgmtWrapper *pWrapper) {
SProcCfg cfg = {.childConsumeFp = (ProcConsumeFp)dndConsumeChildQueue,
.childMallocHeadFp = (ProcMallocFp)taosAllocateQitem,
.childFreeHeadFp = (ProcFreeFp)taosFreeQitem,
.childMallocBodyFp = (ProcMallocFp)rpcMallocCont,
.childFreeBodyFp = (ProcFreeFp)rpcFreeCont,
.parentConsumeFp = (ProcConsumeFp)dndConsumeParentQueue,
.parentMallocHeadFp = (ProcMallocFp)taosMemoryMalloc,
.parentFreeHeadFp = (ProcFreeFp)taosMemoryFree,
.parentMallocBodyFp = (ProcMallocFp)rpcMallocCont,
.parentFreeBodyFp = (ProcFreeFp)rpcFreeCont,
.shm = pWrapper->shm,
.pParent = pWrapper,
.name = pWrapper->name};
return cfg;
} }
\ No newline at end of file
source/dnode/mgmt/main/src/dndEnv.c 已删除 100644 → 0
浏览文件 @ aaf35118
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "dndInt.h"
#include "wal.h"
static int8_t once = DND_ENV_INIT;
int32_t dndInit() {
dDebug("start to init dnode env");
if (atomic_val_compare_exchange_8(&once, DND_ENV_INIT, DND_ENV_READY) != DND_ENV_INIT) {
terrno = TSDB_CODE_REPEAT_INIT;
dError("failed to init dnode env since %s", terrstr());
return -1;
}
taosIgnSIGPIPE();
taosBlockSIGPIPE();
taosResolveCRC();
SMonCfg monCfg = {0};
monCfg.maxLogs = tsMonitorMaxLogs;
monCfg.port = tsMonitorPort;
monCfg.server = tsMonitorFqdn;
monCfg.comp = tsMonitorComp;
if (monInit(&monCfg) != 0) {
dError("failed to init monitor since %s", terrstr());
return -1;
}
dInfo("dnode env is initialized");
return 0;
}
void dndCleanup() {
dDebug("start to cleanup dnode env");
if (atomic_val_compare_exchange_8(&once, DND_ENV_READY, DND_ENV_CLEANUP) != DND_ENV_READY) {
dError("dnode env is already cleaned up");
return;
}
monCleanup();
walCleanUp();
taosStopCacheRefreshWorker();
dInfo("dnode env is cleaned up");
}
source/dnode/mgmt/main/src/dndMsg.c 已删除 100644 → 0
浏览文件 @ aaf35118
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "dndInt.h"
static void dndUpdateMnodeEpSet(SDnode *pDnode, SEpSet *pEpSet) {
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, DNODE);
if (pWrapper != NULL) {
dmUpdateMnodeEpSet(pWrapper->pMgmt, pEpSet);
dndReleaseWrapper(pWrapper);
}
}
static inline NodeMsgFp dndGetMsgFp(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) {
NodeMsgFp msgFp = pWrapper->msgFps[TMSG_INDEX(pRpc->msgType)];
if (msgFp == NULL) {
terrno = TSDB_CODE_MSG_NOT_PROCESSED;
}
return msgFp;
}
static inline int32_t dndBuildMsg(SNodeMsg *pMsg, SRpcMsg *pRpc) {
SRpcConnInfo connInfo = {0};
if ((pRpc->msgType & 1U) && rpcGetConnInfo(pRpc->handle, &connInfo) != 0) {
terrno = TSDB_CODE_MND_NO_USER_FROM_CONN;
dError("failed to build msg since %s, app:%p RPC:%p", terrstr(), pRpc->ahandle, pRpc->handle);
return -1;
}
memcpy(pMsg->user, connInfo.user, TSDB_USER_LEN);
pMsg->clientIp = connInfo.clientIp;
pMsg->clientPort = connInfo.clientPort;
memcpy(&pMsg->rpcMsg, pRpc, sizeof(SRpcMsg));
return 0;
}
void dndProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSet) {
int32_t code = -1;
SNodeMsg *pMsg = NULL;
NodeMsgFp msgFp = NULL;
if (pEpSet && pEpSet->numOfEps > 0 && pRpc->msgType == TDMT_MND_STATUS_RSP) {
dndUpdateMnodeEpSet(pWrapper->pDnode, pEpSet);
}
if (dndMarkWrapper(pWrapper) != 0) goto _OVER;
if ((msgFp = dndGetMsgFp(pWrapper, pRpc)) == NULL) goto _OVER;
if ((pMsg = taosAllocateQitem(sizeof(SNodeMsg))) == NULL) goto _OVER;
if (dndBuildMsg(pMsg, pRpc) != 0) goto _OVER;
if (pWrapper->procType == PROC_SINGLE) {
dTrace("msg:%p, is created, handle:%p app:%p user:%s", pMsg, pRpc->handle, pRpc->ahandle, pMsg->user);
code = (*msgFp)(pWrapper, pMsg);
} else if (pWrapper->procType == PROC_PARENT) {
dTrace("msg:%p, is created and put into child queue, handle:%p app:%p user:%s", pMsg, pRpc->handle, pRpc->ahandle,
pMsg->user);
code = taosProcPutToChildQ(pWrapper->pProc, pMsg, sizeof(SNodeMsg), pRpc->pCont, pRpc->contLen, PROC_REQ);
} else {
dTrace("msg:%p, should not processed in child process, handle:%p app:%p user:%s", pMsg, pRpc->handle, pRpc->ahandle,
pMsg->user);
ASSERT(1);
}
_OVER:
if (code == 0) {
if (pWrapper->procType == PROC_PARENT) {
dTrace("msg:%p, is freed in parent process", pMsg);
taosFreeQitem(pMsg);
rpcFreeCont(pRpc->pCont);
}
} else {
dError("msg:%p, failed to process since 0x%04x:%s", pMsg, code & 0XFFFF, terrstr());
if (pRpc->msgType & 1U) {
SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = terrno};
dndSendRsp(pWrapper, &rsp);
}
dTrace("msg:%p, is freed", pMsg);
taosFreeQitem(pMsg);
rpcFreeCont(pRpc->pCont);
}
dndReleaseWrapper(pWrapper);
}
static int32_t dndProcessCreateNodeMsg(SDnode *pDnode, ENodeType ntype, SNodeMsg *pMsg) {
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, ntype);
if (pWrapper != NULL) {
dndReleaseWrapper(pWrapper);
terrno = TSDB_CODE_NODE_ALREADY_DEPLOYED;
dError("failed to create node since %s", terrstr());
return -1;
}
pWrapper = &pDnode->wrappers[ntype];
if (taosMkDir(pWrapper->path) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to create dir:%s since %s", pWrapper->path, terrstr());
return -1;
}
int32_t code = (*pWrapper->fp.createMsgFp)(pWrapper, pMsg);
if (code != 0) {
dError("node:%s, failed to open since %s", pWrapper->name, terrstr());
} else {
dDebug("node:%s, has been opened", pWrapper->name);
pWrapper->deployed = true;
}
return code;
}
static int32_t dndProcessDropNodeMsg(SDnode *pDnode, ENodeType ntype, SNodeMsg *pMsg) {
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, ntype);
if (pWrapper == NULL) {
terrno = TSDB_CODE_NODE_NOT_DEPLOYED;
dError("failed to drop node since %s", terrstr());
return -1;
}
taosWLockLatch(&pWrapper->latch);
pWrapper->deployed = false;
int32_t code = (*pWrapper->fp.dropMsgFp)(pWrapper, pMsg);
if (code != 0) {
pWrapper->deployed = true;
dError("node:%s, failed to drop since %s", pWrapper->name, terrstr());
} else {
pWrapper->deployed = false;
dDebug("node:%s, has been dropped", pWrapper->name);
}
taosWUnLockLatch(&pWrapper->latch);
dndReleaseWrapper(pWrapper);
return code;
}
int32_t dndProcessNodeMsg(SDnode *pDnode, SNodeMsg *pMsg) {
switch (pMsg->rpcMsg.msgType) {
case TDMT_DND_CREATE_MNODE:
return dndProcessCreateNodeMsg(pDnode, MNODE, pMsg);
case TDMT_DND_DROP_MNODE:
return dndProcessDropNodeMsg(pDnode, MNODE, pMsg);
case TDMT_DND_CREATE_QNODE:
return dndProcessCreateNodeMsg(pDnode, QNODE, pMsg);
case TDMT_DND_DROP_QNODE:
return dndProcessDropNodeMsg(pDnode, QNODE, pMsg);
case TDMT_DND_CREATE_SNODE:
return dndProcessCreateNodeMsg(pDnode, SNODE, pMsg);
case TDMT_DND_DROP_SNODE:
return dndProcessDropNodeMsg(pDnode, SNODE, pMsg);
case TDMT_DND_CREATE_BNODE:
return dndProcessCreateNodeMsg(pDnode, BNODE, pMsg);
case TDMT_DND_DROP_BNODE:
return dndProcessDropNodeMsg(pDnode, BNODE, pMsg);
default:
terrno = TSDB_CODE_MSG_NOT_PROCESSED;
return -1;
}
}
void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc) {
SStartupReq *pStartup = &pDnode->startup;
tstrncpy(pStartup->name, pName, TSDB_STEP_NAME_LEN);
tstrncpy(pStartup->desc, pDesc, TSDB_STEP_DESC_LEN);
pStartup->finished = 0;
}
void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup) {
memcpy(pStartup, &pDnode->startup, sizeof(SStartupReq));
pStartup->finished = (dndGetStatus(pDnode) == DND_STAT_RUNNING);
}
void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) {
dDebug("startup req is received");
SStartupReq *pStartup = rpcMallocCont(sizeof(SStartupReq));
dndGetStartup(pDnode, pStartup);
dDebug("startup req is sent, step:%s desc:%s finished:%d", pStartup->name, pStartup->desc, pStartup->finished);
SRpcMsg rpcRsp = {
.handle = pReq->handle, .pCont = pStartup, .contLen = sizeof(SStartupReq), .ahandle = pReq->ahandle};
rpcSendResponse(&rpcRsp);
}
source/dnode/mgmt/mm/inc/mm.h 已删除 100644 → 0
浏览文件 @ aaf35118
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_DND_MNODE_H_
#define _TD_DND_MNODE_H_
#include "dnd.h"
#ifdef __cplusplus
extern "C" {
#endif
void mmGetMgmtFp(SMgmtWrapper *pMgmt);
int32_t mmGetUserAuth(SMgmtWrapper *pWrapper, char *user, char *spi, char *encrypt, char *secret, char *ckey);
int32_t mmMonitorMnodeInfo(SMgmtWrapper *pWrapper, SMonClusterInfo *pClusterInfo, SMonVgroupInfo *pVgroupInfo,
SMonGrantInfo *pGrantInfo);
#ifdef __cplusplus
}
#endif
#endif /*_TD_DND_MNODE_H_*/
\ No newline at end of file
source/dnode/mgmt/mm/src/mmMsg.c → source/dnode/mgmt/mm/mmHandle.c
浏览文件 @ ef5ca96d
...@@ -75,91 +75,91 @@ int32_t mmProcessAlterReq(SMnodeMgmt *pMgmt, SNodeMsg *pMsg) { ...@@ -75,91 +75,91 @@ int32_t mmProcessAlterReq(SMnodeMgmt *pMgmt, SNodeMsg *pMsg) {
void mmInitMsgHandles(SMgmtWrapper *pWrapper) { void mmInitMsgHandles(SMgmtWrapper *pWrapper) {
// Requests handled by DNODE // Requests handled by DNODE
dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_MNODE_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_MNODE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_ALTER_MNODE_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_ALTER_MNODE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_DROP_MNODE_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_DROP_MNODE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_QNODE_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_QNODE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_DROP_QNODE_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_DROP_QNODE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_SNODE_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_SNODE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_DROP_SNODE_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_DROP_SNODE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_BNODE_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_BNODE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_DROP_BNODE_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_DROP_BNODE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_VNODE_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_VNODE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_ALTER_VNODE_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_ALTER_VNODE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_DROP_VNODE_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_DROP_VNODE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_SYNC_VNODE_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_SYNC_VNODE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_COMPACT_VNODE_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_COMPACT_VNODE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_CONFIG_DNODE_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_CONFIG_DNODE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
// Requests handled by MNODE // Requests handled by MNODE
dndSetMsgHandle(pWrapper, TDMT_MND_CONNECT, mmProcessReadMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_CONNECT, mmProcessReadMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_ACCT, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_ACCT, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_ALTER_ACCT, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_ALTER_ACCT, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_DROP_ACCT, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_DROP_ACCT, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_USER, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_USER, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_ALTER_USER, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_ALTER_USER, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_DROP_USER, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_DROP_USER, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_GET_USER_AUTH, mmProcessReadMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_GET_USER_AUTH, mmProcessReadMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_DNODE, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_DNODE, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_CONFIG_DNODE, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_CONFIG_DNODE, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_DROP_DNODE, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_DROP_DNODE, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_MNODE, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_MNODE, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_DROP_MNODE, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_DROP_MNODE, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_QNODE, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_QNODE, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_DROP_QNODE, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_DROP_QNODE, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_SNODE, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_SNODE, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_DROP_SNODE, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_DROP_SNODE, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_BNODE, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_BNODE, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_DROP_BNODE, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_DROP_BNODE, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_DB, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_DB, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_DROP_DB, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_DROP_DB, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_USE_DB, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_USE_DB, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_ALTER_DB, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_ALTER_DB, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_SYNC_DB, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_SYNC_DB, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_COMPACT_DB, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_COMPACT_DB, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_FUNC, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_FUNC, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_RETRIEVE_FUNC, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_RETRIEVE_FUNC, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_DROP_FUNC, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_DROP_FUNC, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_STB, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_STB, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_ALTER_STB, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_ALTER_STB, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_DROP_STB, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_DROP_STB, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_SMA, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_SMA, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_DROP_SMA, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_DROP_SMA, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_TABLE_META, mmProcessReadMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_TABLE_META, mmProcessReadMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_VGROUP_LIST, mmProcessReadMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_VGROUP_LIST, mmProcessReadMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_KILL_QUERY, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_KILL_QUERY, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_KILL_CONN, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_KILL_CONN, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_HEARTBEAT, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_HEARTBEAT, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_SHOW, mmProcessReadMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_SHOW, mmProcessReadMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_SHOW_RETRIEVE, mmProcessReadMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_SHOW_RETRIEVE, mmProcessReadMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_SYSTABLE_RETRIEVE, mmProcessReadMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_SYSTABLE_RETRIEVE, mmProcessReadMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_STATUS, mmProcessReadMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_STATUS, mmProcessReadMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_KILL_TRANS, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_KILL_TRANS, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_GRANT, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_GRANT, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_AUTH, mmProcessReadMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_AUTH, mmProcessReadMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_ALTER_MNODE, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_ALTER_MNODE, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_TOPIC, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_TOPIC, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_ALTER_TOPIC, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_ALTER_TOPIC, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_DROP_TOPIC, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_DROP_TOPIC, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_SUBSCRIBE, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_SUBSCRIBE, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_MQ_COMMIT_OFFSET, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_MQ_COMMIT_OFFSET, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_GET_SUB_EP, mmProcessReadMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_GET_SUB_EP, mmProcessReadMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_STREAM, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_MND_CREATE_STREAM, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_TASK_DEPLOY_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_TASK_DEPLOY_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
// Requests handled by VNODE // Requests handled by VNODE
dndSetMsgHandle(pWrapper, TDMT_VND_MQ_SET_CONN_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_MQ_SET_CONN_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_MQ_REB_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_MQ_REB_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_CREATE_STB_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_CREATE_STB_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_ALTER_STB_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_ALTER_STB_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_DROP_STB_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_DROP_STB_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_CREATE_SMA_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_CREATE_SMA_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_DROP_SMA_RSP, mmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_DROP_SMA_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_QUERY, mmProcessQueryMsg, MND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_QUERY, mmProcessQueryMsg, MNODE_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_QUERY_CONTINUE, mmProcessQueryMsg, MND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_QUERY_CONTINUE, mmProcessQueryMsg, MNODE_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_FETCH, mmProcessQueryMsg, MND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_FETCH, mmProcessQueryMsg, MNODE_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_DROP_TASK, mmProcessQueryMsg, MND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_DROP_TASK, mmProcessQueryMsg, MNODE_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_QUERY_HEARTBEAT, mmProcessQueryMsg, MND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_QUERY_HEARTBEAT, mmProcessQueryMsg, MNODE_HANDLE);
} }
source/dnode/mgmt/mm/src/mmInt.c → source/dnode/mgmt/mm/mmInt.c
浏览文件 @ ef5ca96d
...@@ -227,7 +227,7 @@ static int32_t mmStart(SMgmtWrapper *pWrapper) { ...@@ -227,7 +227,7 @@ static int32_t mmStart(SMgmtWrapper *pWrapper) {
return mndStart(pMgmt->pMnode); return mndStart(pMgmt->pMnode);
} }
void mmGetMgmtFp(SMgmtWrapper *pWrapper) { void mmSetMgmtFp(SMgmtWrapper *pWrapper) {
SMgmtFp mgmtFp = {0}; SMgmtFp mgmtFp = {0};
mgmtFp.openFp = mmOpen; mgmtFp.openFp = mmOpen;
mgmtFp.closeFp = mmClose; mgmtFp.closeFp = mmClose;
...@@ -241,14 +241,6 @@ void mmGetMgmtFp(SMgmtWrapper *pWrapper) { ...@@ -241,14 +241,6 @@ void mmGetMgmtFp(SMgmtWrapper *pWrapper) {
pWrapper->fp = mgmtFp; pWrapper->fp = mgmtFp;
} }
int32_t mmGetUserAuth(SMgmtWrapper *pWrapper, char *user, char *spi, char *encrypt, char *secret, char *ckey) {
SMnodeMgmt *pMgmt = pWrapper->pMgmt;
int32_t code = mndRetriveAuth(pMgmt->pMnode, user, spi, encrypt, secret, ckey);
dTrace("user:%s, retrieve auth spi:%d encrypt:%d", user, *spi, *encrypt);
return code;
}
int32_t mmMonitorMnodeInfo(SMgmtWrapper *pWrapper, SMonClusterInfo *pClusterInfo, SMonVgroupInfo *pVgroupInfo, int32_t mmMonitorMnodeInfo(SMgmtWrapper *pWrapper, SMonClusterInfo *pClusterInfo, SMonVgroupInfo *pVgroupInfo,
SMonGrantInfo *pGrantInfo) { SMonGrantInfo *pGrantInfo) {
SMnodeMgmt *pMgmt = pWrapper->pMgmt; SMnodeMgmt *pMgmt = pWrapper->pMgmt;
......
source/dnode/mgmt/mm/src/mmWorker.c → source/dnode/mgmt/mm/mmWorker.c
浏览文件 @ ef5ca96d
...@@ -37,7 +37,7 @@ static void mmProcessQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) { ...@@ -37,7 +37,7 @@ static void mmProcessQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
dError("msg:%p, failed to process since %s", pMsg, terrstr()); dError("msg:%p, failed to process since %s", pMsg, terrstr());
} }
SRpcMsg rsp = {.handle = pRpc->handle, .code = code, .contLen = pMsg->rspLen, .pCont = pMsg->pRsp}; SRpcMsg rsp = {.handle = pRpc->handle, .code = code, .contLen = pMsg->rspLen, .pCont = pMsg->pRsp};
dndSendRsp(pMgmt->pWrapper, &rsp); tmsgSendRsp(&rsp);
} }
} }
...@@ -60,7 +60,7 @@ static void mmProcessQueryQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) { ...@@ -60,7 +60,7 @@ static void mmProcessQueryQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
if (pRpc->handle != NULL && code != 0) { if (pRpc->handle != NULL && code != 0) {
dError("msg:%p, failed to process since %s", pMsg, terrstr()); dError("msg:%p, failed to process since %s", pMsg, terrstr());
SRpcMsg rsp = {.handle = pRpc->handle, .code = code, .ahandle = pRpc->ahandle}; SRpcMsg rsp = {.handle = pRpc->handle, .code = code, .ahandle = pRpc->ahandle};
dndSendRsp(pMgmt->pWrapper, &rsp); tmsgSendRsp(&rsp);
} }
} }
......
source/dnode/mgmt/qm/inc/qm.h 已删除 100644 → 0
浏览文件 @ aaf35118
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_DND_QNODE_H_
#define _TD_DND_QNODE_H_
#include "dnd.h"
#ifdef __cplusplus
extern "C" {
#endif
void qmGetMgmtFp(SMgmtWrapper *pMgmt);
#ifdef __cplusplus
}
#endif
#endif /*_TD_DND_QNODE_H_*/
\ No newline at end of file
source/dnode/mgmt/qm/src/qmMsg.c → source/dnode/mgmt/qm/qmHandle.c
浏览文件 @ ef5ca96d
...@@ -56,14 +56,14 @@ int32_t qmProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { ...@@ -56,14 +56,14 @@ int32_t qmProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
void qmInitMsgHandles(SMgmtWrapper *pWrapper) { void qmInitMsgHandles(SMgmtWrapper *pWrapper) {
// Requests handled by VNODE // Requests handled by VNODE
dndSetMsgHandle(pWrapper, TDMT_VND_QUERY, qmProcessQueryMsg, QND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_QUERY, qmProcessQueryMsg, QNODE_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_QUERY_CONTINUE, qmProcessQueryMsg, QND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_QUERY_CONTINUE, qmProcessQueryMsg, QNODE_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_FETCH, qmProcessFetchMsg, QND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_FETCH, qmProcessFetchMsg, QNODE_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_FETCH_RSP, qmProcessFetchMsg, QND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_FETCH_RSP, qmProcessFetchMsg, QNODE_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_RES_READY, qmProcessFetchMsg, QND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_RES_READY, qmProcessFetchMsg, QNODE_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_TASKS_STATUS, qmProcessFetchMsg, QND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_TASKS_STATUS, qmProcessFetchMsg, QNODE_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_CANCEL_TASK, qmProcessFetchMsg, QND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_CANCEL_TASK, qmProcessFetchMsg, QNODE_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_DROP_TASK, qmProcessFetchMsg, QND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_DROP_TASK, qmProcessFetchMsg, QNODE_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_SHOW_TABLES, qmProcessFetchMsg, QND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_SHOW_TABLES, qmProcessFetchMsg, QNODE_HANDLE);
} }
source/dnode/mgmt/qm/src/qmInt.c → source/dnode/mgmt/qm/qmInt.c
浏览文件 @ ef5ca96d
...@@ -112,7 +112,7 @@ int32_t qmOpen(SMgmtWrapper *pWrapper) { ...@@ -112,7 +112,7 @@ int32_t qmOpen(SMgmtWrapper *pWrapper) {
return code; return code;
} }
void qmGetMgmtFp(SMgmtWrapper *pWrapper) { void qmSetMgmtFp(SMgmtWrapper *pWrapper) {
SMgmtFp mgmtFp = {0}; SMgmtFp mgmtFp = {0};
mgmtFp.openFp = qmOpen; mgmtFp.openFp = qmOpen;
mgmtFp.closeFp = qmClose; mgmtFp.closeFp = qmClose;
......
source/dnode/mgmt/qm/src/qmWorker.c → source/dnode/mgmt/qm/qmWorker.c
浏览文件 @ ef5ca96d
...@@ -18,7 +18,7 @@ ...@@ -18,7 +18,7 @@
static void qmSendRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) { static void qmSendRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) {
SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle, .code = code}; SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle, .code = code};
dndSendRsp(pWrapper, &rsp); tmsgSendRsp(&rsp);
} }
static void qmProcessQueryQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) { static void qmProcessQueryQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
......
source/dnode/mgmt/sm/inc/sm.h 已删除 100644 → 0
浏览文件 @ aaf35118
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_DND_SNODE_H_
#define _TD_DND_SNODE_H_
#include "dnd.h"
#ifdef __cplusplus
extern "C" {
#endif
void smGetMgmtFp(SMgmtWrapper *pWrapper);
#ifdef __cplusplus
}
#endif
#endif /*_TD_DND_SNODE_H_*/
source/dnode/mgmt/sm/src/smMsg.c → source/dnode/mgmt/sm/smHandle.c
浏览文件 @ ef5ca96d
...@@ -56,6 +56,6 @@ int32_t smProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { ...@@ -56,6 +56,6 @@ int32_t smProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
void smInitMsgHandles(SMgmtWrapper *pWrapper) { void smInitMsgHandles(SMgmtWrapper *pWrapper) {
// Requests handled by SNODE // Requests handled by SNODE
dndSetMsgHandle(pWrapper, TDMT_SND_TASK_DEPLOY, smProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_SND_TASK_DEPLOY, smProcessMgmtMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_SND_TASK_EXEC, smProcessExecMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_SND_TASK_EXEC, smProcessExecMsg, DEFAULT_HANDLE);
} }
source/dnode/mgmt/sm/src/smInt.c → source/dnode/mgmt/sm/smInt.c
浏览文件 @ ef5ca96d
...@@ -109,7 +109,7 @@ int32_t smOpen(SMgmtWrapper *pWrapper) { ...@@ -109,7 +109,7 @@ int32_t smOpen(SMgmtWrapper *pWrapper) {
return code; return code;
} }
void smGetMgmtFp(SMgmtWrapper *pWrapper) { void smSetMgmtFp(SMgmtWrapper *pWrapper) {
SMgmtFp mgmtFp = {0}; SMgmtFp mgmtFp = {0};
mgmtFp.openFp = smOpen; mgmtFp.openFp = smOpen;
mgmtFp.closeFp = smClose; mgmtFp.closeFp = smClose;
......
source/dnode/mgmt/vm/inc/vm.h 已删除 100644 → 0
浏览文件 @ aaf35118
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_DND_VNODES_H_
#define _TD_DND_VNODES_H_
#include "dnd.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct {
int32_t openVnodes;
int32_t totalVnodes;
int32_t masterNum;
int64_t numOfSelectReqs;
int64_t numOfInsertReqs;
int64_t numOfInsertSuccessReqs;
int64_t numOfBatchInsertReqs;
int64_t numOfBatchInsertSuccessReqs;
} SVnodesStat;
void vmGetMgmtFp(SMgmtWrapper *pWrapper);
void vmMonitorVnodeLoads(SMgmtWrapper *pWrapper, SArray *pLoads);
int32_t vmMonitorTfsInfo(SMgmtWrapper *pWrapper, SMonDiskInfo *pInfo);
void vmMonitorVnodeReqs(SMgmtWrapper *pWrapper, SMonDnodeInfo *pInfo);
#ifdef __cplusplus
}
#endif
#endif /*_TD_DND_VNODES_H_*/
\ No newline at end of file
source/dnode/mgmt/vm/src/vmMsg.c → source/dnode/mgmt/vm/vmHandle.c
浏览文件 @ ef5ca96d
...@@ -235,49 +235,49 @@ int32_t vmProcessCompactVnodeReq(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) { ...@@ -235,49 +235,49 @@ int32_t vmProcessCompactVnodeReq(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
void vmInitMsgHandles(SMgmtWrapper *pWrapper) { void vmInitMsgHandles(SMgmtWrapper *pWrapper) {
// Requests handled by VNODE // Requests handled by VNODE
dndSetMsgHandle(pWrapper, TDMT_VND_SUBMIT, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_SUBMIT, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_QUERY, (NodeMsgFp)vmProcessQueryMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_QUERY, (NodeMsgFp)vmProcessQueryMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_QUERY_CONTINUE, (NodeMsgFp)vmProcessQueryMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_QUERY_CONTINUE, (NodeMsgFp)vmProcessQueryMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_FETCH, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_FETCH, (NodeMsgFp)vmProcessFetchMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_FETCH_RSP, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_FETCH_RSP, (NodeMsgFp)vmProcessFetchMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_ALTER_TABLE, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_ALTER_TABLE, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_UPDATE_TAG_VAL, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_UPDATE_TAG_VAL, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_TABLE_META, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_TABLE_META, (NodeMsgFp)vmProcessFetchMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_TABLES_META, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_TABLES_META, (NodeMsgFp)vmProcessFetchMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_MQ_CONSUME, (NodeMsgFp)vmProcessQueryMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_MQ_CONSUME, (NodeMsgFp)vmProcessQueryMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_MQ_QUERY, (NodeMsgFp)vmProcessQueryMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_MQ_QUERY, (NodeMsgFp)vmProcessQueryMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_MQ_CONNECT, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_MQ_CONNECT, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_MQ_DISCONNECT, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_MQ_DISCONNECT, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_MQ_SET_CUR, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_MQ_SET_CUR, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_RES_READY, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_RES_READY, (NodeMsgFp)vmProcessFetchMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_TASKS_STATUS, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_TASKS_STATUS, (NodeMsgFp)vmProcessFetchMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_CANCEL_TASK, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_CANCEL_TASK, (NodeMsgFp)vmProcessFetchMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_DROP_TASK, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_DROP_TASK, (NodeMsgFp)vmProcessFetchMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_CREATE_STB, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_CREATE_STB, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_ALTER_STB, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_ALTER_STB, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_DROP_STB, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_DROP_STB, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_CREATE_TABLE, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_CREATE_TABLE, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_ALTER_TABLE, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_ALTER_TABLE, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_DROP_TABLE, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_DROP_TABLE, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_CREATE_SMA, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_CREATE_SMA, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_CANCEL_SMA, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_CANCEL_SMA, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_DROP_SMA, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_DROP_SMA, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_SHOW_TABLES, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_SHOW_TABLES, (NodeMsgFp)vmProcessFetchMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_SHOW_TABLES_FETCH, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_SHOW_TABLES_FETCH, (NodeMsgFp)vmProcessFetchMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_MQ_SET_CONN, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_MQ_SET_CONN, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_MQ_REB, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_MQ_REB, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_MQ_SET_CUR, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_MQ_SET_CUR, (NodeMsgFp)vmProcessFetchMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_CONSUME, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_CONSUME, (NodeMsgFp)vmProcessFetchMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_TASK_DEPLOY, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_TASK_DEPLOY, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_QUERY_HEARTBEAT, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_QUERY_HEARTBEAT, (NodeMsgFp)vmProcessFetchMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_TASK_PIPE_EXEC, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_TASK_PIPE_EXEC, (NodeMsgFp)vmProcessFetchMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_TASK_MERGE_EXEC, (NodeMsgFp)vmProcessMergeMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_TASK_MERGE_EXEC, (NodeMsgFp)vmProcessMergeMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_TASK_WRITE_EXEC, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_TASK_WRITE_EXEC, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_STREAM_TRIGGER, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_STREAM_TRIGGER, (NodeMsgFp)vmProcessFetchMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_VNODE, vmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_VNODE, vmProcessMgmtMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_ALTER_VNODE, vmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_ALTER_VNODE, vmProcessMgmtMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_DROP_VNODE, vmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_DROP_VNODE, vmProcessMgmtMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_SYNC_VNODE, vmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_SYNC_VNODE, vmProcessMgmtMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_COMPACT_VNODE, vmProcessMgmtMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_DND_COMPACT_VNODE, vmProcessMgmtMsg, DEFAULT_HANDLE);
} }
source/dnode/mgmt/vm/src/vmInt.c → source/dnode/mgmt/vm/vmInt.c
浏览文件 @ ef5ca96d
...@@ -333,7 +333,7 @@ static int32_t vmRequire(SMgmtWrapper *pWrapper, bool *required) { ...@@ -333,7 +333,7 @@ static int32_t vmRequire(SMgmtWrapper *pWrapper, bool *required) {
return 0; return 0;
} }
void vmGetMgmtFp(SMgmtWrapper *pWrapper) { void vmSetMgmtFp(SMgmtWrapper *pWrapper) {
SMgmtFp mgmtFp = {0}; SMgmtFp mgmtFp = {0};
mgmtFp.openFp = vmInit; mgmtFp.openFp = vmInit;
mgmtFp.closeFp = vmCleanup; mgmtFp.closeFp = vmCleanup;
......
source/dnode/mgmt/vm/src/vmWorker.c → source/dnode/mgmt/vm/vmWorker.c
浏览文件 @ ef5ca96d
...@@ -18,7 +18,7 @@ ...@@ -18,7 +18,7 @@
static void vmSendRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) { static void vmSendRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) {
SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle, .code = code}; SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle, .code = code};
dndSendRsp(pWrapper, &rsp); tmsgSendRsp(&rsp);
} }
static void vmProcessMgmtQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) { static void vmProcessMgmtQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
...@@ -116,7 +116,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO ...@@ -116,7 +116,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
int32_t code = vnodeApplyWMsg(pVnode->pImpl, pRpc, &pRsp); int32_t code = vnodeApplyWMsg(pVnode->pImpl, pRpc, &pRsp);
if (pRsp != NULL) { if (pRsp != NULL) {
pRsp->ahandle = pRpc->ahandle; pRsp->ahandle = pRpc->ahandle;
dndSendRsp(pVnode->pWrapper, pRsp); tmsgSendRsp(pRsp);
taosMemoryFree(pRsp); taosMemoryFree(pRsp);
} else { } else {
if (code != 0 && terrno != 0) code = terrno; if (code != 0 && terrno != 0) code = terrno;
......
source/dnode/mnode/impl/src/mndQuery.c
浏览文件 @ ef5ca96d
...@@ -52,7 +52,7 @@ int32_t mndProcessFetchMsg(SNodeMsg *pReq) { ...@@ -52,7 +52,7 @@ int32_t mndProcessFetchMsg(SNodeMsg *pReq) {
} }
int32_t mndInitQuery(SMnode *pMnode) { int32_t mndInitQuery(SMnode *pMnode) {
if (qWorkerInit(NODE_TYPE_MNODE, MND_VGID, NULL, (void **)&pMnode->pQuery, &pMnode->msgCb) != 0) { if (qWorkerInit(NODE_TYPE_MNODE, MNODE_HANDLE, NULL, (void **)&pMnode->pQuery, &pMnode->msgCb) != 0) {
mError("failed to init qworker in mnode since %s", terrstr()); mError("failed to init qworker in mnode since %s", terrstr());
return -1; return -1;
} }
......
source/libs/planner/src/planPhysiCreater.c
浏览文件 @ ef5ca96d
...@@ -419,7 +419,7 @@ static int32_t createSystemTableScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan* ...@@ -419,7 +419,7 @@ static int32_t createSystemTableScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan*
vgroupInfoToNodeAddr(pScanLogicNode->pVgroupList->vgroups, &pSubplan->execNode); vgroupInfoToNodeAddr(pScanLogicNode->pVgroupList->vgroups, &pSubplan->execNode);
taosArrayPush(pCxt->pExecNodeList, &pSubplan->execNode); taosArrayPush(pCxt->pExecNodeList, &pSubplan->execNode);
} else { } else {
SQueryNodeAddr addr = { .nodeId = MND_VGID, .epSet = pCxt->pPlanCxt->mgmtEpSet }; SQueryNodeAddr addr = { .nodeId = MNODE_HANDLE, .epSet = pCxt->pPlanCxt->mgmtEpSet };
taosArrayPush(pCxt->pExecNodeList, &addr); taosArrayPush(pCxt->pExecNodeList, &addr);
} }
pScan->mgmtEpSet = pCxt->pPlanCxt->mgmtEpSet; pScan->mgmtEpSet = pCxt->pPlanCxt->mgmtEpSet;
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或