Merge pull request #5061 from taosdata/feature/sim

Feature/sim

src/kit/shell/src/shellLinux.c

@@ -47,7 +47,7 @@ static struct argp_option options[] = {
   {"thread",     'T', "THREADNUM",  0,                   "Number of threads when using multi-thread to import data."},
   {"database",   'd', "DATABASE",   0,                   "Database to use when connecting to the server."},
   {"timezone",   't', "TIMEZONE",   0,                   "Time zone of the shell, default is local."},
-  {"netrole",    'n', "NETROLE",    0,                   "Net role when network connectivity test, default is startup, options: client|server|rpc|startup."},
+  {"netrole",    'n', "NETROLE",    0,                   "Net role when network connectivity test, default is startup, options: client|server|rpc|startup|sync."},
   {"pktlen",     'l', "PKTLEN",     0,                   "Packet length used for net test, default is 1000 bytes."},
   {0}};
 

src/kit/shell/src/shellWindows.c

@@ -52,7 +52,7 @@ void printHelp() {
   printf("%s%s\n", indent, "-t");
   printf("%s%s%s\n", indent, indent, "Time zone of the shell, default is local.");
   printf("%s%s\n", indent, "-n");
-  printf("%s%s%s\n", indent, indent, "Net role when network connectivity test, default is startup, options: client|server|rpc|startup.");
+  printf("%s%s%s\n", indent, indent, "Net role when network connectivity test, default is startup, options: client|server|rpc|startup|sync.");
   printf("%s%s\n", indent, "-l");
   printf("%s%s%s\n", indent, indent, "Packet length used for net test, default is 1000 bytes.");
   printf("%s%s\n", indent, "-V");

src/sync/inc/syncMsg.h

@@ -37,7 +37,8 @@ typedef enum {
   TAOS_SMSG_SETUP_RSP     = 12,
   TAOS_SMSG_SYNC_FILE     = 13,
   TAOS_SMSG_SYNC_FILE_RSP = 14,
-  TAOS_SMSG_END           = 15,
+  TAOS_SMSG_TEST          = 15,
+  TAOS_SMSG_END           = 16
 } ESyncMsgType;
 
 typedef enum {
@@ -132,6 +133,7 @@ void syncBuildSyncReqMsg(SSyncMsg *pMsg, int32_t vgId);
 void syncBuildSyncDataMsg(SSyncMsg *pMsg, int32_t vgId);
 void syncBuildSyncSetupMsg(SSyncMsg *pMsg, int32_t vgId);
 void syncBuildPeersStatus(SPeersStatus *pMsg, int32_t vgId);
+void syncBuildSyncTestMsg(SSyncMsg *pMsg, int32_t vgId);
 
 void syncBuildFileAck(SFileAck *pMsg, int32_t vgId);
 void syncBuildFileInfo(SFileInfo *pMsg, int32_t vgId);

src/sync/src/syncArbitrator.c

@@ -27,6 +27,7 @@
 #include "syncInt.h"
 #include "syncTcp.h"
 
+extern void    syncProcessTestMsg(SSyncMsg *pMsg, SOCKET connFd);
 static void    arbSignalHandler(int32_t signum, void *sigInfo, void *context);
 static void    arbProcessIncommingConnection(SOCKET connFd, uint32_t sourceIp);
 static void    arbProcessBrokenLink(int64_t rid);
@@ -118,6 +119,11 @@ static void arbProcessIncommingConnection(SOCKET connFd, uint32_t sourceIp) {
     return;
   }
 
+  if (msg.head.type == TAOS_SMSG_TEST) {
+    syncProcessTestMsg(&msg, connFd);
+    return;
+  }
+
   SNodeConn *pNode = calloc(sizeof(SNodeConn), 1);
   if (pNode == NULL) {
     sError("failed to allocate memory since %s", strerror(errno));

src/sync/src/syncMain.c

@@ -1179,6 +1179,20 @@ static void syncCreateRestoreDataThread(SSyncPeer *pPeer) {
   }
 }
 
+void syncProcessTestMsg(SSyncMsg *pMsg, SOCKET connFd) {
+  sInfo("recv sync test msg");
+
+  SSyncMsg rsp;
+  syncBuildSyncTestMsg(&rsp, -1);
+  if (taosWriteMsg(connFd, &rsp, sizeof(SSyncMsg)) != sizeof(SSyncMsg)) {
+    sInfo("failed to send sync test rsp since %s", strerror(errno));
+  }
+
+  sInfo("send sync test rsp");
+  taosMsleep(1000);
+  taosCloseSocket(connFd);
+}
+
 static void syncProcessIncommingConnection(SOCKET connFd, uint32_t sourceIp) {
   char    ipstr[24];
   int32_t i;
@@ -1200,6 +1214,11 @@ static void syncProcessIncommingConnection(SOCKET connFd, uint32_t sourceIp) {
     return;
   }
 
+  if (msg.head.type == TAOS_SMSG_TEST) {
+    syncProcessTestMsg(&msg, connFd);
+    return;
+  }
+
   int32_t     vgId = msg.head.vgId;
   SSyncNode **ppNode = taosHashGet(tsVgIdHash, &vgId, sizeof(int32_t));
   if (ppNode == NULL || *ppNode == NULL) {

src/sync/src/syncMsg.c

@@ -86,6 +86,7 @@ static void syncBuildMsg(SSyncMsg *pMsg, int32_t vgId, ESyncMsgType type) {
 void syncBuildSyncReqMsg(SSyncMsg *pMsg, int32_t vgId) { syncBuildMsg(pMsg, vgId, TAOS_SMSG_SYNC_REQ); }
 void syncBuildSyncDataMsg(SSyncMsg *pMsg, int32_t vgId) { syncBuildMsg(pMsg, vgId, TAOS_SMSG_SYNC_DATA); }
 void syncBuildSyncSetupMsg(SSyncMsg *pMsg, int32_t vgId) { syncBuildMsg(pMsg, vgId, TAOS_SMSG_SETUP); }
+void syncBuildSyncTestMsg(SSyncMsg *pMsg, int32_t vgId) { syncBuildMsg(pMsg, vgId, TAOS_SMSG_TEST); }
 
 void syncBuildPeersStatus(SPeersStatus *pMsg, int32_t vgId) {
   pMsg->head.type = TAOS_SMSG_STATUS;

src/util/CMakeLists.txt

@@ -2,6 +2,7 @@ CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
 PROJECT(TDengine)
 
 INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/rpc/inc)
+INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/sync/inc)
 AUX_SOURCE_DIRECTORY(src SRC)
 ADD_LIBRARY(tutil ${SRC})
 TARGET_LINK_LIBRARIES(tutil pthread osdetail lz4 z)

src/util/src/tnettest.c

@@ -23,6 +23,8 @@
 #include "tsocket.h"
 #include "trpc.h"
 #include "rpcHead.h"
+#include "tchecksum.h"
+#include "syncMsg.h"
 
 #define MAX_PKG_LEN (64 * 1000)
 #define BUFFER_SIZE (MAX_PKG_LEN + 1024)
@@ -408,13 +410,51 @@ static void taosNetTestStartup(char *host, int32_t port) {
   free(pStep);
 }
 
+static void taosNetCheckSync(char *host, int32_t port) {
+  uint32_t ip = taosGetIpv4FromFqdn(host);
+  if (ip == 0xffffffff) {
+    uError("failed to get IP address from %s since %s", host, strerror(errno));
+    return;
+  }
+
+  SOCKET connFd = taosOpenTcpClientSocket(ip, (uint16_t)port, 0);
+  if (connFd < 0) {
+    uError("failed to create socket while test port:%d since %s", port, strerror(errno));
+    return;
+  }
+
+  SSyncMsg msg;
+  memset(&msg, 0, sizeof(SSyncMsg));
+  SSyncHead *pHead = &msg.head;
+  pHead->type = TAOS_SMSG_TEST;
+  pHead->protocol = SYNC_PROTOCOL_VERSION;
+  pHead->signature = SYNC_SIGNATURE;
+  pHead->code = 0;
+  pHead->cId = 0;
+  pHead->vgId = -1;
+  pHead->len = sizeof(SSyncMsg) - sizeof(SSyncHead);
+  taosCalcChecksumAppend(0, (uint8_t *)pHead, sizeof(SSyncHead));
+
+  if (taosWriteMsg(connFd, &msg, sizeof(SSyncMsg)) != sizeof(SSyncMsg)) {
+    uError("failed to test port:%d while send msg since %s", port, strerror(errno));
+    return;
+  }
+
+  if (taosReadMsg(connFd, &msg, sizeof(SSyncMsg)) != sizeof(SSyncMsg)) {
+    uError("failed to test port:%d while recv msg since %s", port, strerror(errno));
+  }
+
+  uInfo("successed to test TCP port:%d", port);
+  taosCloseSocket(connFd);
+}
+
 static void taosNetTestRpc(char *host, int32_t startPort, int32_t pkgLen) {
-  int32_t endPort = startPort + 9;
+  int32_t endPort = startPort + TSDB_PORT_SYNC;
   char    spi = 0;
 
   uInfo("check rpc, host:%s startPort:%d endPort:%d pkgLen:%d\n", host, startPort, endPort, pkgLen);
-    
+
-  for (uint16_t port = startPort; port <= endPort; port++) {
+  for (uint16_t port = startPort; port < endPort; port++) {
     int32_t sendpkgLen;
     if (pkgLen <= tsRpcMaxUdpSize) {
       sendpkgLen = tsRpcMaxUdpSize + 1000;
@@ -442,6 +482,9 @@ static void taosNetTestRpc(char *host, int32_t startPort, int32_t pkgLen) {
       uInfo("successed to test UDP port:%d", port);
     }
   }
+
+  taosNetCheckSync(host, startPort + TSDB_PORT_SYNC);
+  taosNetCheckSync(host, startPort + TSDB_PORT_ARBITRATOR);
 }
 
 static void taosNetTestClient(char *host, int32_t startPort, int32_t pkgLen) {
@@ -508,6 +551,8 @@ void taosNetTest(char *role, char *host, int32_t port, int32_t pkgLen) {
     taosNetTestServer(host, port, pkgLen);
   } else if (0 == strcmp("rpc", role)) {
     taosNetTestRpc(host, port, pkgLen);
+  } else if (0 == strcmp("sync", role)) {
+    taosNetCheckSync(host, port);
   } else if (0 == strcmp("startup", role)) {
     taosNetTestStartup(host, port);
   } else {