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

refactor(tools) add network speed test codes

This commit is contained in:
Shengliang Guan 2022-04-24 11:19:06 +08:00
parent eaaee9c56c
commit 70905e2128
6 changed files with 97 additions and 483 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/common/tmsgdef.h
View File

@ -85,6 +85,7 @@ enum {
TD_DEF_MSG_TYPE(TDMT_DND_DROP_VNODE, "dnode-drop-vnode", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_DND_DROP_VNODE, "dnode-drop-vnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_CONFIG_DNODE, "dnode-config-dnode", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_DND_CONFIG_DNODE, "dnode-config-dnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_SERVER_STATUS, "dnode-server-status", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_DND_SERVER_STATUS, "dnode-server-status", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_DND_NET_TEST, "dnode-net-test", NULL, NULL)
// Requests handled by MNODE // Requests handled by MNODE
TD_NEW_MSG_SEG(TDMT_MND_MSG) TD_NEW_MSG_SEG(TDMT_MND_MSG)

6
source/dnode/mgmt/implement/src/dmTransport.c
View File

@ -131,6 +131,12 @@ static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
return; return;
} }
if (msgType == TDMT_DND_NET_TEST) {
dTrace("net test req will be processed, handle:%p, app:%p", pMsg->handle, pMsg->ahandle);
dmProcessServerStatusReq(pDnode, pMsg);
return;
}
if (pDnode->status != DND_STAT_RUNNING) { if (pDnode->status != DND_STAT_RUNNING) {
dError("msg:%s ignored since dnode not running, handle:%p app:%p", TMSG_INFO(msgType), pMsg->handle, pMsg->ahandle); dError("msg:%s ignored since dnode not running, handle:%p app:%p", TMSG_INFO(msgType), pMsg->handle, pMsg->ahandle);
if (isReq) { if (isReq) {

1
source/dnode/mgmt/interface/inc/dmInt.h
View File

@ -37,6 +37,7 @@ void dmSetMsgHandle(SMgmtWrapper *pWrapper, tmsg_t msgType, NodeMsgFp nodeMsgF
void dmReportStartup(SDnode *pDnode, const char *pName, const char *pDesc); void dmReportStartup(SDnode *pDnode, const char *pName, const char *pDesc);
void dmReportStartupByWrapper(SMgmtWrapper *pWrapper, const char *pName, const char *pDesc); void dmReportStartupByWrapper(SMgmtWrapper *pWrapper, const char *pName, const char *pDesc);
void dmProcessServerStatusReq(SDnode *pDnode, SRpcMsg *pMsg); void dmProcessServerStatusReq(SDnode *pDnode, SRpcMsg *pMsg);
void dmProcessNettestReq(SDnode *pDnode, SRpcMsg *pMsg);
void dmGetMonitorSysInfo(SMonSysInfo *pInfo); void dmGetMonitorSysInfo(SMonSysInfo *pInfo);
// dmFile.c // dmFile.c

8
source/dnode/mgmt/interface/src/dmInt.c
View File

@ -171,6 +171,14 @@ static void dmGetServerStatus(SDnode *pDnode, SServerStatusRsp *pStatus) {
} }
} }
void dmProcessNettestReq(SDnode *pDnode, SRpcMsg *pRpc) {
dDebug("net test req is received");
SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = 0};
rsp.pCont = rpcMallocCont(shell.args.pktLen);
rsp.contLen = shell.args.pktLen;
rpcSendResponse(&rsp);
}
void dmProcessServerStatusReq(SDnode *pDnode, SRpcMsg *pReq) { void dmProcessServerStatusReq(SDnode *pDnode, SRpcMsg *pReq) {
dDebug("server status req is received"); dDebug("server status req is received");

13
tools/shell/inc/shellInt.h
View File

@ -22,6 +22,7 @@
#include "taoserror.h" #include "taoserror.h"
#include "tconfig.h" #include "tconfig.h"
#include "tglobal.h" #include "tglobal.h"
#include "trpc.h"
#include "ttypes.h" #include "ttypes.h"
#include "tutil.h" #include "tutil.h"
@ -30,12 +31,12 @@
#define SHELL_HISTORY_FILE ".taos_history" #define SHELL_HISTORY_FILE ".taos_history"
#define SHELL_DEFAULT_RES_SHOW_NUM 100 #define SHELL_DEFAULT_RES_SHOW_NUM 100
#define SHELL_DEFAULT_MAX_BINARY_DISPLAY_WIDTH 30 #define SHELL_DEFAULT_MAX_BINARY_DISPLAY_WIDTH 30
#define SHELL_MAX_PKG_LEN 2 * 1024 * 1024 #define SHELL_MAX_PKG_LEN 2 * 1024 * 1024
#define SHELL_MIN_PKG_LEN 1 #define SHELL_MIN_PKG_LEN 1
#define SHELL_DEF_PKG_LEN 1024 #define SHELL_DEF_PKG_LEN 1024
#define SHELL_MAX_PKG_NUM 1* 1024 * 1024 #define SHELL_MAX_PKG_NUM 1 * 1024 * 1024
#define SHELL_MIN_PKG_NUM 1 #define SHELL_MIN_PKG_NUM 1
#define SHELL_DEF_PKG_NUM 100 #define SHELL_DEF_PKG_NUM 100
typedef struct { typedef struct {
char* hist[SHELL_MAX_HISTORY_SIZE]; char* hist[SHELL_MAX_HISTORY_SIZE];

551
tools/shell/src/shellNettest.c
View File

@ -16,507 +16,104 @@
#define _GNU_SOURCE #define _GNU_SOURCE
#include "shellInt.h" #include "shellInt.h"
void shellTestNetWork() {} static void shellWorkAsClient() {
SRpcInit rpcInit = {0};
SEpSet epSet = {.inUse = 0, .numOfEps = 1};
SRpcMsg rpcRsp = {0};
void *clientRpc = NULL;
char pass[TSDB_PASSWORD_LEN + 1] = {0};
#if 0 taosEncryptPass_c((uint8_t *)("_pwd"), strlen("_pwd"), pass);
#define ALLOW_FORBID_FUNC rpcInit.label = "CHK";
#include "os.h" rpcInit.numOfThreads = 1;
#include "rpcHead.h" rpcInit.sessions = 16;
#include "syncMsg.h" rpcInit.connType = TAOS_CONN_CLIENT;
#include "taosdef.h" rpcInit.idleTime = tsShellActivityTimer * 1000;
#include "taoserror.h" rpcInit.user = "_dnd";
#include "tchecksum.h" rpcInit.ckey = "_key";
#include "tglobal.h" rpcInit.spi = 1;
#include "tlog.h" rpcInit.secret = pass;
#include "tmsg.h"
#include "trpc.h"
#include "osSocket.h" clientRpc = rpcOpen(&rpcInit);
if (clientRpc == NULL) {
printf("failed to init net test client since %s\n", terrstr());
goto _OVER;
}
printf("net test client is initialized\n");
#define MAX_PKG_LEN (64 * 1000) tstrncpy(epSet.eps[0].fqdn, shell.args.host, TSDB_FQDN_LEN);
#define MAX_SPEED_PKG_LEN (1024 * 1024 * 1024) epSet.eps[0].port = (uint16_t)shell.args.port;
#define MIN_SPEED_PKG_LEN 1024
#define MAX_SPEED_PKG_NUM 10000
#define MIN_SPEED_PKG_NUM 1
#define BUFFER_SIZE (MAX_PKG_LEN + 1024)
extern int tsRpcMaxUdpSize; int32_t totalSucc = 0;
uint64_t startTime = taosGetTimestampUs();
typedef struct { for (int32_t i = 0; i < shell.args.pktNum; ++i) {
char * hostFqdn; SRpcMsg rpcMsg = {.ahandle = (void *)0x9525, .msgType = TDMT_DND_NET_TEST};
uint32_t hostIp; rpcMsg.pCont = rpcMallocCont(shell.args.pktLen);
int32_t port; rpcMsg.contLen = shell.args.pktLen;
int32_t pktLen;
} STestInfo;
static void *taosNetBindUdpPort(void *sarg) { printf("net test request is sent, size:%d\n", rpcMsg.contLen);
STestInfo *pinfo = (STestInfo *)sarg; rpcSendRecv(clientRpc, &epSet, &rpcMsg, &rpcRsp);
int32_t port = pinfo->port; printf("net test response is received, size:%d\n", rpcMsg.contLen);
SOCKET serverSocket;
char buffer[BUFFER_SIZE];
int32_t iDataNum;
socklen_t sin_size;
int32_t bufSize = 1024000;
struct sockaddr_in server_addr; if (rpcRsp.code == 0) totalSucc++;
struct sockaddr_in clientAddr;
setThreadName("netBindUdpPort"); rpcFreeCont(rpcRsp.pCont);
rpcRsp.pCont = NULL;
if ((serverSocket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) {
uError("failed to create UDP socket since %s", strerror(errno));
return NULL;
} }
bzero(&server_addr, sizeof(server_addr)); uint64_t endTime = taosGetTimestampUs();
server_addr.sin_family = AF_INET; uint64_t elT = endTime - startTime;
server_addr.sin_port = htons(port);
server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
if (bind(serverSocket, (struct sockaddr *)&server_addr, sizeof(server_addr)) < 0) { printf("\ntotal succ:%5d/%d\tcost:%8.2lf ms\tspeed:%8.2lf MB/s\n", totalSucc, shell.args.pktNum, elT / 1000.0,
uError("failed to bind UDP port:%d since %s", port, strerror(errno)); shell.args.pktLen / (elT / 1000000.0) / 1024.0 / 1024.0 * totalSucc);
return NULL;
_OVER:
if (clientRpc != NULL) {
rpcClose(clientRpc);
} }
if (rpcRsp.pCont != NULL) {
TdSocketPtr pSocket = (TdSocketPtr)taosMemoryMalloc(sizeof(TdSocket)); rpcFreeCont(rpcRsp.pCont);
if (pSocket == NULL) {
taosCloseSocketNoCheck1(serverSocket);
return NULL;
}
pSocket->fd = serverSocket;
pSocket->refId = 0;
if (taosSetSockOpt(pSocket, SOL_SOCKET, SO_SNDBUF, (void *)&bufSize, sizeof(bufSize)) != 0) {
uError("failed to set the send buffer size for UDP socket\n");
taosCloseSocket(&pSocket);
return NULL;
}
if (taosSetSockOpt(pSocket, SOL_SOCKET, SO_RCVBUF, (void *)&bufSize, sizeof(bufSize)) != 0) {
uError("failed to set the receive buffer size for UDP socket\n");
taosCloseSocket(&pSocket);
return NULL;
}
uInfo("UDP server at port:%d is listening", port);
while (1) {
memset(buffer, 0, BUFFER_SIZE);
sin_size = sizeof(*(struct sockaddr *)&server_addr);
iDataNum = recvfrom(serverSocket, buffer, BUFFER_SIZE, 0, (struct sockaddr *)&clientAddr, &sin_size);
if (iDataNum < 0) {
uDebug("failed to perform recvfrom func at %d since %s", port, strerror(errno));
continue;
}
uInfo("UDP: recv:%d bytes from %s at %d", iDataNum, taosInetNtoa(clientAddr.sin_addr), port);
if (iDataNum > 0) {
iDataNum = taosSendto(pSocket, buffer, iDataNum, 0, (struct sockaddr *)&clientAddr, (int32_t)sin_size);
}
uInfo("UDP: send:%d bytes to %s at %d", iDataNum, taosInetNtoa(clientAddr.sin_addr), port);
}
taosCloseSocket(&pSocket);
return NULL;
}
static void *taosNetBindTcpPort(void *sarg) {
struct sockaddr_in server_addr;
struct sockaddr_in clientAddr;
STestInfo *pinfo = sarg;
int32_t port = pinfo->port;
SOCKET serverSocket;
int32_t addr_len = sizeof(clientAddr);
SOCKET client;
char buffer[BUFFER_SIZE];
setThreadName("netBindTcpPort");
if ((serverSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) {
uError("failed to create TCP socket since %s", strerror(errno));
return NULL;
}
bzero(&server_addr, sizeof(server_addr));
server_addr.sin_family = AF_INET;
server_addr.sin_port = htons(port);
server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
int32_t reuse = 1;
TdSocketPtr pSocket = (TdSocketPtr)taosMemoryMalloc(sizeof(TdSocket));
if (pSocket == NULL) {
taosCloseSocketNoCheck1(serverSocket);
return NULL;
}
pSocket->fd = serverSocket;
pSocket->refId = 0;
if (taosSetSockOpt(pSocket, SOL_SOCKET, SO_REUSEADDR, (void *)&reuse, sizeof(reuse)) < 0) {
uError("setsockopt SO_REUSEADDR failed: %d (%s)", errno, strerror(errno));
taosCloseSocket(&pSocket);
return NULL;
}
if (bind(serverSocket, (struct sockaddr *)&server_addr, sizeof(server_addr)) < 0) {
uError("failed to bind TCP port:%d since %s", port, strerror(errno));
taosCloseSocket(&pSocket);
return NULL;
}
if (taosKeepTcpAlive(pSocket) < 0) {
uError("failed to set tcp server keep-alive option since %s", strerror(errno));
taosCloseSocket(&pSocket);
return NULL;
}
if (listen(serverSocket, 10) < 0) {
uError("failed to listen TCP port:%d since %s", port, strerror(errno));
taosCloseSocket(&pSocket);
return NULL;
}
uInfo("TCP server at port:%d is listening", port);
while (1) {
client = accept(serverSocket, (struct sockaddr *)&clientAddr, (socklen_t *)&addr_len);
if (client < 0) {
uDebug("TCP: failed to accept at port:%d since %s", port, strerror(errno));
continue;
}
int32_t ret = taosReadMsg(pSocket, buffer, pinfo->pktLen);
if (ret < 0 || ret != pinfo->pktLen) {
uError("TCP: failed to read %d bytes at port:%d since %s", pinfo->pktLen, port, strerror(errno));
taosCloseSocket(&pSocket);
return NULL;
}
uInfo("TCP: read:%d bytes from %s at %d", pinfo->pktLen, taosInetNtoa(clientAddr.sin_addr), port);
ret = taosWriteMsg(pSocket, buffer, pinfo->pktLen);
if (ret < 0) {
uError("TCP: failed to write %d bytes at %d since %s", pinfo->pktLen, port, strerror(errno));
taosCloseSocket(&pSocket);
return NULL;
}
uInfo("TCP: write:%d bytes to %s at %d", pinfo->pktLen, taosInetNtoa(clientAddr.sin_addr), port);
}
taosCloseSocket(&pSocket);
return NULL;
}
static int32_t taosNetCheckTcpPort(STestInfo *info) {
SOCKET clientSocket;
char buffer[BUFFER_SIZE] = {0};
if ((clientSocket = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
uError("failed to create TCP client socket since %s", strerror(errno));
return -1;
}
int32_t reuse = 1;
TdSocketPtr pSocket = (TdSocketPtr)taosMemoryMalloc(sizeof(TdSocket));
if (pSocket == NULL) {
taosCloseSocketNoCheck1(clientSocket);
return -1;
}
pSocket->fd = clientSocket;
pSocket->refId = 0;
if (taosSetSockOpt(pSocket, SOL_SOCKET, SO_REUSEADDR, (void *)&reuse, sizeof(reuse)) < 0) {
uError("setsockopt SO_REUSEADDR failed: %d (%s)", errno, strerror(errno));
taosCloseSocket(&pSocket);
return -1;
}
struct sockaddr_in serverAddr;
memset((char *)&serverAddr, 0, sizeof(serverAddr));
serverAddr.sin_family = AF_INET;
serverAddr.sin_port = (uint16_t)htons((uint16_t)info->port);
serverAddr.sin_addr.s_addr = info->hostIp;
if (connect(clientSocket, (struct sockaddr *)&serverAddr, sizeof(serverAddr)) < 0) {
uError("TCP: failed to connect port %s:%d since %s", taosIpStr(info->hostIp), info->port, strerror(errno));
taosCloseSocket(&pSocket);
return -1;
}
taosKeepTcpAlive(pSocket);
sprintf(buffer, "client send TCP pkg to %s:%d, content: 1122334455", taosIpStr(info->hostIp), info->port);
sprintf(buffer + info->pktLen - 16, "1122334455667788");
int32_t ret = taosWriteMsg(pSocket, buffer, info->pktLen);
if (ret < 0) {
uError("TCP: failed to write msg to %s:%d since %s", taosIpStr(info->hostIp), info->port, strerror(errno));
taosCloseSocket(&pSocket);
return -1;
}
ret = taosReadMsg(pSocket, buffer, info->pktLen);
if (ret < 0) {
uError("TCP: failed to read msg from %s:%d since %s", taosIpStr(info->hostIp), info->port, strerror(errno));
taosCloseSocket(&pSocket);
return -1;
}
taosCloseSocket(&pSocket);
return 0;
}
static int32_t taosNetCheckUdpPort(STestInfo *info) {
SOCKET clientSocket;
char buffer[BUFFER_SIZE] = {0};
int32_t iDataNum = 0;
int32_t bufSize = 1024000;
struct sockaddr_in serverAddr;
if ((clientSocket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) {
uError("failed to create udp client socket since %s", strerror(errno));
return -1;
}
TdSocketPtr pSocket = (TdSocketPtr)taosMemoryMalloc(sizeof(TdSocket));
if (pSocket == NULL) {
taosCloseSocketNoCheck1(clientSocket);
return -1;
}
pSocket->fd = clientSocket;
pSocket->refId = 0;
if (taosSetSockOpt(pSocket, SOL_SOCKET, SO_SNDBUF, (void *)&bufSize, sizeof(bufSize)) != 0) {
uError("failed to set the send buffer size for UDP socket\n");
taosCloseSocket(&pSocket);
return -1;
}
if (taosSetSockOpt(pSocket, SOL_SOCKET, SO_RCVBUF, (void *)&bufSize, sizeof(bufSize)) != 0) {
uError("failed to set the receive buffer size for UDP socket\n");
taosCloseSocket(&pSocket);
return -1;
}
serverAddr.sin_family = AF_INET;
serverAddr.sin_port = htons(info->port);
serverAddr.sin_addr.s_addr = info->hostIp;
struct in_addr ipStr;
memcpy(&ipStr, &info->hostIp, 4);
sprintf(buffer, "client send udp pkg to %s:%d, content: 1122334455", taosInetNtoa(ipStr), info->port);
sprintf(buffer + info->pktLen - 16, "1122334455667788");
socklen_t sin_size = sizeof(*(struct sockaddr *)&serverAddr);
iDataNum = taosSendto(pSocket, buffer, info->pktLen, 0, (struct sockaddr *)&serverAddr, (int32_t)sin_size);
if (iDataNum < 0 || iDataNum != info->pktLen) {
uError("UDP: failed to perform sendto func since %s", strerror(errno));
taosCloseSocket(&pSocket);
return -1;
}
memset(buffer, 0, BUFFER_SIZE);
sin_size = sizeof(*(struct sockaddr *)&serverAddr);
iDataNum = recvfrom(clientSocket, buffer, BUFFER_SIZE, 0, (struct sockaddr *)&serverAddr, &sin_size);
if (iDataNum < 0 || iDataNum != info->pktLen) {
uError("UDP: received ack:%d bytes(expect:%d) from port:%d since %s", iDataNum, info->pktLen, info->port,
strerror(errno)); taosCloseSocket(&pSocket); return -1;
}
taosCloseSocket(&pSocket);
return 0;
}
static void taosNetCheckPort(uint32_t hostIp, int32_t startPort, int32_t endPort, int32_t pktLen) {
int32_t ret;
STestInfo info;
memset(&info, 0, sizeof(STestInfo));
info.hostIp = hostIp;
info.pktLen = pktLen;
for (int32_t port = startPort; port <= endPort; port++) {
info.port = port;
ret = taosNetCheckTcpPort(&info);
if (ret != 0) {
printf("failed to test TCP port:%d\n", port);
} else {
printf("successed to test TCP port:%d\n", port);
}
ret = taosNetCheckUdpPort(&info);
if (ret != 0) {
printf("failed to test UDP port:%d\n", port);
} else {
printf("successed to test UDP port:%d\n", port);
}
} }
} }
static void taosNetTestClient(char *host, int32_t startPort, int32_t pkgLen) { static void shellProcessMsg(void *p, SRpcMsg *pRpc, SEpSet *pEpSet) {
uInfo("work as client, host:%s Port:%d pkgLen:%d\n", host, startPort, pkgLen); printf("net test request is received, size:%d\n", pRpc->contLen);
SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = 0};
uint32_t serverIp = taosGetIpv4FromFqdn(host); rsp.pCont = rpcMallocCont(shell.args.pktLen);
if (serverIp == 0xFFFFFFFF) { rsp.contLen = shell.args.pktLen;
uError("failed to resolve fqdn:%s", host); rpcSendResponse(&rsp);
exit(-1);
}
uInfo("server ip:%s is resolved from host:%s", taosIpStr(serverIp), host);
taosNetCheckPort(serverIp, startPort, startPort, pkgLen);
} }
static void taosNetTestServer(char *host, int32_t startPort, int32_t pkgLen) { void shellNettestHandler(int32_t signum, void *sigInfo, void *context) { shellExit(); }
uInfo("work as server, host:%s Port:%d pkgLen:%d\n", host, startPort, pkgLen);
int32_t port = startPort; static void shellWorkAsServer() {
int32_t num = 1; SRpcInit rpcInit = {0};
if (num < 0) num = 1; rpcInit.localPort = shell.args.port;
rpcInit.label = "CHK";
rpcInit.numOfThreads = tsNumOfRpcThreads;
rpcInit.cfp = (RpcCfp)shellProcessMsg;
rpcInit.sessions = 10;
rpcInit.connType = TAOS_CONN_SERVER;
rpcInit.idleTime = tsShellActivityTimer * 1000;
TdThread *pids = taosMemoryMalloc(2 * num * sizeof(TdThread)); void *serverRpc = rpcOpen(&rpcInit);
STestInfo *tinfos = taosMemoryMalloc(num * sizeof(STestInfo)); if (serverRpc == NULL) {
STestInfo *uinfos = taosMemoryMalloc(num * sizeof(STestInfo)); printf("failed to init net test server since %s", terrstr());
for (int32_t i = 0; i < num; i++) {
STestInfo *tcpInfo = tinfos + i;
tcpInfo->port = port + i;
tcpInfo->pktLen = pkgLen;
if (taosThreadCreate(pids + i, NULL, taosNetBindTcpPort, tcpInfo) != 0) {
uInfo("failed to create TCP test thread, %s:%d", tcpInfo->hostFqdn, tcpInfo->port);
exit(-1);
}
STestInfo *udpInfo = uinfos + i;
udpInfo->port = port + i;
tcpInfo->pktLen = pkgLen;
if (taosThreadCreate(pids + num + i, NULL, taosNetBindUdpPort, udpInfo) != 0) {
uInfo("failed to create UDP test thread, %s:%d", tcpInfo->hostFqdn, tcpInfo->port);
exit(-1);
}
} }
for (int32_t i = 0; i < num; i++) { printf("net test server is initialized\n");
taosThreadJoin(pids[i], NULL);
taosThreadJoin(pids[(num + i)], NULL); taosSetSignal(SIGTERM, shellNettestHandler);
} while (1) taosMsleep(10);
} }
static void taosNetCheckSpeed(char *host, int32_t port, int32_t pkgLen, void shellTestNetWork() {
int32_t pkgNum, char *pkgType) { if (strcmp(shell.args.netrole, "client") == 0) {
#if 0 shellWorkAsClient();
// record config
int32_t compressTmp = tsCompressMsgSize;
int32_t maxUdpSize = tsRpcMaxUdpSize;
int32_t forceTcp = tsRpcForceTcp;
if (0 == strcmp("tcp", pkgType)){
tsRpcForceTcp = 1;
tsRpcMaxUdpSize = 0; // force tcp
} else {
tsRpcForceTcp = 0;
tsRpcMaxUdpSize = INT_MAX;
}
tsCompressMsgSize = -1;
SEpSet epSet;
SRpcMsg reqMsg;
SRpcMsg rspMsg;
void * pRpcConn;
char secretEncrypt[32] = {0};
char spi = 0;
pRpcConn = taosNetInitRpc(secretEncrypt, spi);
if (NULL == pRpcConn) {
uError("failed to init client rpc");
return;
} }
printf("check net spend, host:%s port:%d pkgLen:%d pkgNum:%d pkgType:%s\n\n", host, port, pkgLen, pkgNum, pkgType); if (strcmp(shell.args.netrole, "server") == 0) {
int32_t totalSucc = 0; shellWorkAsServer();
uint64_t startT = taosGetTimestampUs();
for (int32_t i = 1; i <= pkgNum; i++) {
uint64_t startTime = taosGetTimestampUs();
memset(&epSet, 0, sizeof(SEpSet));
strcpy(epSet.eps[0].fqdn, host);
epSet.eps[0].port = port;
epSet.numOfEps = 1;
reqMsg.msgType = TDMT_DND_NETWORK_TEST;
reqMsg.pCont = rpcMallocCont(pkgLen);
reqMsg.contLen = pkgLen;
reqMsg.code = 0;
reqMsg.handle = NULL; // rpc handle returned to app
reqMsg.ahandle = NULL; // app handle set by client
strcpy(reqMsg.pCont, "nettest speed");
rpcSendRecv(pRpcConn, &epSet, &reqMsg, &rspMsg);
int code = 0;
if ((rspMsg.code != 0) || (rspMsg.msgType != TDMT_DND_NETWORK_TEST + 1)) {
uError("ret code 0x%x %s", rspMsg.code, tstrerror(rspMsg.code));
code = -1;
}else{
totalSucc ++;
}
rpcFreeCont(rspMsg.pCont);
uint64_t endTime = taosGetTimestampUs();
uint64_t el = endTime - startTime;
printf("progress:%5d/%d\tstatus:%d\tcost:%8.2lf ms\tspeed:%8.2lf MB/s\n", i, pkgNum, code, el/1000.0,
pkgLen/(el/1000000.0)/1024.0/1024.0);
} }
int64_t endT = taosGetTimestampUs();
uint64_t elT = endT - startT;
printf("\ntotal succ:%5d/%d\tcost:%8.2lf ms\tspeed:%8.2lf MB/s\n", totalSucc, pkgNum, elT/1000.0,
pkgLen/(elT/1000000.0)/1024.0/1024.0*totalSucc);
rpcClose(pRpcConn);
// return config
tsCompressMsgSize = compressTmp;
tsRpcMaxUdpSize = maxUdpSize;
tsRpcForceTcp = forceTcp;
return;
#endif
} }
void taosNetTest(char *role, char *host, int32_t port, int32_t pkgLen, int32_t pkgNum, char *pkgType) {
tsLogEmbedded = 1;
if (host == NULL) host = tsLocalFqdn;
if (port == 0) port = tsServerPort;
if (0 == strcmp("speed", role)) {
if (pkgLen <= MIN_SPEED_PKG_LEN) pkgLen = MIN_SPEED_PKG_LEN;
if (pkgLen > MAX_SPEED_PKG_LEN) pkgLen = MAX_SPEED_PKG_LEN;
if (pkgNum <= MIN_SPEED_PKG_NUM) pkgNum = MIN_SPEED_PKG_NUM;
if (pkgNum > MAX_SPEED_PKG_NUM) pkgNum = MAX_SPEED_PKG_NUM;
} else {
if (pkgLen <= 10) pkgLen = 1000;
if (pkgLen > MAX_PKG_LEN) pkgLen = MAX_PKG_LEN;
}
if (0 == strcmp("client", role)) {
taosNetTestClient(host, port, pkgLen);
} else if (0 == strcmp("server", role)) {
taosNetTestServer(host, port, pkgLen);
} else if (0 == strcmp("speed", role)) {
tsLogEmbedded = 0;
char type[10] = {0};
taosNetCheckSpeed(host, port, pkgLen, pkgNum, strtolower(type, pkgType));
} else {
TASSERT(1);
}
tsLogEmbedded = 0;
}
#endif