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

remove race condition in retry timer

This commit is contained in:
Jeff Tao 2020-06-16 11:03:23 +00:00
parent be1791e425
commit 646b5e53a7
1 changed files with 22 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
src/rpc/src/rpcMain.c
View File

@ -555,18 +555,10 @@ static SRpcConn *rpcOpenConn(SRpcInfo *pRpc, char *peerFqdn, uint16_t peerPort,
return pConn; return pConn;
} }
static void rpcCloseConn(void *thandle) { static void rpcReleaseConn(SRpcConn *pConn) {
SRpcConn *pConn = (SRpcConn *)thandle;
SRpcInfo *pRpc = pConn->pRpc; SRpcInfo *pRpc = pConn->pRpc;
if (pConn->user[0] == 0) return; if (pConn->user[0] == 0) return;
rpcLockConn(pConn);
if (pConn->user[0] == 0) {
rpcUnlockConn(pConn);
return;
}
pConn->user[0] = 0; pConn->user[0] = 0;
if (taosCloseConn[pConn->connType]) (*taosCloseConn[pConn->connType])(pConn->chandle); if (taosCloseConn[pConn->connType]) (*taosCloseConn[pConn->connType])(pConn->chandle);
@ -591,7 +583,21 @@ static void rpcCloseConn(void *thandle) {
taosFreeId(pRpc->idPool, pConn->sid); taosFreeId(pRpc->idPool, pConn->sid);
pConn->pContext = NULL; pConn->pContext = NULL;
tTrace("%s, rpc connection is closed", pConn->info); tTrace("%s, rpc connection is released", pConn->info);
}
static void rpcCloseConn(void *thandle) {
SRpcConn *pConn = (SRpcConn *)thandle;
if (pConn->user[0] == 0) return;
rpcLockConn(pConn);
if (pConn->user[0] == 0) {
rpcUnlockConn(pConn);
return;
}
rpcReleaseConn(pConn);
rpcUnlockConn(pConn); rpcUnlockConn(pConn);
} }
@ -911,8 +917,8 @@ static void rpcProcessBrokenLink(SRpcConn *pConn) {
if (pConn->inType) rpcReportBrokenLinkToServer(pConn); if (pConn->inType) rpcReportBrokenLinkToServer(pConn);
rpcReleaseConn(pConn);
rpcUnlockConn(pConn); rpcUnlockConn(pConn);
rpcCloseConn(pConn);
} }
static void *rpcProcessMsgFromPeer(SRecvInfo *pRecv) { static void *rpcProcessMsgFromPeer(SRecvInfo *pRecv) {
@ -1217,7 +1223,6 @@ static void rpcProcessConnError(void *param, void *id) {
static void rpcProcessRetryTimer(void *param, void *tmrId) { static void rpcProcessRetryTimer(void *param, void *tmrId) {
SRpcConn *pConn = (SRpcConn *)param; SRpcConn *pConn = (SRpcConn *)param;
SRpcInfo *pRpc = pConn->pRpc; SRpcInfo *pRpc = pConn->pRpc;
int reportDisc = 0;
rpcLockConn(pConn); rpcLockConn(pConn);
@ -1233,19 +1238,17 @@ static void rpcProcessRetryTimer(void *param, void *tmrId) {
} else { } else {
// close the connection // close the connection
tTrace("%s, failed to send msg:%s to %s:%hu", pConn->info, taosMsg[pConn->outType], pConn->peerFqdn, pConn->peerPort); tTrace("%s, failed to send msg:%s to %s:%hu", pConn->info, taosMsg[pConn->outType], pConn->peerFqdn, pConn->peerPort);
reportDisc = 1; if (pConn->pContext) {
pConn->pContext->code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
rpcProcessConnError(pConn->pContext, NULL);
rpcReleaseConn(pConn);
}
} }
} else { } else {
tTrace("%s, retry timer not processed", pConn->info); tTrace("%s, retry timer not processed", pConn->info);
} }
rpcUnlockConn(pConn); rpcUnlockConn(pConn);
if (reportDisc && pConn->pContext) {
pConn->pContext->code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
rpcProcessConnError(pConn->pContext, NULL);
rpcCloseConn(pConn);
}
} }
static void rpcProcessIdleTimer(void *param, void *tmrId) { static void rpcProcessIdleTimer(void *param, void *tmrId) {