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

change lock type in dndDnode

This commit is contained in:
Shengliang Guan 2021-11-27 16:20:47 +08:00
parent c049678ade
commit c3a25985a7
3 changed files with 46 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

81
source/dnode/mgmt/impl/src/dndDnode.c
View File

@ -18,32 +18,27 @@
#include "dndTransport.h"
#include "dndVnodes.h"
static inline void dndRLockDnode(SDnode *pDnode) { taosRLockLatch(&pDnode->dmgmt.latch); }
static inline void dndRUnLockDnode(SDnode *pDnode) { taosRUnLockLatch(&pDnode->dmgmt.latch); }
static inline void dndWLockDnode(SDnode *pDnode) { taosWLockLatch(&pDnode->dmgmt.latch); }
static inline void dndWUnLockDnode(SDnode *pDnode) { taosWUnLockLatch(&pDnode->dmgmt.latch); }
int32_t dndGetDnodeId(SDnode *pDnode) {
dndRLockDnode(pDnode);
int32_t dnodeId = pDnode->dmgmt.dnodeId;
dndRUnLockDnode(pDnode);
SDnodeMgmt *pMgmt = &pDnode->dmgmt;
taosRLockLatch(&pMgmt->latch);
int32_t dnodeId = pMgmt->dnodeId;
taosRUnLockLatch(&pMgmt->latch);
return dnodeId;
}
int64_t dndGetClusterId(SDnode *pDnode) {
dndRLockDnode(pDnode);
int64_t clusterId = pDnode->dmgmt.clusterId;
dndRUnLockDnode(pDnode);
SDnodeMgmt *pMgmt = &pDnode->dmgmt;
taosRLockLatch(&pMgmt->latch);
int64_t clusterId = pMgmt->clusterId;
taosRUnLockLatch(&pMgmt->latch);
return clusterId;
}
void dndGetDnodeEp(SDnode *pDnode, int32_t dnodeId, char *pEp, char *pFqdn, uint16_t *pPort) {
dndRLockDnode(pDnode);
SDnodeMgmt *pMgmt = &pDnode->dmgmt;
taosRLockLatch(&pMgmt->latch);
SDnodeEp *pDnodeEp = taosHashGet(pDnode->dmgmt.dnodeHash, &dnodeId, sizeof(int32_t));
SDnodeEp *pDnodeEp = taosHashGet(pMgmt->dnodeHash, &dnodeId, sizeof(int32_t));
if (pDnodeEp != NULL) {
if (pPort != NULL) {
*pPort = pDnodeEp->port;
@ -56,13 +51,14 @@ void dndGetDnodeEp(SDnode *pDnode, int32_t dnodeId, char *pEp, char *pFqdn, uint
}
}
dndRUnLockDnode(pDnode);
taosRUnLockLatch(&pMgmt->latch);
}
void dndGetMnodeEpSet(SDnode *pDnode, SEpSet *pEpSet) {
dndRLockDnode(pDnode);
*pEpSet = pDnode->dmgmt.mnodeEpSet;
dndRUnLockDnode(pDnode);
SDnodeMgmt *pMgmt = &pDnode->dmgmt;
taosRLockLatch(&pMgmt->latch);
*pEpSet = pMgmt->mnodeEpSet;
taosRUnLockLatch(&pMgmt->latch);
}
void dndSendRedirectMsg(SDnode *pDnode, SRpcMsg *pMsg) {
@ -87,14 +83,15 @@ void dndSendRedirectMsg(SDnode *pDnode, SRpcMsg *pMsg) {
static void dndUpdateMnodeEpSet(SDnode *pDnode, SEpSet *pEpSet) {
dInfo("mnode is changed, num:%d inUse:%d", pEpSet->numOfEps, pEpSet->inUse);
dndWLockDnode(pDnode);
SDnodeMgmt *pMgmt = &pDnode->dmgmt;
taosWLockLatch(&pMgmt->latch);
pDnode->dmgmt.mnodeEpSet = *pEpSet;
pMgmt->mnodeEpSet = *pEpSet;
for (int32_t i = 0; i < pEpSet->numOfEps; ++i) {
dInfo("mnode index:%d %s:%u", i, pEpSet->fqdn[i], pEpSet->port[i]);
}
dndWUnLockDnode(pDnode);
taosWUnLockLatch(&pMgmt->latch);
}
static void dndPrintDnodes(SDnode *pDnode) {
@ -145,16 +142,18 @@ static void dndResetDnodes(SDnode *pDnode, SDnodeEps *pDnodeEps) {
static bool dndIsEpChanged(SDnode *pDnode, int32_t dnodeId, char *pEp) {
bool changed = false;
dndRLockDnode(pDnode);
SDnodeEp *pDnodeEp = taosHashGet(pDnode->dmgmt.dnodeHash, &dnodeId, sizeof(int32_t));
SDnodeMgmt *pMgmt = &pDnode->dmgmt;
taosRLockLatch(&pMgmt->latch);
SDnodeEp *pDnodeEp = taosHashGet(pMgmt->dnodeHash, &dnodeId, sizeof(int32_t));
if (pDnodeEp != NULL) {
char epstr[TSDB_EP_LEN + 1];
snprintf(epstr, TSDB_EP_LEN, "%s:%u", pDnodeEp->fqdn, pDnodeEp->port);
changed = strcmp(pEp, epstr) != 0;
}
dndRUnLockDnode(pDnode);
taosRUnLockLatch(&pMgmt->latch);
return changed;
}
@ -342,11 +341,14 @@ static void dndSendStatusMsg(SDnode *pDnode) {
return;
}
dndRLockDnode(pDnode);
bool changed = false;
SDnodeMgmt *pMgmt = &pDnode->dmgmt;
taosRLockLatch(&pMgmt->latch);
pStatus->sversion = htonl(pDnode->opt.sver);
pStatus->dnodeId = htonl(pDnode->dmgmt.dnodeId);
pStatus->clusterId = htobe64(pDnode->dmgmt.clusterId);
pStatus->rebootTime = htonl(pDnode->dmgmt.rebootTime);
pStatus->dnodeId = htonl(pMgmt->dnodeId);
pStatus->clusterId = htobe64(pMgmt->clusterId);
pStatus->rebootTime = htonl(pMgmt->rebootTime);
pStatus->numOfCores = htonl(pDnode->opt.numOfCores);
tstrncpy(pStatus->dnodeEp, pDnode->opt.localEp, TSDB_EP_LEN);
pStatus->clusterCfg.statusInterval = htonl(pDnode->opt.statusInterval);
@ -356,7 +358,7 @@ static void dndSendStatusMsg(SDnode *pDnode) {
pStatus->clusterCfg.checkTime = 0;
char timestr[32] = "1970-01-01 00:00:00.00";
(void)taosParseTime(timestr, &pStatus->clusterCfg.checkTime, (int32_t)strlen(timestr), TSDB_TIME_PRECISION_MILLI, 0);
dndRUnLockDnode(pDnode);
taosRUnLockLatch(&pMgmt->latch);
dndGetVnodeLoads(pDnode, &pStatus->vnodeLoads);
contLen = sizeof(SStatusMsg) + pStatus->vnodeLoads.num * sizeof(SVnodeLoad);
@ -370,32 +372,33 @@ static void dndUpdateDnodeCfg(SDnode *pDnode, SDnodeCfg *pCfg) {
if (pMgmt->dnodeId == 0 || pMgmt->dropped != pCfg->dropped) {
dInfo("set dnodeId:%d clusterId:%" PRId64 " dropped:%d", pCfg->dnodeId, pCfg->clusterId, pCfg->dropped);
dndWLockDnode(pDnode);
taosWLockLatch(&pMgmt->latch);
pMgmt->dnodeId = pCfg->dnodeId;
pMgmt->clusterId = pCfg->clusterId;
pMgmt->dropped = pCfg->dropped;
(void)dndWriteDnodes(pDnode);
dndWUnLockDnode(pDnode);
taosWUnLockLatch(&pMgmt->latch);
}
}
static void dndUpdateDnodeEps(SDnode *pDnode, SDnodeEps *pDnodeEps) {
if (pDnodeEps == NULL || pDnodeEps->num <= 0) return;
dndWLockDnode(pDnode);
SDnodeMgmt *pMgmt = &pDnode->dmgmt;
taosWLockLatch(&pMgmt->latch);
if (pDnodeEps->num != pDnode->dmgmt.dnodeEps->num) {
if (pDnodeEps->num != pMgmt->dnodeEps->num) {
dndResetDnodes(pDnode, pDnodeEps);
dndWriteDnodes(pDnode);
} else {
int32_t size = pDnodeEps->num * sizeof(SDnodeEp) + sizeof(SDnodeEps);
if (memcmp(pDnode->dmgmt.dnodeEps, pDnodeEps, size) != 0) {
if (memcmp(pMgmt->dnodeEps, pDnodeEps, size) != 0) {
dndResetDnodes(pDnode, pDnodeEps);
dndWriteDnodes(pDnode);
}
}
dndWUnLockDnode(pDnode);
taosWUnLockLatch(&pMgmt->latch);
}
static void dndProcessStatusRsp(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
@ -512,7 +515,7 @@ void dndCleanupDnode(SDnode *pDnode) {
pMgmt->threadId = NULL;
}
dndWLockDnode(pDnode);
taosWLockLatch(&pMgmt->latch);
if (pMgmt->dnodeEps != NULL) {
free(pMgmt->dnodeEps);
@ -529,7 +532,7 @@ void dndCleanupDnode(SDnode *pDnode) {
pMgmt->file = NULL;
}
dndWUnLockDnode(pDnode);
taosWUnLockLatch(&pMgmt->latch);
dInfo("dnode-dnode is cleaned up");
}

2
source/dnode/mgmt/impl/src/dndTransport.c
View File

@ -187,6 +187,7 @@ static void dndProcessRequest(void *param, SRpcMsg *pMsg, SEpSet *pEpSet) {
int32_t msgType = pMsg->msgType;
if (msgType == TSDB_MSG_TYPE_NETWORK_TEST) {
dTrace("RPC %p, network test req will be processed", pMsg->handle);
dndProcessDnodeReq(pDnode, pMsg, pEpSet);
return;
}
@ -206,6 +207,7 @@ static void dndProcessRequest(void *param, SRpcMsg *pMsg, SEpSet *pEpSet) {
}
if (pMsg->pCont == NULL) {
dTrace("RPC %p, req:%s not processed since content is null", pMsg->handle, taosMsg[msgType]);
SRpcMsg rspMsg = {.handle = pMsg->handle, .code = TSDB_CODE_DND_INVALID_MSG_LEN};
rpcSendResponse(&rspMsg);
return;

5
source/dnode/mgmt/impl/src/dnode.c
View File

@ -26,7 +26,7 @@
EStat dndGetStat(SDnode *pDnode) { return pDnode->stat; }
void dndSetStat(SDnode *pDnode, EStat stat) {
dDebug("dnode stat set from %s to %s", dndStatStr(pDnode->stat), dndStatStr(stat));
dDebug("dnode status set from %s to %s", dndStatStr(pDnode->stat), dndStatStr(stat));
pDnode->stat = stat;
}
@ -214,8 +214,7 @@ void dndCleanup(SDnode *pDnode) {
dndCleanupDnode(pDnode);
walCleanUp();
rpcCleanup();
dInfo("TDengine is cleaned up successfully");
dndCleanupEnv(pDnode);
free(pDnode);
dInfo("TDengine is cleaned up successfully");
}