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

Merge pull request #1877 from taosdata/feature/rpcIpSet 

support RpcIpSet change
This commit is contained in:
Shengliang Guan 2020-05-11 22:18:06 +08:00 committed by GitHub
parent bd49f05819 feb4bbcde0
commit ce90bbecf9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 29 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/client/inc/tsclient.h
View File

@ -365,7 +365,7 @@ void tscInitMsgsFp();
int tsParseSql(SSqlObj *pSql, bool multiVnodeInsertion);
void tscProcessMsgFromServer(SRpcMsg *rpcMsg);
void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet);
int tscProcessSql(SSqlObj *pSql);
int tscRenewMeterMeta(SSqlObj *pSql, char *tableId);

8
src/client/src/tscServer.c
View File

@ -221,7 +221,7 @@ int tscSendMsgToServer(SSqlObj *pSql) {
return TSDB_CODE_SUCCESS;
}
void tscProcessMsgFromServer(SRpcMsg *rpcMsg) {
void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet) {
SSqlObj *pSql = (SSqlObj *)rpcMsg->handle;
if (pSql == NULL) {
tscError("%p sql is already released", pSql->signature);
@ -245,6 +245,12 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg) {
return;
}
if (pCmd->command < TSDB_SQL_MGMT) {
if (pIpSet) pSql->ipList = *pIpSet;
} else {
if (pIpSet) tscMgmtIpSet = *pIpSet;
}
if (rpcMsg->pCont == NULL) {
rpcMsg->code = TSDB_CODE_NETWORK_UNAVAIL;
} else {

4
src/dnode/src/dnodeMgmt.c
View File

@ -266,8 +266,8 @@ static int32_t dnodeProcessConfigDnodeMsg(SRpcMsg *pMsg) {
return taosCfgDynamicOptions(pCfg->config);
}
void dnodeUpdateIpSet(void *ahandle, SRpcIpSet *pIpSet) {
dPrint("mnode IP list is changed for ufp is called, numOfIps:%d inUse:%d", pIpSet->numOfIps, pIpSet->inUse);
void dnodeUpdateIpSet(SRpcIpSet *pIpSet) {
dPrint("mnode IP list is changed, numOfIps:%d inUse:%d", pIpSet->numOfIps, pIpSet->inUse);
for (int i = 0; i < pIpSet->numOfIps; ++i) {
dPrint("mnode index:%d %s:%u", i, pIpSet->fqdn[i], pIpSet->port[i])
}

12
src/dnode/src/dnodePeer.c
View File

@ -29,11 +29,11 @@
#include "dnodeVWrite.h"
#include "mnode.h"
extern void dnodeUpdateIpSet(void *ahandle, SRpcIpSet *pIpSet);
extern void dnodeUpdateIpSet(SRpcIpSet *pIpSet);
static void (*dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
static void dnodeProcessReqMsgFromDnode(SRpcMsg *pMsg);
static void dnodeProcessReqMsgFromDnode(SRpcMsg *pMsg, SRpcIpSet *);
static void (*dnodeProcessRspMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *rpcMsg);
static void dnodeProcessRspFromDnode(SRpcMsg *pMsg);
static void dnodeProcessRspFromDnode(SRpcMsg *pMsg, SRpcIpSet *pIpSet);
static void *tsDnodeServerRpc = NULL;
static void *tsDnodeClientRpc = NULL;
@ -81,7 +81,7 @@ void dnodeCleanupServer() {
}
}
static void dnodeProcessReqMsgFromDnode(SRpcMsg *pMsg) {
static void dnodeProcessReqMsgFromDnode(SRpcMsg *pMsg, SRpcIpSet *pIpSet) {
SRpcMsg rspMsg;
rspMsg.handle = pMsg->handle;
rspMsg.pCont = NULL;
@ -119,7 +119,6 @@ int32_t dnodeInitClient() {
rpcInit.label = "DND-C";
rpcInit.numOfThreads = 1;
rpcInit.cfp = dnodeProcessRspFromDnode;
rpcInit.ufp = dnodeUpdateIpSet;
rpcInit.sessions = 100;
rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.idleTime = tsShellActivityTimer * 1000;
@ -145,9 +144,10 @@ void dnodeCleanupClient() {
}
}
static void dnodeProcessRspFromDnode(SRpcMsg *pMsg) {
static void dnodeProcessRspFromDnode(SRpcMsg *pMsg, SRpcIpSet *pIpSet) {
if (dnodeProcessRspMsgFp[pMsg->msgType]) {
if (pMsg->msgType == TSDB_MSG_TYPE_DM_STATUS_RSP && pIpSet) dnodeUpdateIpSet(pIpSet);
(*dnodeProcessRspMsgFp[pMsg->msgType])(pMsg);
} else {
dError("RPC %p, msg:%s is not processed", pMsg->handle, taosMsg[pMsg->msgType]);

4
src/dnode/src/dnodeShell.c
View File

@ -28,7 +28,7 @@
#include "dnodeShell.h"
static void (*dnodeProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
static void dnodeProcessMsgFromShell(SRpcMsg *pMsg);
static void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcIpSet *);
static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char *secret, char *ckey);
static void * tsDnodeShellRpc = NULL;
static int32_t tsDnodeQueryReqNum = 0;
@ -106,7 +106,7 @@ void dnodeCleanupShell() {
}
}
void dnodeProcessMsgFromShell(SRpcMsg *pMsg) {
void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcIpSet *pIpSet) {
SRpcMsg rpcMsg;
rpcMsg.handle = pMsg->handle;
rpcMsg.pCont = NULL;

5
src/inc/trpc.h
View File

@ -66,10 +66,7 @@ typedef struct {
char *ckey; // ciphering key
// call back to process incoming msg, code shall be ignored by server app
void (*cfp)(SRpcMsg *);
// call back to process notify the ipSet changes, for client app only
void (*ufp)(void *ahandle, SRpcIpSet *pIpSet);
void (*cfp)(SRpcMsg *, SRpcIpSet *);
// call back to retrieve the client auth info, for server app only
int (*afp)(char *tableId, char *spi, char *encrypt, char *secret, char *ckey);

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

@ -55,9 +55,8 @@ typedef struct {
char secret[TSDB_KEY_LEN]; // secret for the link
char ckey[TSDB_KEY_LEN]; // ciphering key
void (*cfp)(SRpcMsg *);
void (*cfp)(SRpcMsg *, SRpcIpSet *);
int (*afp)(char *user, char *spi, char *encrypt, char *secret, char *ckey);
void (*ufp)(void *ahandle, SRpcIpSet *pIpSet);
void *idPool; // handle to ID pool
void *tmrCtrl; // handle to timer
@ -222,7 +221,6 @@ void *rpcOpen(const SRpcInit *pInit) {
if (pInit->secret) strcpy(pRpc->secret, pInit->secret);
if (pInit->ckey) strcpy(pRpc->ckey, pInit->ckey);
pRpc->spi = pInit->spi;
pRpc->ufp = pInit->ufp;
pRpc->cfp = pInit->cfp;
pRpc->afp = pInit->afp;
@ -900,10 +898,11 @@ static void rpcNotifyClient(SRpcReqContext *pContext, SRpcMsg *pMsg) {
memcpy(pContext->pRsp, pMsg, sizeof(SRpcMsg));
} else {
// for asynchronous API
if (pRpc->ufp && (pContext->ipSet.inUse != pContext->oldInUse || pContext->redirect))
(*pRpc->ufp)(pContext->ahandle, &pContext->ipSet); // notify the update of ipSet
SRpcIpSet *pIpSet = NULL;
if (pContext->ipSet.inUse != pContext->oldInUse || pContext->redirect)
pIpSet = &pContext->ipSet;
(*pRpc->cfp)(pMsg);
(*pRpc->cfp)(pMsg, pIpSet);
}
// free the request message
@ -924,7 +923,7 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead) {
if ( rpcIsReq(pHead->msgType) ) {
rpcMsg.handle = pConn;
taosTmrReset(rpcProcessProgressTimer, tsRpcTimer/2, pConn, pRpc->tmrCtrl, &pConn->pTimer);
(*(pRpc->cfp))(&rpcMsg);
(*(pRpc->cfp))(&rpcMsg, NULL);
} else {
// it's a response
SRpcReqContext *pContext = pConn->pContext;

13
src/rpc/test/rclient.c
View File

@ -31,22 +31,16 @@ typedef struct {
void *pRpc;
} SInfo;
static void processResponse(SRpcMsg *pMsg) {
static void processResponse(SRpcMsg *pMsg, SRpcIpSet *pIpSet) {
SInfo *pInfo = (SInfo *)pMsg->handle;
tTrace("thread:%d, response is received, type:%d contLen:%d code:0x%x", pInfo->index, pMsg->msgType, pMsg->contLen, pMsg->code);
if (pIpSet) pInfo->ipSet = *pIpSet;
rpcFreeCont(pMsg->pCont);
sem_post(&pInfo->rspSem);
}
static void processUpdateIpSet(void *handle, SRpcIpSet *pIpSet) {
SInfo *pInfo = (SInfo *)handle;
tTrace("thread:%d, ip set is changed, index:%d", pInfo->index, pIpSet->inUse);
pInfo->ipSet = *pIpSet;
}
static int tcount = 0;
static void *sendRequest(void *param) {
@ -99,7 +93,6 @@ int main(int argc, char *argv[]) {
rpcInit.label = "APP";
rpcInit.numOfThreads = 1;
rpcInit.cfp = processResponse;
rpcInit.ufp = processUpdateIpSet;
rpcInit.sessions = 100;
rpcInit.idleTime = tsShellActivityTimer*1000;
rpcInit.user = "michael";

8
src/rpc/test/rsclient.c
View File

@ -32,12 +32,6 @@ typedef struct {
void *pRpc;
} SInfo;
static void processUpdateIpSet(void *handle, SRpcIpSet *pIpSet) {
SInfo *pInfo = (SInfo *)handle;
tTrace("thread:%d, ip set is changed, index:%d", pInfo->index, pIpSet->inUse);
pInfo->ipSet = *pIpSet;
}
static int tcount = 0;
static int terror = 0;
@ -100,8 +94,6 @@ int main(int argc, char *argv[]) {
rpcInit.localPort = 0;
rpcInit.label = "APP";
rpcInit.numOfThreads = 1;
// rpcInit.cfp = processResponse;
rpcInit.ufp = processUpdateIpSet;
rpcInit.sessions = 100;
rpcInit.idleTime = tsShellActivityTimer*1000;
rpcInit.user = "michael";

2
src/rpc/test/rserver.c
View File

@ -113,7 +113,7 @@ int retrieveAuthInfo(char *meterId, char *spi, char *encrypt, char *secret, char
return ret;
}
void processRequestMsg(SRpcMsg *pMsg) {
void processRequestMsg(SRpcMsg *pMsg, SRpcIpSet *pIpSet) {
SRpcMsg *pTemp;
pTemp = taosAllocateQitem(sizeof(SRpcMsg));