[TD-1076]

src/kit/CMakeLists.txt

@@ -4,4 +4,6 @@ PROJECT(TDengine)
 ADD_SUBDIRECTORY(shell)
 ADD_SUBDIRECTORY(taosdemo)
 ADD_SUBDIRECTORY(taosdump)
-ADD_SUBDIRECTORY(taosmigrate)
+ADD_SUBDIRECTORY(taosmigrate)
+#ADD_SUBDIRECTORY(taosClusterTest)
+ADD_SUBDIRECTORY(taosnetwork)

src/kit/taosnetwork/CMakeLists.txt

@@ -0,0 +1,10 @@
+CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
+PROJECT(TDengine)
+
+IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
+  AUX_SOURCE_DIRECTORY(. SRC)
+  ADD_EXECUTABLE(taosClient client.c)
+  ADD_EXECUTABLE(taosServer server.c)
+  TARGET_LINK_LIBRARIES( taosServer -lpthread -lm -lrt )
+  TARGET_LINK_LIBRARIES( taosClient -lpthread -lm -lrt )
+ENDIF ()

src/kit/taosnetwork/client.c

@@ -28,23 +28,27 @@
 #include <sys/types.h>
 #include <unistd.h>
 
-#define BUFFER_SIZE 200
+#define MAX_PKG_LEN (16*1000)
+#define BUFFER_SIZE (MAX_PKG_LEN + 1024)
 
 typedef struct {
   int   port;
   char *host;
+  uint16_t pktLen;
 } info_s;
 
 typedef struct Arguments {
   char *   host;
   uint16_t port;
   uint16_t max_port;
+  uint16_t pktLen;
 } SArguments;
 
 static struct argp_option options[] = {
     {0, 'h', "host", 0, "The host to connect to TDEngine. Default is localhost.", 0},
-    {0, 'p', "port", 0, "The TCP or UDP port number to use for the connection. Default is 6041.", 1},
+    {0, 'p', "port", 0, "The TCP or UDP port number to use for the connection. Default is 6030.", 1},
-    {0, 'm', "max port", 0, "The max TCP or UDP port number to use for the connection. Default is 6050.", 2}};
+    {0, 'm', "max port", 0, "The max TCP or UDP port number to use for the connection. Default is 6060.", 2},
+    {0, 'l', "test pkg len", 0, "The len of pkg for test. Default is 1000 Bytes, max not greater than 16k Bytes.\nNotes: This parameter must be consistent between the client and the server.", 3}};
 
 static error_t parse_opt(int key, char *arg, struct argp_state *state) {
 
@@ -59,6 +63,12 @@ static error_t parse_opt(int key, char *arg, struct argp_state *state) {
     case 'm':
       arguments->max_port = atoi(arg);
       break;
+    case 'l':
+      arguments->pktLen = atoi(arg);
+      break;
+    default:
+      printf("unknow parameter!\n");
+      break;
   }
   return 0;
 }
@@ -75,7 +85,7 @@ int checkTcpPort(info_s *info) {
   char               recvbuf[BUFFER_SIZE];
   int                iDataNum;
   if ((clientSocket = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
-    perror("socket");
+    printf("socket() fail: %s\n", strerror(errno));
     return -1;
   }
   serverAddr.sin_family = AF_INET;
@@ -85,26 +95,30 @@ int checkTcpPort(info_s *info) {
 
   //printf("=================================\n");
   if (connect(clientSocket, (struct sockaddr *)&serverAddr, sizeof(serverAddr)) < 0) {
-    perror("connect");
+    printf("connect() fail: %s\n", strerror(errno));
     return -1;
   }
   //printf("Connect to: %s:%d...success\n", host, port);
+  memset(sendbuf, 0, BUFFER_SIZE);
+  memset(recvbuf, 0, BUFFER_SIZE);
 
-  sprintf(sendbuf, "send port_%d", port);
+  sprintf(sendbuf, "client send tcp pkg to %s:%d, content: 1122334455", host, port);
-  send(clientSocket, sendbuf, strlen(sendbuf), 0);
+  sprintf(sendbuf + info->pktLen - 16, "1122334455667788");
-  //printf("Send msg_%d: %s\n", port, sendbuf);
+
+  send(clientSocket, sendbuf, info->pktLen, 0);
 
-  recvbuf[0] = '\0';
   iDataNum = recv(clientSocket, recvbuf, BUFFER_SIZE, 0);
-  recvbuf[iDataNum] = '\0';
+  if (iDataNum < info->pktLen) {
-  //printf("Read ack msg_%d: %s\n", port, recvbuf);
+    printf("Read ack pkg len: %d, less than req pkg len: %d from tcp port: %d\n", iDataNum, info->pktLen, port);
+    return -1;
+  }
+  //printf("Read ack pkg len:%d from tcp port: %d, buffer: %s  %s\n", info->pktLen, port, recvbuf, recvbuf+iDataNum-8);
 
-  //printf("=================================\n");
   close(clientSocket);
   return 0;
 }
 
-void *checkUdpPort(info_s *info) {
+int checkUdpPort(info_s *info) {
   int   port = info->port;
   char *host = info->host;
   int   clientSocket;
@@ -117,60 +131,74 @@ void *checkUdpPort(info_s *info) {
     perror("socket");
     return -1;
   }
+  
   serverAddr.sin_family = AF_INET;
   serverAddr.sin_port = htons(port);
   serverAddr.sin_addr.s_addr = inet_addr(host);
+  
+  memset(sendbuf, 0, BUFFER_SIZE);
+  memset(recvbuf, 0, BUFFER_SIZE);
 
-  sprintf(sendbuf, "send msg port_%d by udp", port);
+  sprintf(sendbuf, "client send udp pkg to %s:%d, content: 1122334455", host, port);
+  sprintf(sendbuf + info->pktLen - 16, "1122334455667788");
 
   socklen_t sin_size = sizeof(*(struct sockaddr *)&serverAddr);
 
-  int code = sendto(clientSocket, sendbuf, strlen(sendbuf), 0, (struct sockaddr *)&serverAddr, (int)sin_size);
+  int code = sendto(clientSocket, sendbuf, info->pktLen, 0, (struct sockaddr *)&serverAddr, (int)sin_size);
   if (code < 0) {
     perror("sendto");
     return -1;
   }
 
-  //printf("Send msg_%d by udp: %s\n", port, sendbuf);
-
-  recvbuf[0] = '\0';
   iDataNum = recvfrom(clientSocket, recvbuf, BUFFER_SIZE, 0, (struct sockaddr *)&serverAddr, &sin_size);
-  recvbuf[iDataNum] = '\0';
-  //printf("Read ack msg_%d from udp: %s\n", port, recvbuf);
 
+  if (iDataNum < info->pktLen) {
+    printf("Read ack pkg len: %d, less than req pkg len: %d from udp port: %d\n", iDataNum, info->pktLen, port);
+    return -1;
+  }
+  
+  //printf("Read ack pkg len:%d from udp port: %d, buffer: %s  %s\n", info->pktLen, port, recvbuf, recvbuf+iDataNum-8);
   close(clientSocket);
   return 0;
 }
 
 int main(int argc, char *argv[]) {
-  SArguments arguments = {"127.0.0.1", 6030, 6060};
+  SArguments arguments = {"127.0.0.1", 6030, 6060, 1000};
   info_s  info;
   int ret;
   
   argp_parse(&argp, argc, argv, 0, 0, &arguments);
+  if (arguments.pktLen > MAX_PKG_LEN) {
+    printf("test pkg len overflow: %d, max len not greater than %d bytes\n", arguments.pktLen, MAX_PKG_LEN);
+    exit(0);
+  }
 
-  printf("host: %s\tport: %d\tmax_port: %d\n\n", arguments.host, arguments.port, arguments.max_port);
+  printf("host: %s\tport: %d\tmax_port: %d\tpkgLen: %d\n", arguments.host, arguments.port, arguments.max_port, arguments.pktLen);
 
   int   port = arguments.port;
 
   info.host = arguments.host;
+  info.pktLen = arguments.pktLen;
 
-  for (; port < arguments.max_port; port++) {
+  for (; port <= arguments.max_port; port++) {
-    printf("test: %s:%d\n", info.host, port);
+    //printf("test: %s:%d\n", info.host, port);
+    printf("\n");
 
     info.port = port;
     ret = checkTcpPort(&info);
     if (ret != 0) {
-      printf("tcp port:%d test fail.", port);
+      printf("tcp port:%d test fail.\t\t", port);
     } else {
-      printf("tcp port:%d test ok.", port);
+      printf("tcp port:%d test ok.\t\t", port);
     }
     
-    checkUdpPort(&info);
+    ret = checkUdpPort(&info);
     if (ret != 0) {
-      printf("udp port:%d test fail.", port);
+      printf("udp port:%d test fail.\t\t", port);
     } else {
-      printf("udp port:%d test ok.", port);
+      printf("udp port:%d test ok.\t\t", port);
     }
   }
+  printf("\n");
+  return 0;
 }

src/kit/taosnetwork/server.c

@@ -29,22 +29,26 @@
 #include <unistd.h>
 #include <pthread.h>
 
-#define BUFFER_SIZE 200
+#define MAX_PKG_LEN (16*1000)
+#define BUFFER_SIZE (MAX_PKG_LEN + 1024)
 
 typedef struct {
   int port;
+  uint16_t pktLen;
 } info_s;
 
 typedef struct Arguments {
   char *   host;
   uint16_t port;
   uint16_t max_port;
+  uint16_t pktLen;
 } SArguments;
 
 static struct argp_option options[] = {
     {0, 'h', "host", 0, "The host to connect to TDEngine. Default is localhost.", 0},
     {0, 'p', "port", 0, "The TCP or UDP port number to use for the connection. Default is 6041.", 1},
-    {0, 'm', "max port", 0, "The max TCP or UDP port number to use for the connection. Default is 6050.", 2}};
+    {0, 'm', "max port", 0, "The max TCP or UDP port number to use for the connection. Default is 6050.", 2},
+    {0, 'l', "test pkg len", 0, "The len of pkg for test. Default is 1000 Bytes, max not greater than 16k Bytes.\nNotes: This parameter must be consistent between the client and the server.", 3}};
 
 static error_t parse_opt(int key, char *arg, struct argp_state *state) {
 
@@ -59,6 +63,11 @@ static error_t parse_opt(int key, char *arg, struct argp_state *state) {
     case 'm':
       arguments->max_port = atoi(arg);
       break;
+    case 'l':
+      arguments->pktLen = atoi(arg);
+      break;
+    default:
+      break;
   }
   return 0;
 }
@@ -75,10 +84,10 @@ static void *bindTcpPort(void *sarg) {
   int                addr_len = sizeof(clientAddr);
   int                client;
   char               buffer[BUFFER_SIZE];
-  int                iDataNum;
+  int                iDataNum = 0;
 
   if ((serverSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) {
-    perror("socket");
+    printf("socket() fail: %s", strerror(errno));
     return NULL;
   }
 
@@ -88,12 +97,12 @@ static void *bindTcpPort(void *sarg) {
   server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
 
   if (bind(serverSocket, (struct sockaddr *)&server_addr, sizeof(server_addr)) < 0) {
-    perror("connect");
+    printf("port:%d bind() fail: %s", port, strerror(errno));
     return NULL;
   }
 
   if (listen(serverSocket, 5) < 0) {
-    perror("listen");
+    printf("listen() fail: %s", strerror(errno));
     return NULL;
   }
 
@@ -101,34 +110,39 @@ static void *bindTcpPort(void *sarg) {
   while (1) {
     client = accept(serverSocket, (struct sockaddr *)&clientAddr, (socklen_t *)&addr_len);
     if (client < 0) {
-      perror("accept");
+      printf("accept() fail: %s", strerror(errno));
       continue;
     }
-    //printf("=================================\n");
 
-    printf("Client ip is %s, Server port is %d\n", inet_ntoa(clientAddr.sin_addr), port);
+    memset(buffer, 0, BUFFER_SIZE);
-    while (1) {
+    int   nleft, nread;
-      buffer[0] = '\0';
+    char *ptr = buffer;
-      iDataNum = recv(client, buffer, BUFFER_SIZE, 0);
+    nleft = pinfo->pktLen;
+    while (nleft > 0) {
+      nread = recv(client, buffer, BUFFER_SIZE, 0);
 
-      if (iDataNum < 0) {
+      if (nread == 0) {
-        perror("recv null");
-        continue;
-      }
-      if (iDataNum > 0) {
-        buffer[iDataNum] = '\0';
-        //printf("read msg:%s\n", buffer);
-        if (strcmp(buffer, "quit") == 0) break;
-        buffer[0] = '\0';
-
-        sprintf(buffer, "ack port_%d", port);
-        //printf("send ack msg:%s\n", buffer);
-
-        send(client, buffer, strlen(buffer), 0);
         break;
-      }
+      } else if (nread < 0) {
+        if (errno == EINTR) {
+          continue;
+        } else {
+          printf("recv Client: %s pkg from TCP port: %d fail:%s.\n", inet_ntoa(clientAddr.sin_addr), port, strerror(errno));
+          close(serverSocket);
+          return NULL;
+        }
+      } else {
+        nleft -= nread;
+        ptr += nread;
+        iDataNum += nread;
+      }      
+    }
+    
+    printf("recv Client: %s pkg from TCP port: %d, pkg len: %d\n", inet_ntoa(clientAddr.sin_addr), port, iDataNum);
+    if (iDataNum > 0) {
+      send(client, buffer, iDataNum, 0);
+      break;
     }
-    //printf("=================================\n");
   }
   close(serverSocket);
   return NULL;
@@ -143,7 +157,7 @@ static void *bindUdpPort(void *sarg) {
   struct sockaddr_in clientAddr;
   char               buffer[BUFFER_SIZE];
   int                iDataNum;
-
+    
   if ((serverSocket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) {
     perror("socket");
     return NULL;
@@ -160,10 +174,9 @@ static void *bindUdpPort(void *sarg) {
   }
 
   socklen_t sin_size;
-  //printf("Bind port: %d success\n", port);
 
   while (1) {
-    buffer[0] = '\0';
+    memset(buffer, 0, BUFFER_SIZE);
 
     sin_size = sizeof(*(struct sockaddr *)&server_addr);
 
@@ -174,19 +187,10 @@ static void *bindUdpPort(void *sarg) {
       continue;
     }
     if (iDataNum > 0) {
-      //printf("=================================\n");
+      printf("recv Client: %s pkg from UDP port: %d, pkg len: %d\n", inet_ntoa(clientAddr.sin_addr), port, iDataNum);
+      //printf("Read msg from udp:%s ... %s\n", buffer, buffer+iDataNum-16);
 
-      printf("Client ip is %s, Server port is %d\n", inet_ntoa(clientAddr.sin_addr), port);
+      sendto(serverSocket, buffer, iDataNum, 0, (struct sockaddr *)&clientAddr, (int)sin_size);
-      buffer[iDataNum] = '\0';
-      //printf("Read msg from udp:%s\n", buffer);
-      if (strcmp(buffer, "quit") == 0) break;
-      buffer[0] = '\0';
-
-      sprintf(buffer, "ack port_%d by udp", port);
-      //printf("Send ack msg by udp:%s\n", buffer);
-
-      sendto(serverSocket, buffer, strlen(buffer), 0, (struct sockaddr *)&clientAddr, (int)sin_size);
-      //printf("=================================\n");
     }
   }
 
@@ -196,8 +200,13 @@ static void *bindUdpPort(void *sarg) {
 
 
 int main(int argc, char *argv[]) {
-  SArguments arguments = {"127.0.0.1", 6030, 6060};
+  SArguments arguments = {"127.0.0.1", 6030, 6060, 1000};
   argp_parse(&argp, argc, argv, 0, 0, &arguments);
+  if (arguments.pktLen > MAX_PKG_LEN) {
+    printf("test pkg len overflow: %d, max len not greater than %d bytes\n", arguments.pktLen, MAX_PKG_LEN);
+    exit(0);
+  }
+
   int port = arguments.port;
 
   int num = arguments.max_port - arguments.port + 1;
@@ -212,6 +221,7 @@ int main(int argc, char *argv[]) {
   for (size_t i = 0; i < num; i++) {
     info_s *tcpInfo = tinfos + i;
     tcpInfo->port = port + i;
+    tcpInfo->pktLen = arguments.pktLen;
 
     if (pthread_create(pids + i, NULL, bindTcpPort, tcpInfo) != 0) 
     {