xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refact transport

This commit is contained in:
Shengliang Guan 2022-04-02 18:58:28 +08:00
parent 2eb30ee0f0
commit 745fa09bbb
19 changed files with 45 additions and 273 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
source/dnode/mgmt/bm/inc/bm.h
View File

@ -1,31 +0,0 @@
/*
* 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_*/

2
source/dnode/mgmt/bm/inc/bmInt.h
View File

@ -16,7 +16,7 @@
#ifndef _TD_DND_BNODE_INT_H_
#define _TD_DND_BNODE_INT_H_
#include "bm.h"
#include "dnd.h"
#include "bnode.h"
#ifdef __cplusplus

38
source/dnode/mgmt/dm/inc/dm.h
View File

@ -1,38 +0,0 @@
/*
* 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_*/

3
source/dnode/mgmt/dm/inc/dmInt.h
View File

@ -16,7 +16,7 @@
#ifndef _TD_DND_DNODE_INT_H_
#define _TD_DND_DNODE_INT_H_
#include "dm.h"
#include "dnd.h"
#ifdef __cplusplus
extern "C" {
@ -45,6 +45,7 @@ int32_t dmWriteFile(SDnodeMgmt *pMgmt);
void dmUpdateDnodeEps(SDnodeMgmt *pMgmt, SArray *pDnodeEps);
// dmMsg.c
void dmInitMsgHandles(SMgmtWrapper *pWrapper);
void dmSendStatusReq(SDnodeMgmt *pMgmt);
int32_t dmProcessConfigReq(SDnodeMgmt *pMgmt, SNodeMsg *pMsg);
int32_t dmProcessStatusRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg);

1
source/dnode/mgmt/dm/src/dmMsg.c
View File

@ -15,7 +15,6 @@
#define _DEFAULT_SOURCE
#include "dmInt.h"
#include "vm.h"
void dmSendStatusReq(SDnodeMgmt *pMgmt) {
SDnode *pDnode = pMgmt->pDnode;

5
source/dnode/mgmt/dm/src/dmWorker.c
View File

@ -14,12 +14,7 @@
*/
#define _DEFAULT_SOURCE
#include "bm.h"
#include "dmInt.h"
#include "mm.h"
#include "qm.h"
#include "sm.h"
#include "vm.h"
static void *dmThreadRoutine(void *param) {
SDnodeMgmt *pMgmt = param;

30
source/dnode/mgmt/main/inc/dnd.h
View File

@ -165,7 +165,37 @@ int32_t dndInitTrans(SDnode *pDnode);
void dndCleanupTrans(SDnode *pDnode);
SMsgCb dndCreateMsgcb(SMgmtWrapper *pWrapper);
SProcCfg dndGenProcCfg(SMgmtWrapper *pWrapper);
int32_t dndInitMsgHandle(SDnode *pDnode);
// mgmt
void dmGetMgmtFp(SMgmtWrapper *pWrapper);
void bmGetMgmtFp(SMgmtWrapper *pWrapper);
void qmGetMgmtFp(SMgmtWrapper *pMgmt);
void smGetMgmtFp(SMgmtWrapper *pWrapper);
void vmGetMgmtFp(SMgmtWrapper *pWrapper);
void mmGetMgmtFp(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
}
#endif

10
source/dnode/mgmt/main/inc/dndInt.h
View File

@ -18,13 +18,6 @@
#include "dnd.h"
#include "bm.h"
#include "dm.h"
#include "mm.h"
#include "qm.h"
#include "sm.h"
#include "vm.h"
#ifdef __cplusplus
extern "C" {
#endif
@ -47,9 +40,6 @@ void dndHandleEvent(SDnode *pDnode, EDndEvent event);
void dndProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, SEpSet *pEpSet);
void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pMsg);
// dndTransport.c
int32_t dndInitMsgHandle(SDnode *pDnode);
// dndFile.c
TdFilePtr dndCheckRunning(const char *dataDir);
int32_t dndReadShmFile(SDnode *pDnode);

2
source/dnode/mgmt/main/src/dndStr.c
View File

@ -25,7 +25,7 @@ const char *dndStatStr(EDndStatus status) {
case DND_STAT_STOPPED:
return "stopped";
default:
return "unknown";
return "UNKNOWN";
}
}

37
source/dnode/mgmt/main/src/dndTransport.c
View File

@ -142,20 +142,14 @@ static void dndProcessRequest(SDnode *pDnode, SRpcMsg *pReq, SEpSet *pEpSet) {
}
}
static void dndSendMsgToMnodeRecv(SDnode *pDnode, SRpcMsg *pRpcMsg, SRpcMsg *pRpcRsp) {
STransMgmt *pMgmt = &pDnode->trans;
SEpSet epSet = {0};
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, DNODE);
if (pWrapper != NULL) {
dmGetMnodeEpSet(pWrapper->pMgmt, &epSet);
dndReleaseWrapper(pWrapper);
}
rpcSendRecv(pMgmt->clientRpc, &epSet, pRpcMsg, pRpcRsp);
static inline void dndSendMsgToMnodeRecv(SDnode *pDnode, SRpcMsg *pReq, SRpcMsg *pRsp) {
SEpSet epSet = {0};
SMgmtWrapper *pWrapper = &pDnode->wrappers[DNODE];
dmGetMnodeEpSet(pWrapper->pMgmt, &epSet);
rpcSendRecv(pDnode->trans.clientRpc, &epSet, pReq, pRsp);
}
static int32_t dndGetHideUserAuth(SDnode *pDnode, char *user, char *spi, char *encrypt, char *secret, char *ckey) {
static inline int32_t dndGetHideUserAuth(SDnode *pDnode, char *user, char *spi, char *encrypt, char *secret, char *ckey) {
int32_t code = 0;
char pass[TSDB_PASSWORD_LEN + 1] = {0};
@ -183,21 +177,6 @@ static int32_t dndRetrieveUserAuthInfo(SDnode *pDnode, char *user, char *spi, ch
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};
tstrncpy(authReq.user, user, TSDB_USER_LEN);
int32_t contLen = tSerializeSAuthReq(NULL, 0, &authReq);
@ -285,11 +264,9 @@ int32_t dndInitMsgHandle(SDnode *pDnode) {
for (int32_t msgIndex = 0; msgIndex < TDMT_MAX; ++msgIndex) {
NodeMsgFp msgFp = pWrapper->msgFps[msgIndex];
int32_t vgId = pWrapper->msgVgIds[msgIndex];
int8_t vgId = pWrapper->msgVgIds[msgIndex];
if (msgFp == NULL) continue;
// dTrace("msg:%s will be processed by %s, vgId:%d", tMsgInfo[msgIndex], pWrapper->name, vgId);
SMsgHandle *pHandle = &pMgmt->msgHandles[msgIndex];
if (vgId == QND_VGID) {
if (pHandle->pQndWrapper != NULL) {

35
source/dnode/mgmt/mm/inc/mm.h
View File

@ -1,35 +0,0 @@
/*
* 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_*/

2
source/dnode/mgmt/mm/inc/mmInt.h
View File

@ -16,7 +16,7 @@
#ifndef _TD_DND_MNODE_INT_H_
#define _TD_DND_MNODE_INT_H_
#include "mm.h"
#include "dnd.h"
#include "mnode.h"
#ifdef __cplusplus

8
source/dnode/mgmt/mm/src/mmInt.c
View File

@ -241,14 +241,6 @@ void mmGetMgmtFp(SMgmtWrapper *pWrapper) {
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,
SMonGrantInfo *pGrantInfo) {
SMnodeMgmt *pMgmt = pWrapper->pMgmt;

31
source/dnode/mgmt/qm/inc/qm.h
View File

@ -1,31 +0,0 @@
/*
* 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_*/

2
source/dnode/mgmt/qm/inc/qmInt.h
View File

@ -16,7 +16,7 @@
#ifndef _TD_DND_QNODE_INT_H_
#define _TD_DND_QNODE_INT_H_
#include "qm.h"
#include "dnd.h"
#include "qnode.h"
#ifdef __cplusplus

31
source/dnode/mgmt/sm/inc/sm.h
View File

@ -1,31 +0,0 @@
/*
* 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_*/

2
source/dnode/mgmt/sm/inc/smInt.h
View File

@ -16,7 +16,7 @@
#ifndef _TD_DND_SNODE_INT_H_
#define _TD_DND_SNODE_INT_H_
#include "sm.h"
#include "dnd.h"
#include "snode.h"
#ifdef __cplusplus

46
source/dnode/mgmt/vm/inc/vm.h
View File

@ -1,46 +0,0 @@
/*
* 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_*/

2
source/dnode/mgmt/vm/inc/vmInt.h
View File

@ -17,7 +17,7 @@
#define _TD_DND_VNODES_INT_H_
#include "sync.h"
#include "vm.h"
#include "dnd.h"
#include "vnode.h"
#ifdef __cplusplus