Merge pull request #3532 from taosdata/feature/http

Feature/http

src/client/src/tscParseInsert.c

@@ -526,7 +526,7 @@ int tsParseValues(char **str, STableDataBlocks *pDataBlock, STableMeta *pTableMe
   int32_t   index = 0;
   SStrToken sToken;
 
-  int16_t numOfRows = 0;
+  int32_t numOfRows = 0;
 
   SSchema *pSchema = tscGetTableSchema(pTableMeta);
   STableComInfo tinfo = tscGetTableInfo(pTableMeta);

src/plugins/http/src/httpParser.c

@@ -238,6 +238,7 @@ static int32_t httpOnParseHeaderField(HttpParser *parser, const char *key, const
     httpTrace("context:%p, fd:%d, keepAlive:%d", pContext, pContext->fd, pContext->parser->keepAlive);
   }
 
+#if 0
   else if (0 == strcasecmp(key, "Content-Encoding")) {
     if (0 == strcmp(val, "gzip")) {
       parser->contentChunked = 1;
@@ -245,8 +246,9 @@ static int32_t httpOnParseHeaderField(HttpParser *parser, const char *key, const
     }
     return 0;
   }
+  #endif
 
-  else if (0 == strcasecmp(key, "Transfer-Encoding")) {
+  else if (0 == strcasecmp(key, "Transfer-Encoding") || 0 == strcasecmp(key, "Content-Encoding")) {
     if (strstr(val, "gzip")) {
       parser->transferGzip = 1;
       ehttp_gzip_conf_t      conf = {0};

src/plugins/http/src/httpServer.c

@@ -317,10 +317,10 @@ static bool httpReadData(HttpContext *pContext) {
   pContext->lastAccessTime = taosGetTimestampSec();
 
   char    buf[HTTP_STEP_SIZE + 1] = {0};
-  int32_t nread = (int32_t)taosReadSocket(pContext->fd, buf, sizeof(buf));
+  int32_t nread = (int32_t)taosReadSocket(pContext->fd, buf, HTTP_STEP_SIZE);
   if (nread > 0) {
     buf[nread] = '\0';
-    httpTrace("context:%p, fd:%d, nread:%d", pContext, pContext->fd, nread);
+    httpTraceL("context:%p, fd:%d, nread:%d content:%s", pContext, pContext->fd, nread, buf);
     int32_t ok = httpParseBuf(pParser, buf, nread);
 
     if (ok) {
@@ -341,7 +341,7 @@ static bool httpReadData(HttpContext *pContext) {
       httpTrace("context:%p, fd:%d, read not over yet, len:%d", pContext, pContext->fd, pParser->body.pos);
       return false;
     } else {
-      httpTraceL("context:%p, fd:%d, len:%d, body:%s", pContext, pContext->fd, pParser->body.pos, pParser->body.str);
+      httpDebug("context:%p, fd:%d, totalLen:%d", pContext, pContext->fd, pParser->body.pos);
       return true;
     }
   } else if (nread < 0) {

tests/script/general/http/autocreate.sim

@@ -24,5 +24,10 @@ print curl 127.0.0.1:7111/rest/sql -----> $system_content
 #  return -1
 #endi
 
+sql select * from db.win_cpu_windows_1_processor
+print rows: $rows
+if $rows != 1 then
+  return -1
+endi
 
 #system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file

tests/script/general/http/chunked.sim

+system sh/stop_dnodes.sh
+sleep 3000
+system sh/deploy.sh -n dnode1 -i 1
+system sh/cfg.sh -n dnode1 -c wallevel -v 0
+system sh/cfg.sh -n dnode1 -c http -v 1
+system sh/cfg.sh -n dnode1 -c maxSQLLength -v 7340032
+system sh/exec.sh -n dnode1 -s start
+
+sleep 3000
+sql connect
+
+print ============================ dnode1 start
+
+print ===============  step1 - prepare data
+sql create database d1
+sql use d1
+
+sql create table table_rest (ts timestamp, i int) 
+print sql length is 270KB
+restful d1 table_rest 1591072800 10000
+restful d1 table_rest 1591172800 10000
+restful d1 table_rest 1591272800 10000
+restful d1 table_rest 1591372800 10000
+restful d1 table_rest 1591472800 10000
+restful d1 table_rest 1591572800 10000
+restful d1 table_rest 1591672800 10000
+restful d1 table_rest 1591772800 10000
+restful d1 table_rest 1591872800 10000
+restful d1 table_rest 1591972800 10000
+
+sql select * from table_rest;
+print rows: $rows
+if $rows != 100000 then
+  return -1
+endi
+
+system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file

tests/script/general/http/gzip.sim

+system sh/stop_dnodes.sh
+sleep 3000
+system sh/deploy.sh -n dnode1 -i 1
+system sh/cfg.sh -n dnode1 -c wallevel -v 0
+system sh/cfg.sh -n dnode1 -c http -v 1
+system sh/cfg.sh -n dnode1 -c maxSQLLength -v 7340032
+system sh/exec.sh -n dnode1 -s start
+
+sleep 3000
+sql connect
+
+print ============================ dnode1 start
+
+print ===============  step1 - prepare data
+sql create database d1
+sql use d1
+
+sql create table table_rest (ts timestamp, i int) 
+print sql length is 270KB
+restful d1 table_rest 1591072800 10000 gzip
+sql select * from table_rest;
+print rows: $rows
+if $rows != 10000 then
+  return -1
+endi
+
+system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file

tests/script/general/http/testSuite.sim

+run general/http/autocreate.sim
+run general/http/chunked.sim
 run general/http/restful.sim
 run general/http/restful_insert.sim
 run general/http/restful_limit.sim

tests/script/jenkins/basic.txt

@@ -77,6 +77,8 @@ cd ../../../debug; make
 ./test.sh -f general/field/smallint.sim
 ./test.sh -f general/field/tinyint.sim
 
+./test.sh -f general/http/autocreate.sim
+./test.sh -f general/http/chunked.sim
 ./test.sh -f general/http/restful.sim
 ./test.sh -f general/http/restful_insert.sim
 ./test.sh -f general/http/restful_limit.sim

tests/test/c/CMakeLists.txt

@@ -39,5 +39,8 @@ IF (TD_LINUX)
   
   add_executable(httpTest httpTest.c)
   target_link_libraries(httpTest taos_static tutil common pthread mnode monitor http tsdb twal vnode cJson lz4)
+
+  add_executable(cacheTest cacheTest.c)
+  target_link_libraries(cacheTest taos_static tutil common pthread mnode monitor http tsdb twal vnode cJson lz4)
 ENDIF()
 

tests/test/c/cacheTest.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#define _DEFAULT_SOURCE
+#include <malloc.h>
+#include "os.h"
+#include "taos.h"
+#include "tcache.h"
+#include "tulog.h"
+#include "tutil.h"
+
+#define MAX_REFRESH_TIME_SEC 2
+#define MAX_RANDOM_POINTS 20000
+#define GREEN "\033[1;32m"
+#define NC "\033[0m"
+
+int32_t tsKeepTimeInSec = 3;
+int32_t tsNumOfRows = 1000000;
+int32_t tsSizeOfRow = 64 * 1024;
+void *  tsCacheObj = NULL;
+int32_t destroyTimes = 0;
+
+typedef int64_t CacheTestKey;
+typedef struct CacheTestRow {
+  int32_t index;
+  void ** ppRow;
+  void *  data;
+} CacheTestRow;
+
+CacheTestRow *initRow(int32_t index) {
+  CacheTestRow *row = calloc(sizeof(CacheTestRow), 1);
+  row->index = index;
+  row->data = malloc(tsSizeOfRow * sizeof(int8_t));
+  return row;
+}
+
+void detroyRow(void *data) {
+  CacheTestRow *row = *(CacheTestRow **)data;
+  free(row->data);
+  free(row);
+  destroyTimes++;
+  if (destroyTimes % 50000 == 0) {
+    pPrint("%s ===> destroyTimes:%d %s", GREEN, destroyTimes, NC);
+  }
+}
+
+void initCache() {
+  tsCacheObj = taosCacheInit(TSDB_DATA_TYPE_BIGINT, MAX_REFRESH_TIME_SEC, true, detroyRow, "cachetest");
+}
+
+void putRowInCache() {
+  for (int index = 0; index < tsNumOfRows; ++index) {
+    CacheTestRow *row = initRow(index);
+    uint64_t      key = (uint64_t)row;
+    void **ppRow = taosCachePut(tsCacheObj, &key, sizeof(int64_t), &row, sizeof(int64_t), tsKeepTimeInSec * 1000);
+    row->ppRow = ppRow;
+    taosCacheRelease(tsCacheObj, (void **)&ppRow, false);
+  }
+}
+
+void cleanupCache() {
+  taosCacheCleanup(tsCacheObj);
+}
+
+void initGetMemory() {
+  osInit();
+  taos_init();
+}
+
+float getProcMemory() {
+  float procMemoryUsedMB = 0;
+  taosGetProcMemory(&procMemoryUsedMB);
+  return procMemoryUsedMB;
+}
+
+void doTest() {
+  initCache();
+  pPrint("%s initialize procMemory %f MB %s", GREEN, getProcMemory(), NC);
+
+  putRowInCache();
+  pPrint("%s insert %d rows, procMemory %f MB %s", GREEN, tsNumOfRows, getProcMemory(), NC);
+
+  int32_t sleepMs = (MAX_REFRESH_TIME_SEC * 3) * 1000 + tsKeepTimeInSec * 1000;
+  taosMsleep(sleepMs);
+  pPrint("%s after sleep %d ms, procMemory %f MB %s", GREEN, sleepMs, getProcMemory(), NC);
+
+  cleanupCache();
+  taosMsleep(sleepMs);
+  pPrint("%s after cleanup cache, procMemory %f MB %s", GREEN, getProcMemory(), NC);
+  
+  malloc_trim(0);
+  taosMsleep(sleepMs);
+  pPrint("%s after malloc_trim, procMemory %f MB %s", GREEN, getProcMemory(), NC);
+}
+
+void printHelp() {
+  char indent[10] = "        ";
+  printf("Used to test the performance of cache\n");
+
+  printf("%s%s\n", indent, "-k");
+  printf("%s%s%s%d\n", indent, indent, "KeepTimeInSec, default is ", tsKeepTimeInSec);
+  printf("%s%s\n", indent, "-n");
+  printf("%s%s%s%d\n", indent, indent, "NumOfRows, default is ", tsNumOfRows);
+  printf("%s%s\n", indent, "-s");
+  printf("%s%s%s%d\n", indent, indent, "SizeOfData, default is ", tsSizeOfRow);
+
+  exit(EXIT_SUCCESS);
+}
+
+void parseArgument(int argc, char *argv[]) {
+  for (int i = 1; i < argc; i++) {
+    if (strcmp(argv[i], "-h") == 0 || strcmp(argv[i], "--help") == 0) {
+      printHelp();
+      exit(0);
+    } else if (strcmp(argv[i], "-k") == 0) {
+      tsKeepTimeInSec = atoi(argv[++i]);
+    } else if (strcmp(argv[i], "-n") == 0) {
+      tsNumOfRows = atoi(argv[++i]);
+    } else if (strcmp(argv[i], "-s") == 0) {
+      tsSizeOfRow = atoi(argv[++i]);
+    } else {
+    }
+  }
+
+  pPrint("%s KeepTimeInSec:%d %s", GREEN, tsKeepTimeInSec, NC);
+  pPrint("%s NumOfRows:%d %s", GREEN, tsNumOfRows, NC);
+  pPrint("%s SizeOfData:%d %s", GREEN, tsSizeOfRow, NC);
+}
+
+int main(int argc, char *argv[]) {
+  initGetMemory();
+  parseArgument(argc, argv);
+  doTest();
+}

tests/tsim/inc/sim.h

@@ -84,6 +84,7 @@ enum {
   SIM_CMD_SQL,
   SIM_CMD_SQL_ERROR,
   SIM_CMD_SQL_SLOW,
+  SIM_CMD_RESTFUL,
   SIM_CMD_TEST,
   SIM_CMD_RETURN,
   SIM_CMD_END
@@ -172,6 +173,7 @@ bool simExecuteReturnCmd(SScript *script, char *option);
 bool simExecuteSqlCmd(SScript *script, char *option);
 bool simExecuteSqlErrorCmd(SScript *script, char *rest);
 bool simExecuteSqlSlowCmd(SScript *script, char *option);
+bool simExecuteRestfulCmd(SScript *script, char *rest);
 void simVisuallizeOption(SScript *script, char *src, char *dst);
 
 #endif
\ No newline at end of file

tests/tsim/src/simExe.c

@@ -915,6 +915,47 @@ bool simExecuteSqlSlowCmd(SScript *script, char *rest) {
   return simExecuteSqlImpCmd(script, rest, isSlow);
 }
 
+bool simExecuteRestfulCmd(SScript *script, char *rest) {
+  FILE *fp = NULL;
+  char filename[256];
+  sprintf(filename, "%s/tmp.sql", tsScriptDir);  
+  fp = fopen(filename, "w");
+  if (fp == NULL) {
+    fprintf(stderr, "ERROR: failed to open file: %s\n", filename);
+    return false;
+  }
+
+  char db[64] = {0};
+  char tb[64] = {0};
+  char gzip[32] = {0};
+  int32_t ts;
+  int32_t times;
+  sscanf(rest, "%s %s %d %d %s", db, tb, &ts, &times, gzip);
+  
+  fprintf(fp, "insert into %s.%s values ", db, tb);
+  for (int i = 0; i < times; ++i) {
+    fprintf(fp, "(%d000, %d)", ts + i, ts);
+  }
+  fprintf(fp, "  \n");
+  fflush(fp);
+  fclose(fp);
+
+  char cmd[1024] = {0};
+  if (strcmp(gzip, "gzip") == 0) {
+    sprintf(cmd,
+            "curl -H 'Authorization: Taosd /KfeAzX/f9na8qdtNZmtONryp201ma04bEl8LcvLUd7a8qdtNZmtONryp201ma04' --header "
+            "--compressed --data-ascii @%s 127.0.0.1:7111/rest/sql",
+            filename);
+  } else {
+    sprintf(cmd,
+            "curl -H 'Authorization: Taosd /KfeAzX/f9na8qdtNZmtONryp201ma04bEl8LcvLUd7a8qdtNZmtONryp201ma04' --header "
+            "'Transfer-Encoding: chunked' --data-ascii @%s 127.0.0.1:7111/rest/sql",
+            filename);
+  }
+
+  return simExecuteSystemCmd(script, cmd);
+}
+
 bool simExecuteSqlErrorCmd(SScript *script, char *rest) {
   char buf[3000];
   SCmdLine *line = &script->lines[script->linePos];

tests/tsim/src/simParse.c

@@ -721,6 +721,12 @@ bool simParseSqlSlowCmd(char *rest, SCommand *pCmd, int lineNum) {
   return true;
 }
 
+bool simParseRestfulCmd(char *rest, SCommand *pCmd, int lineNum) {
+  simParseSqlCmd(rest, pCmd, lineNum);
+  cmdLine[numOfLines - 1].cmdno = SIM_CMD_RESTFUL;
+  return true;
+}
+
 bool simParseSystemCmd(char *rest, SCommand *pCmd, int lineNum) {
   int expLen;
 
@@ -1020,6 +1026,14 @@ void simInitsimCmdList() {
   simCmdList[cmdno].executeCmd = simExecuteSqlSlowCmd;
   simAddCmdIntoHash(&(simCmdList[cmdno]));
 
+  cmdno = SIM_CMD_RESTFUL;
+  simCmdList[cmdno].cmdno = cmdno;
+  strcpy(simCmdList[cmdno].name, "restful");
+  simCmdList[cmdno].nlen = (int16_t)strlen(simCmdList[cmdno].name);
+  simCmdList[cmdno].parseCmd = simParseRestfulCmd;
+  simCmdList[cmdno].executeCmd = simExecuteRestfulCmd;
+  simAddCmdIntoHash(&(simCmdList[cmdno]));
+
   /* test is only an internal command */
   cmdno = SIM_CMD_TEST;
   simCmdList[cmdno].cmdno = cmdno;