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

Merge pull request #1848 from taosdata/hotfix/unified 

port in redirect message shall be htons
This commit is contained in:
slguan 2020-05-08 21:02:40 +08:00 committed by GitHub
parent 4a8c1e7ea0 599711dd6b
commit b50d04efe4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 17 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/dnode/src/dnodeVRead.c
View File

@ -81,6 +81,8 @@ void dnodeCleanupRead() {
} }
taosCloseQset(readQset); taosCloseQset(readQset);
free(readPool.readWorker);
dPrint("dnode read is closed"); dPrint("dnode read is closed");
} }

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

@ -226,17 +226,6 @@ void *rpcOpen(const SRpcInit *pInit) {
pRpc->cfp = pInit->cfp; pRpc->cfp = pInit->cfp;
pRpc->afp = pInit->afp; pRpc->afp = pInit->afp;
pRpc->tcphandle = (*taosInitConn[pRpc->connType|RPC_CONN_TCP])(0, pRpc->localPort, pRpc->label,
pRpc->numOfThreads, rpcProcessMsgFromPeer, pRpc);
pRpc->udphandle = (*taosInitConn[pRpc->connType])(0, pRpc->localPort, pRpc->label,
pRpc->numOfThreads, rpcProcessMsgFromPeer, pRpc);
if (pRpc->tcphandle == NULL || pRpc->udphandle == NULL) {
tError("%s failed to init network, port:%d", pRpc->label, pRpc->localPort);
rpcClose(pRpc);
return NULL;
}
size_t size = sizeof(SRpcConn) * pRpc->sessions; size_t size = sizeof(SRpcConn) * pRpc->sessions;
pRpc->connList = (SRpcConn *)calloc(1, size); pRpc->connList = (SRpcConn *)calloc(1, size);
if (pRpc->connList == NULL) { if (pRpc->connList == NULL) {
@ -277,6 +266,17 @@ void *rpcOpen(const SRpcInit *pInit) {
pthread_mutex_init(&pRpc->mutex, NULL); pthread_mutex_init(&pRpc->mutex, NULL);
pRpc->tcphandle = (*taosInitConn[pRpc->connType|RPC_CONN_TCP])(0, pRpc->localPort, pRpc->label,
pRpc->numOfThreads, rpcProcessMsgFromPeer, pRpc);
pRpc->udphandle = (*taosInitConn[pRpc->connType])(0, pRpc->localPort, pRpc->label,
pRpc->numOfThreads, rpcProcessMsgFromPeer, pRpc);
if (pRpc->tcphandle == NULL || pRpc->udphandle == NULL) {
tError("%s failed to init network, port:%d", pRpc->label, pRpc->localPort);
rpcClose(pRpc);
return NULL;
}
tTrace("%s RPC is openned, numOfThreads:%d", pRpc->label, pRpc->numOfThreads); tTrace("%s RPC is openned, numOfThreads:%d", pRpc->label, pRpc->numOfThreads);
return pRpc; return pRpc;
@ -869,9 +869,9 @@ static void *rpcProcessMsgFromPeer(SRecvInfo *pRecv) {
pConn = rpcProcessMsgHead(pRpc, pRecv); pConn = rpcProcessMsgHead(pRpc, pRecv);
if (pHead->msgType < TSDB_MSG_TYPE_CM_HEARTBEAT || (rpcDebugFlag & 16)) { if (pHead->msgType < TSDB_MSG_TYPE_CM_HEARTBEAT || (rpcDebugFlag & 16)) {
tTrace("%s %p, %s received from 0x%x:%hu, parse code:0x%x len:%d sig:0x%08x:0x%08x:%d", tTrace("%s %p, %s received from 0x%x:%hu, parse code:0x%x len:%d sig:0x%08x:0x%08x:%d code:0x%x",
pRpc->label, pConn, taosMsg[pHead->msgType], pRecv->ip, pRecv->port, terrno, pRpc->label, pConn, taosMsg[pHead->msgType], pRecv->ip, pRecv->port, terrno,
pRecv->msgLen, pHead->sourceId, pHead->destId, pHead->tranId, pHead->port); pRecv->msgLen, pHead->sourceId, pHead->destId, pHead->tranId, pHead->code);
} }
int32_t code = terrno; int32_t code = terrno;
@ -939,6 +939,8 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead) {
pContext->numOfTry = 0; pContext->numOfTry = 0;
memcpy(&pContext->ipSet, pHead->content, sizeof(pContext->ipSet)); memcpy(&pContext->ipSet, pHead->content, sizeof(pContext->ipSet));
tTrace("%s %p, redirect is received, numOfIps:%d", pRpc->label, pConn, pContext->ipSet.numOfIps); tTrace("%s %p, redirect is received, numOfIps:%d", pRpc->label, pConn, pContext->ipSet.numOfIps);
for (int i=0; i<pContext->ipSet.numOfIps; ++i)
pContext->ipSet.port[i] = htons(pContext->ipSet.port[i]);
rpcSendReqToServer(pRpc, pContext); rpcSendReqToServer(pRpc, pContext);
} else if (pHead->code == TSDB_CODE_NOT_READY) { } else if (pHead->code == TSDB_CODE_NOT_READY) {
pContext->code = pHead->code; pContext->code = pHead->code;