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

Merge pull request #6516 from taosdata/fix/TD-4672 

[TD-4672]<fix>: remove peer fqdn IP caching
This commit is contained in:
Shengliang Guan 2021-06-21 11:40:17 +08:00 committed by GitHub
parent 1bc6382f96 c50a85e9da
commit 1af6c53986
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 34 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/sync/inc/syncInt.h
View File

@ -132,6 +132,7 @@ void * syncRestoreData(void *param);
int32_t syncSaveIntoBuffer(SSyncPeer *pPeer, SWalHead *pHead);
void syncRestartConnection(SSyncPeer *pPeer);
void syncBroadcastStatus(SSyncNode *pNode);
uint32_t syncResolvePeerFqdn(SSyncPeer *pPeer);
SSyncPeer *syncAcquirePeer(int64_t rid);
void syncReleasePeer(SSyncPeer *pPeer);

34
src/sync/src/syncMain.c
View File

@ -559,7 +559,8 @@ static void syncClosePeerConn(SSyncPeer *pPeer) {
static void syncRemovePeer(SSyncPeer *pPeer) {
sInfo("%s, it is removed", pPeer->id);
pPeer->ip = 0;
//pPeer->ip = 0;
pPeer->fqdn[0] = '\0';
syncClosePeerConn(pPeer);
//taosRemoveRef(tsPeerRefId, pPeer->rid);
syncReleasePeer(pPeer);
@ -585,20 +586,31 @@ static void syncStopCheckPeerConn(SSyncPeer *pPeer) {
sDebug("%s, stop check peer connection", pPeer->id);
}
uint32_t syncResolvePeerFqdn(SSyncPeer *pPeer) {
uint32_t ip = taosGetIpv4FromFqdn(pPeer->fqdn);
if (ip == 0xFFFFFFFF) {
sError("failed to resolve peer fqdn:%s since %s", pPeer->fqdn, strerror(errno));
terrno = TSDB_CODE_RPC_FQDN_ERROR;
return 0;
}
return ip;
}
static SSyncPeer *syncAddPeer(SSyncNode *pNode, const SNodeInfo *pInfo) {
uint32_t ip = taosGetIpv4FromFqdn(pInfo->nodeFqdn);
/*uint32_t ip = taosGetIpv4FromFqdn(pInfo->nodeFqdn);
if (ip == 0xFFFFFFFF) {
sError("failed to add peer, can resolve fqdn:%s since %s", pInfo->nodeFqdn, strerror(errno));
terrno = TSDB_CODE_RPC_FQDN_ERROR;
return NULL;
}
*/
SSyncPeer *pPeer = calloc(1, sizeof(SSyncPeer));
if (pPeer == NULL) return NULL;
pPeer->nodeId = pInfo->nodeId;
tstrncpy(pPeer->fqdn, pInfo->nodeFqdn, sizeof(pPeer->fqdn));
pPeer->ip = ip;
//pPeer->ip = ip;
pPeer->port = pInfo->nodePort;
pPeer->fqdn[sizeof(pPeer->fqdn) - 1] = 0;
snprintf(pPeer->id, sizeof(pPeer->id), "vgId:%d, nodeId:%d", pNode->vgId, pPeer->nodeId);
@ -864,7 +876,7 @@ static void syncRestartPeer(SSyncPeer *pPeer) {
}
void syncRestartConnection(SSyncPeer *pPeer) {
if (pPeer->ip == 0) return;
if (pPeer->fqdn[0] == '\0') return;
if (syncAcquirePeer(pPeer->rid) == NULL) return;
@ -878,7 +890,7 @@ static void syncProcessSyncRequest(char *msg, SSyncPeer *pPeer) {
SSyncNode *pNode = pPeer->pSyncNode;
sInfo("%s, sync-req is received", pPeer->id);
if (pPeer->ip == 0) return;
if (pPeer->fqdn[0] == '\0') return;
if (nodeRole != TAOS_SYNC_ROLE_MASTER) {
sError("%s, I am not master anymore", pPeer->id);
@ -1090,7 +1102,7 @@ static int32_t syncProcessPeerMsg(int64_t rid, void *buffer) {
}
static int32_t syncSendPeersStatusMsgToPeer(SSyncPeer *pPeer, char ack, int8_t type, uint16_t tranId) {
if (pPeer->peerFd < 0 || pPeer->ip == 0) {
if (pPeer->peerFd < 0 || pPeer->fqdn[0] == '\0') {
sDebug("%s, failed to send status msg, restart fd:%d", pPeer->id, pPeer->peerFd);
syncRestartConnection(pPeer);
return -1;
@ -1135,7 +1147,13 @@ static void syncSetupPeerConnection(SSyncPeer *pPeer) {
return;
}
SOCKET connFd = taosOpenTcpClientSocket(pPeer->ip, pPeer->port, 0);
uint32_t ip = syncResolvePeerFqdn(pPeer);
if (!ip) {
taosTmrReset(syncCheckPeerConnection, SYNC_CHECK_INTERVAL, (void *)pPeer->rid, tsSyncTmrCtrl, &pPeer->timer);
return;
}
SOCKET connFd = taosOpenTcpClientSocket(ip, pPeer->port, 0);
if (connFd <= 0) {
sDebug("%s, failed to open tcp socket since %s", pPeer->id, strerror(errno));
taosTmrReset(syncCheckPeerConnection, SYNC_CHECK_INTERVAL, (void *)pPeer->rid, tsSyncTmrCtrl, &pPeer->timer);

8
src/sync/src/syncRetrieve.c
View File

@ -422,6 +422,12 @@ void *syncRetrieveData(void *param) {
return NULL;
}
uint32_t ip = syncResolvePeerFqdn(pPeer);
if (!ip) {
syncReleasePeer(pPeer);
return NULL;
}
SSyncNode *pNode = pPeer->pSyncNode;
taosBlockSIGPIPE();
@ -430,7 +436,7 @@ void *syncRetrieveData(void *param) {
if (pNode->notifyFlowCtrlFp) (*pNode->notifyFlowCtrlFp)(pNode->vgId, pPeer->numOfRetrieves);
pPeer->syncFd = taosOpenTcpClientSocket(pPeer->ip, pPeer->port, 0);
pPeer->syncFd = taosOpenTcpClientSocket(ip, pPeer->port, 0);
if (pPeer->syncFd < 0) {
sError("%s, failed to open socket to sync", pPeer->id);
} else {