taosdemo.c 41.3 KB
Newer Older
H
hzcheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * 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/>.
 */

S
slguan 已提交
16
#define _GNU_SOURCE
H
hzcheng 已提交
17 18 19

#include <argp.h>
#include <assert.h>
L
lihui 已提交
20
#include <inttypes.h>
S
#1022  
slguan 已提交
21 22

#ifndef _ALPINE
H
hzcheng 已提交
23
#include <error.h>
F
Frozen 已提交
24
#endif
H
hzcheng 已提交
25 26 27 28 29 30 31 32 33 34
#include <pthread.h>
#include <semaphore.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include <time.h>
#include <unistd.h>
#include <wordexp.h>
H
Hui Li 已提交
35
#include <regex.h>
H
hzcheng 已提交
36 37

#include "taos.h"
H
Hui Li 已提交
38
#include "tutil.h"
S
slguan 已提交
39 40 41

extern char configDir[];

H
hzcheng 已提交
42 43 44
#define BUFFER_SIZE      65536
#define MAX_DB_NAME_SIZE 64
#define MAX_TB_NAME_SIZE 64
L
Liu Tao 已提交
45 46
#define MAX_DATA_SIZE    16000
#define MAX_NUM_DATATYPE 10
H
hzcheng 已提交
47
#define OPT_ABORT        1 /* –abort */
48
#define STRING_LEN       512
L
Liu Tao 已提交
49
#define MAX_PREPARED_RAND 1000000
H
hzcheng 已提交
50 51 52 53 54 55

/* The options we understand. */
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/IP port number to use for the connection. Default is 0.",                                                  1},
  {0, 'u', "user",                     0, "The TDEngine user name to use when connecting to the server. Default is 'root'.",                                  2},
S
Shuaiqiang Chang 已提交
56 57
  {0, 'P', "password",                 0, "The password to use when connecting to the server. Default is 'taosdata'.",                                        3},
  {0, 'd', "database",                 0, "Destination database. Default is 'test'.",                                                                         3},
H
hzcheng 已提交
58
  {0, 'm', "table_prefix",             0, "Table prefix name. Default is 't'.",                                                                               3},
H
Hui Li 已提交
59
  {0, 's', "sql file",                 0, "The select sql file.",                                                                                             3},
H
hzcheng 已提交
60 61 62 63 64
  {0, 'M', 0,                          0, "Use metric flag.",                                                                                                 13},
  {0, 'o', "outputfile",               0, "Direct output to the named file. Default is './output.txt'.",                                                      14},
  {0, 'q', "query_mode",               0, "Query mode--0: SYNC, 1: ASYNC. Default is SYNC.",                                                                  6},
  {0, 'b', "type_of_cols",             0, "The data_type of columns: 'INT', 'TINYINT', 'SMALLINT', 'BIGINT', 'FLOAT', 'DOUBLE', 'BINARY'. Default is 'INT'.", 7},
  {0, 'w', "length_of_binary",         0, "The length of data_type 'BINARY'. Only applicable when type of cols is 'BINARY'. Default is 8",                    8},
65
  {0, 'l', "num_of_cols_per_record",   0, "The number of columns per record. Default is 3.",                                                                  8},
S
Shuaiqiang Chang 已提交
66
  {0, 'T', "num_of_threads",           0, "The number of threads. Default is 10.",                                                                            9},
S
slguan 已提交
67 68 69
  {0, 'r', "num_of_records_per_req",   0, "The number of records per request. Default is 1000.",                                                              10},
  {0, 't', "num_of_tables",            0, "The number of tables. Default is 10000.",                                                                          11},
  {0, 'n', "num_of_records_per_table", 0, "The number of records per table. Default is 100000.",                                                              12},
S
Shuaiqiang Chang 已提交
70
  {0, 'c', "config_directory",         0, "Configuration directory. Default is '/etc/taos/'.",                                                                14},
H
hzcheng 已提交
71
  {0, 'x', 0,                          0, "Insert only flag.",                                                                                                13},
S
Shuaiqiang Chang 已提交
72 73 74
  {0, 'O', "order",                    0, "Insert mode--0: In order, 1: Out of order. Default is in order.",                                                  14},
  {0, 'R', "rate",                     0, "Out of order data's rate--if order=1 Default 10, min: 0, max: 50.",                                                14},
  {0, 'D', "delete table",             0, "Delete data methods——0: don't delete, 1: delete by table, 2: delete by stable, 3: delete by database",             14},
H
hzcheng 已提交
75 76 77
  {0}};

/* Used by main to communicate with parse_opt. */
78
typedef struct DemoArguments {
H
hzcheng 已提交
79
  char  *host;
L
lihui 已提交
80
  uint16_t    port;
H
hzcheng 已提交
81 82 83 84
  char  *user;
  char  *password;
  char  *database;
  char  *tb_prefix;
H
Hui Li 已提交
85
  char  *sqlFile;
H
hzcheng 已提交
86 87 88 89
  bool   use_metric;
  bool   insert_only;
  char  *output_file;
  int    mode;
H
Hui Li 已提交
90
  char  *datatype[MAX_NUM_DATATYPE+1];
H
hzcheng 已提交
91 92
  int    len_of_binary;
  int    num_of_CPR;
S
Shuaiqiang Chang 已提交
93
  int    num_of_threads;
H
hzcheng 已提交
94 95 96 97
  int    num_of_RPR;
  int    num_of_tables;
  int    num_of_DPT;
  int    abort;
S
Shuaiqiang Chang 已提交
98 99
  int    order;
  int    rate;
S
Shuaiqiang Chang 已提交
100
  int    method_of_delete;
H
hzcheng 已提交
101
  char **arg_list;
102
} SDemoArguments;
H
hzcheng 已提交
103 104 105 106 107

/* Parse a single option. */
static error_t parse_opt(int key, char *arg, struct argp_state *state) {
  /* Get the input argument from argp_parse, which we
     know is a pointer to our arguments structure. */
108
  SDemoArguments *arguments = state->input;
H
hzcheng 已提交
109 110 111 112 113 114 115 116 117 118 119 120
  wordexp_t full_path;
  char **sptr;
  switch (key) {
    case 'h':
      arguments->host = arg;
      break;
    case 'p':
      arguments->port = atoi(arg);
      break;
    case 'u':
      arguments->user = arg;
      break;
S
Shuaiqiang Chang 已提交
121
    case 'P':
H
hzcheng 已提交
122 123 124 125 126
      arguments->password = arg;
      break;
    case 'o':
      arguments->output_file = arg;
      break;
H
Hui Li 已提交
127 128 129
    case 's':
      arguments->sqlFile = arg;
      break;
H
hzcheng 已提交
130 131 132
    case 'q':
      arguments->mode = atoi(arg);
      break;
S
Shuaiqiang Chang 已提交
133 134
    case 'T':
      arguments->num_of_threads = atoi(arg);
H
hzcheng 已提交
135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157
      break;
    case 'r':
      arguments->num_of_RPR = atoi(arg);
      break;
    case 't':
      arguments->num_of_tables = atoi(arg);
      break;
    case 'n':
      arguments->num_of_DPT = atoi(arg);
      break;
    case 'd':
      arguments->database = arg;
      break;
    case 'l':
      arguments->num_of_CPR = atoi(arg);
      break;
    case 'b':
      sptr = arguments->datatype;
      if (strstr(arg, ",") == NULL) {
        if (strcasecmp(arg, "INT") != 0 && strcasecmp(arg, "FLOAT") != 0 &&
            strcasecmp(arg, "TINYINT") != 0 && strcasecmp(arg, "BOOL") != 0 &&
            strcasecmp(arg, "SMALLINT") != 0 &&
            strcasecmp(arg, "BIGINT") != 0 && strcasecmp(arg, "DOUBLE") != 0 &&
L
Liu Tao 已提交
158
            strcasecmp(arg, "BINARY") && strcasecmp(arg, "NCHAR")) {
H
hzcheng 已提交
159 160 161 162 163 164 165 166 167 168 169 170 171 172 173
          argp_error(state, "Invalid data_type!");
        }
        sptr[0] = arg;
      } else {
        int index = 0;
        char *dupstr = strdup(arg);
        char *running = dupstr;
        char *token = strsep(&running, ",");
        while (token != NULL) {
          if (strcasecmp(token, "INT") != 0 &&
              strcasecmp(token, "FLOAT") != 0 &&
              strcasecmp(token, "TINYINT") != 0 &&
              strcasecmp(token, "BOOL") != 0 &&
              strcasecmp(token, "SMALLINT") != 0 &&
              strcasecmp(token, "BIGINT") != 0 &&
L
Liu Tao 已提交
174
              strcasecmp(token, "DOUBLE") != 0 && strcasecmp(token, "BINARY") && strcasecmp(arg, "NCHAR")) {
H
hzcheng 已提交
175 176 177 178
            argp_error(state, "Invalid data_type!");
          }
          sptr[index++] = token;
          token = strsep(&running, ", ");
T
Tao Liu 已提交
179
          if (index >= MAX_NUM_DATATYPE) break;
H
hzcheng 已提交
180 181 182 183 184 185 186 187 188 189
        }
      }
      break;
    case 'w':
      arguments->len_of_binary = atoi(arg);
      break;
    case 'm':
      arguments->tb_prefix = arg;
      break;
    case 'M':
H
Hui Li 已提交
190
      arguments->use_metric = false;
H
hzcheng 已提交
191 192
      break;
    case 'x':
H
Hui Li 已提交
193
      arguments->insert_only = false;
H
hzcheng 已提交
194
      break;
S
Shuaiqiang Chang 已提交
195
    case 'c':
H
hzcheng 已提交
196 197 198 199
      if (wordexp(arg, &full_path, 0) != 0) {
        fprintf(stderr, "Invalid path %s\n", arg);
        return -1;
      }
200
      taos_options(TSDB_OPTION_CONFIGDIR, full_path.we_wordv[0]);
H
hzcheng 已提交
201 202
      wordfree(&full_path);
      break;
S
Shuaiqiang Chang 已提交
203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219
    case 'O':
      arguments->order = atoi(arg);
      if (arguments->order > 1 || arguments->order < 0)
      {
        arguments->order = 0;
      } else if (arguments->order == 1)
      {
        arguments->rate = 10;
      }
      break;
    case 'R':
      arguments->rate = atoi(arg);
      if (arguments->order == 1 && (arguments->rate > 50 || arguments->rate <= 0))
      {
        arguments->rate = 10;
      }
      break;
S
Shuaiqiang Chang 已提交
220 221 222 223 224 225 226
    case 'D':
      arguments->method_of_delete = atoi(arg);
      if (arguments->method_of_delete < 0 || arguments->method_of_delete > 3)
      {
        arguments->method_of_delete = 0;
      }
      break;
H
hzcheng 已提交
227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259
    case OPT_ABORT:
      arguments->abort = 1;
      break;
    case ARGP_KEY_ARG:
      /*arguments->arg_list = &state->argv[state->next-1];
      state->next = state->argc;*/
      argp_usage(state);
      break;

    default:
      return ARGP_ERR_UNKNOWN;
  }
  return 0;
}

/* ******************************* Structure
 * definition*******************************  */
enum MODE {
  SYNC, ASYNC
};
typedef struct {
  TAOS *taos;
  int threadID;
  char db_name[MAX_DB_NAME_SIZE];
  char fp[4096];
  char **datatype;
  int len_of_binary;
  char tb_prefix[MAX_TB_NAME_SIZE];
  int start_table_id;
  int end_table_id;
  int ncols_per_record;
  int nrecords_per_table;
  int nrecords_per_request;
S
Shuaiqiang Chang 已提交
260 261
  int data_of_order;
  int data_of_rate;
S
slguan 已提交
262
  int64_t start_time;
H
hzcheng 已提交
263
  bool do_aggreFunc;
H
Hui Li 已提交
264 265 266
  
  char* cols;  
  bool  use_metric;
H
hzcheng 已提交
267 268 269 270 271 272 273 274 275 276

  sem_t mutex_sem;
  int notFinished;
  sem_t lock_sem;
} info;

typedef struct {
  TAOS  *taos;

  char   tb_name[MAX_TB_NAME_SIZE];
S
slguan 已提交
277
  int64_t   timestamp;
H
hzcheng 已提交
278 279 280 281 282 283
  int    target;
  int    counter;
  int    nrecords_per_request;
  int    ncols_per_record;
  char **data_type;
  int    len_of_binary;
S
Shuaiqiang Chang 已提交
284 285
  int data_of_order;
  int data_of_rate;
H
hzcheng 已提交
286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307

  sem_t *mutex_sem;
  int   *notFinished;
  sem_t *lock_sem;
} sTable;

/* ******************************* Global
 * variables*******************************  */
char *aggreFunc[] = {"*", "count(*)", "avg(f1)", "sum(f1)", "max(f1)", "min(f1)", "first(f1)", "last(f1)"};

/* ******************************* Global
 * functions*******************************  */
static struct argp argp = {options, parse_opt, 0, 0};

void queryDB(TAOS *taos, char *command);

void *readTable(void *sarg);

void *readMetric(void *sarg);

void *syncWrite(void *sarg);

S
Shuaiqiang Chang 已提交
308 309
void *deleteTable();

H
hzcheng 已提交
310 311
void *asyncWrite(void *sarg);

312
int generateData(char *res, char **data_type, int num_of_cols, int64_t timestamp, int len_of_binary);
H
hzcheng 已提交
313 314 315

void rand_string(char *str, int size);

L
Liu Tao 已提交
316 317
void init_rand_data();

H
hzcheng 已提交
318 319 320
double getCurrentTime();

void callBack(void *param, TAOS_RES *res, int code);
H
Hui Li 已提交
321 322
void multiThreadCreateTable(char* cols, bool use_metric, int threads, int ntables, char* db_name, char* tb_prefix, char *ip_addr, uint16_t port, char *user, char *pass);
void querySqlFile(TAOS* taos, char* sqlFile);
H
hzcheng 已提交
323 324

int main(int argc, char *argv[]) {
325
  SDemoArguments arguments = {  NULL,            // host
H
Hongze Cheng 已提交
326 327 328 329 330
                                0,               // port
                                "root",          // user
                                "taosdata",      // password
                                "test",          // database
                                "t",             // tb_prefix
H
Hui Li 已提交
331
                                NULL,
H
Hongze Cheng 已提交
332 333 334 335 336 337
                                false,           // use_metric
                                false,           // insert_only
                                "./output.txt",  // output_file
                                0,               // mode
                                {
                                "int",           // datatype
H
hzcheng 已提交
338 339 340 341 342 343
                                "",
                                "",
                                "",
                                "",
                                "",
                                "",
H
Hongze Cheng 已提交
344 345 346 347
                                ""
                                },
                                8,               // len_of_binary
                                1,               // num_of_CPR
S
Shuaiqiang Chang 已提交
348
                                1,               // num_of_connections/thread
H
Hongze Cheng 已提交
349 350 351 352
                                1,               // num_of_RPR
                                1,               // num_of_tables
                                50000,           // num_of_DPT
                                0,               // abort
S
Shuaiqiang Chang 已提交
353 354
                                0,               // order
                                0,               // rate
S
Shuaiqiang Chang 已提交
355
                                0,               // method_of_delete
H
Hongze Cheng 已提交
356 357
                                NULL             // arg_list
                                };
H
hzcheng 已提交
358 359 360

  /* Parse our arguments; every option seen by parse_opt will be
     reflected in arguments. */
361
  // For demo use, change default values for some parameters;
362
  arguments.num_of_tables = 10000;
363
  arguments.num_of_CPR = 3; 
S
Shuaiqiang Chang 已提交
364
  arguments.num_of_threads = 10;
365 366
  arguments.num_of_DPT = 100000;
  arguments.num_of_RPR = 1000;
367
  arguments.use_metric = true;
T
Tao Liu 已提交
368
  arguments.insert_only = true;
S
slguan 已提交
369
  // end change
370

H
hzcheng 已提交
371 372
  argp_parse(&argp, argc, argv, 0, 0, &arguments);

F
Frozen 已提交
373
  if (arguments.abort) {
S
#1022  
slguan 已提交
374
    #ifndef _ALPINE
F
Frozen 已提交
375 376 377 378 379
      error(10, 0, "ABORTED");
    #else
      abort();
    #endif
  }
H
Hui Li 已提交
380

H
hzcheng 已提交
381 382
  enum MODE query_mode = arguments.mode;
  char *ip_addr = arguments.host;
L
lihui 已提交
383
  uint16_t port = arguments.port;
H
hzcheng 已提交
384 385 386 387 388 389
  char *user = arguments.user;
  char *pass = arguments.password;
  char *db_name = arguments.database;
  char *tb_prefix = arguments.tb_prefix;
  int len_of_binary = arguments.len_of_binary;
  int ncols_per_record = arguments.num_of_CPR;
S
Shuaiqiang Chang 已提交
390 391
  int order = arguments.order;
  int rate = arguments.rate;
S
Shuaiqiang Chang 已提交
392
  int method_of_delete = arguments.method_of_delete;
H
hzcheng 已提交
393
  int ntables = arguments.num_of_tables;
S
Shuaiqiang Chang 已提交
394
  int threads = arguments.num_of_threads;
H
hzcheng 已提交
395 396 397 398 399 400
  int nrecords_per_table = arguments.num_of_DPT;
  int nrecords_per_request = arguments.num_of_RPR;
  bool use_metric = arguments.use_metric;
  bool insert_only = arguments.insert_only;
  char **data_type = arguments.datatype;
  int count_data_type = 0;
401
  char dataString[STRING_LEN];
H
hzcheng 已提交
402
  bool do_aggreFunc = true;
403

H
Hui Li 已提交
404 405 406 407 408 409
  if (NULL != arguments.sqlFile) {
    TAOS* qtaos = taos_connect(ip_addr, user, pass, db_name, port);
    querySqlFile(qtaos, arguments.sqlFile);  
    taos_close(qtaos);
    return 0;
  }
L
Liu Tao 已提交
410
  init_rand_data();
H
Hui Li 已提交
411

412 413
  memset(dataString, 0, STRING_LEN);
  int len = 0;
414

L
Liu Tao 已提交
415
  if (strcasecmp(data_type[0], "BINARY") == 0 || strcasecmp(data_type[0], "BOOL") == 0 || strcasecmp(data_type[0], "NCHAR") == 0 ) {
H
hzcheng 已提交
416 417 418
    do_aggreFunc = false;
  }
  for (; count_data_type <= MAX_NUM_DATATYPE; count_data_type++) {
T
Tao Liu 已提交
419
    if (data_type[count_data_type] == NULL) {
H
hzcheng 已提交
420 421
      break;
    }
422 423

    len += snprintf(dataString + len, STRING_LEN - len, "%s ", data_type[count_data_type]);
H
hzcheng 已提交
424 425 426
  }

  FILE *fp = fopen(arguments.output_file, "a");
427 428 429 430 431
  if (NULL == fp) {
    fprintf(stderr, "Failed to open %s for writing\n", arguments.output_file);
    return 1;
  };
  
H
hzcheng 已提交
432 433
  time_t tTime = time(NULL);
  struct tm tm = *localtime(&tTime);
S
Shuaiqiang Chang 已提交
434 435 436 437 438 439 440
  printf("###################################################################\n");
  printf("# Server IP:                         %s:%hu\n", ip_addr == NULL ? "localhost" : ip_addr, port);
  printf("# User:                              %s\n", user);
  printf("# Password:                          %s\n", pass);
  printf("# Use metric:                        %s\n", use_metric ? "true" : "false");
  printf("# Datatype of Columns:               %s\n", dataString);
  printf("# Binary Length(If applicable):      %d\n",
L
Liu Tao 已提交
441
          (strcasestr(dataString, "BINARY") != NULL || strcasestr(dataString, "NCHAR") != NULL ) ? len_of_binary : -1);
S
Shuaiqiang Chang 已提交
442
  printf("# Number of Columns per record:      %d\n", ncols_per_record);
S
Shuaiqiang Chang 已提交
443
  printf("# Number of Threads:                 %d\n", threads);
S
Shuaiqiang Chang 已提交
444 445 446 447 448
  printf("# Number of Tables:                  %d\n", ntables);
  printf("# Number of Data per Table:          %d\n", nrecords_per_table);
  printf("# Records/Request:                   %d\n", nrecords_per_request);
  printf("# Database name:                     %s\n", db_name);
  printf("# Table prefix:                      %s\n", tb_prefix);
S
Shuaiqiang Chang 已提交
449 450
  if (order == 1)
  {
S
Shuaiqiang Chang 已提交
451 452
    printf("# Data order:                        %d\n", order);
    printf("# Data out of order rate:            %d\n", rate);
S
Shuaiqiang Chang 已提交
453 454

  }
S
Shuaiqiang Chang 已提交
455
  printf("# Delete method:                     %d\n", method_of_delete);
S
Shuaiqiang Chang 已提交
456 457 458 459
  printf("# Test time:                         %d-%02d-%02d %02d:%02d:%02d\n", tm.tm_year + 1900, tm.tm_mon + 1,
          tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec);
  printf("###################################################################\n\n");
  printf("Press enter key to continue");
H
Hui Li 已提交
460
  (void)getchar();
H
hzcheng 已提交
461 462

  fprintf(fp, "###################################################################\n");
L
lihui 已提交
463
  fprintf(fp, "# Server IP:                         %s:%hu\n", ip_addr == NULL ? "localhost" : ip_addr, port);
H
hzcheng 已提交
464 465 466 467 468
  fprintf(fp, "# User:                              %s\n", user);
  fprintf(fp, "# Password:                          %s\n", pass);
  fprintf(fp, "# Use metric:                        %s\n", use_metric ? "true" : "false");
  fprintf(fp, "# Datatype of Columns:               %s\n", dataString);
  fprintf(fp, "# Binary Length(If applicable):      %d\n",
L
Liu Tao 已提交
469
          (strcasestr(dataString, "BINARY") != NULL || strcasestr(dataString, "NCHAR") != NULL ) ? len_of_binary : -1);
H
hzcheng 已提交
470
  fprintf(fp, "# Number of Columns per record:      %d\n", ncols_per_record);
S
Shuaiqiang Chang 已提交
471
  fprintf(fp, "# Number of Threads:                 %d\n", threads);
H
hzcheng 已提交
472 473 474 475 476
  fprintf(fp, "# Number of Tables:                  %d\n", ntables);
  fprintf(fp, "# Number of Data per Table:          %d\n", nrecords_per_table);
  fprintf(fp, "# Records/Request:                   %d\n", nrecords_per_request);
  fprintf(fp, "# Database name:                     %s\n", db_name);
  fprintf(fp, "# Table prefix:                      %s\n", tb_prefix);
S
Shuaiqiang Chang 已提交
477 478 479 480 481 482
  if (order == 1)
  {
    printf("# Data order:                        %d\n", order);
    printf("# Data out of order rate:            %d\n", rate);

  }
H
hzcheng 已提交
483 484 485 486 487 488 489 490
  fprintf(fp, "# Test time:                         %d-%02d-%02d %02d:%02d:%02d\n", tm.tm_year + 1900, tm.tm_mon + 1,
          tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec);
  fprintf(fp, "###################################################################\n\n");
  fprintf(fp, "|  WRecords  | Records/Second | Requests/Second |  WLatency(ms) |\n");

  taos_init();
  TAOS *taos = taos_connect(ip_addr, user, pass, NULL, port);
  if (taos == NULL) {
S
Shuaiqiang Chang 已提交
491
    fprintf(stderr, "Failed to connect to TDengine, reason:%s\n", taos_errstr(NULL));
H
hzcheng 已提交
492 493 494 495 496
    return 1;
  }
  char command[BUFFER_SIZE] = "\0";

  sprintf(command, "drop database %s;", db_name);
497 498
  TAOS_RES* res = taos_query(taos, command);
  taos_free_result(res);
H
hzcheng 已提交
499 500

  sprintf(command, "create database %s;", db_name);
501 502
  res = taos_query(taos, command);
  taos_free_result(res);
H
hzcheng 已提交
503

504
  char cols[STRING_LEN] = "\0";
H
hzcheng 已提交
505
  int colIndex = 0;
506
  len = 0;
H
hzcheng 已提交
507 508

  for (; colIndex < ncols_per_record - 1; colIndex++) {
L
Liu Tao 已提交
509
    if (strcasecmp(data_type[colIndex % count_data_type], "BINARY") != 0 && strcasecmp(data_type[colIndex % count_data_type], "NCHAR") != 0) {
510
      len += snprintf(cols + len, STRING_LEN - len, ",f%d %s", colIndex + 1, data_type[colIndex % count_data_type]);
H
hzcheng 已提交
511
    } else {
512
      len += snprintf(cols + len, STRING_LEN - len, ",f%d %s(%d)", colIndex + 1, data_type[colIndex % count_data_type], len_of_binary);
H
hzcheng 已提交
513 514 515 516
    }
  }

  if (strcasecmp(data_type[colIndex % count_data_type], "BINARY") != 0) {
517
    len += snprintf(cols + len, STRING_LEN - len, ",f%d %s)", colIndex + 1, data_type[colIndex % count_data_type]);
H
hzcheng 已提交
518
  } else {
519
    len += snprintf(cols + len, STRING_LEN - len, ",f%d %s(%d))", colIndex + 1, data_type[colIndex % count_data_type], len_of_binary);
H
hzcheng 已提交
520 521
  }

H
Hui Li 已提交
522
  if (use_metric) {
H
hzcheng 已提交
523
    /* Create metric table */
524
    printf("Creating meters super table...\n");
L
Liu Tao 已提交
525
    snprintf(command, BUFFER_SIZE, "create table if not exists %s.meters (ts timestamp%s) tags (areaid int, loc binary(10)", db_name, cols);
H
hzcheng 已提交
526
    queryDB(taos, command);
527
    printf("meters created!\n");
H
hzcheng 已提交
528
  }
H
Hui Li 已提交
529
  taos_close(taos);
H
Hui Li 已提交
530
  
H
hzcheng 已提交
531
  /* Wait for table to create  */
H
Hui Li 已提交
532
  multiThreadCreateTable(cols, use_metric, threads, ntables, db_name, tb_prefix, ip_addr, port, user, pass);
S
Shuaiqiang Chang 已提交
533
  
H
hzcheng 已提交
534 535 536
  /* Insert data */
  double ts = getCurrentTime();
  printf("Inserting data......\n");
S
Shuaiqiang Chang 已提交
537 538
  pthread_t *pids = malloc(threads * sizeof(pthread_t));
  info *infos = malloc(threads * sizeof(info));
H
hzcheng 已提交
539

S
Shuaiqiang Chang 已提交
540
  int a = ntables / threads;
H
hzcheng 已提交
541
  if (a < 1) {
S
Shuaiqiang Chang 已提交
542
    threads = ntables;
H
hzcheng 已提交
543 544
    a = 1;
  }
S
Shuduo Sang 已提交
545 546 547 548

  int b = 0;
  if (threads != 0)
    b = ntables % threads;
H
hzcheng 已提交
549
  int last = 0;
S
Shuaiqiang Chang 已提交
550
  for (int i = 0; i < threads; i++) {
H
hzcheng 已提交
551 552
    info *t_info = infos + i;
    t_info->threadID = i;
H
Hui Li 已提交
553 554
    tstrncpy(t_info->db_name, db_name, MAX_DB_NAME_SIZE);
    tstrncpy(t_info->tb_prefix, tb_prefix, MAX_TB_NAME_SIZE);
H
hzcheng 已提交
555 556 557 558 559 560 561 562
    t_info->datatype = data_type;
    t_info->ncols_per_record = ncols_per_record;
    t_info->nrecords_per_table = nrecords_per_table;
    t_info->start_time = 1500000000000;
    t_info->taos = taos_connect(ip_addr, user, pass, db_name, port);
    t_info->len_of_binary = len_of_binary;
    t_info->nrecords_per_request = nrecords_per_request;
    t_info->start_table_id = last;
S
Shuaiqiang Chang 已提交
563 564
    t_info->data_of_order = order;
    t_info->data_of_rate = rate;
H
hzcheng 已提交
565 566 567 568 569 570 571 572 573 574 575 576 577
    t_info->end_table_id = i < b ? last + a : last + a - 1;
    last = t_info->end_table_id + 1;

    sem_init(&(t_info->mutex_sem), 0, 1);
    t_info->notFinished = t_info->end_table_id - t_info->start_table_id + 1;
    sem_init(&(t_info->lock_sem), 0, 0);

    if (query_mode == SYNC) {
      pthread_create(pids + i, NULL, syncWrite, t_info);
    } else {
      pthread_create(pids + i, NULL, asyncWrite, t_info);
    }
  }
S
Shuaiqiang Chang 已提交
578
  for (int i = 0; i < threads; i++) {
H
hzcheng 已提交
579 580 581 582 583
    pthread_join(pids[i], NULL);
  }

  double t = getCurrentTime() - ts;
  if (query_mode == SYNC) {
S
Shuaiqiang Chang 已提交
584
    printf("SYNC Insert with %d connections:\n", threads);
H
hzcheng 已提交
585
  } else {
S
Shuaiqiang Chang 已提交
586
    printf("ASYNC Insert with %d connections:\n", threads);
H
hzcheng 已提交
587 588 589 590 591 592 593 594 595 596 597
  }

  fprintf(fp, "|%10.d  |  %10.2f    |  %10.2f     |  %10.4f   |\n\n",
          ntables * nrecords_per_table, ntables * nrecords_per_table / t,
          (ntables * nrecords_per_table) / (t * nrecords_per_request),
          t * 1000);

  printf("Spent %.4f seconds to insert %d records with %d record(s) per request: %.2f records/second\n",
         t, ntables * nrecords_per_table, nrecords_per_request,
         ntables * nrecords_per_table / t);

S
Shuaiqiang Chang 已提交
598
  for (int i = 0; i < threads; i++) {
H
hzcheng 已提交
599 600 601 602 603 604 605 606 607 608
    info *t_info = infos + i;
    taos_close(t_info->taos);
    sem_destroy(&(t_info->mutex_sem));
    sem_destroy(&(t_info->lock_sem));
  }

  free(pids);
  free(infos);
  fclose(fp);

S
Shuaiqiang Chang 已提交
609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657
  if (method_of_delete != 0)
  {
    TAOS *dtaos = taos_connect(ip_addr, user, pass, db_name, port);
    double dts = getCurrentTime();
    printf("Deleteing %d table(s)......\n", ntables);

    switch (method_of_delete)
    {
    case 1:
      // delete by table
      /* Create all the tables; */
      for (int i = 0; i < ntables; i++) {
        sprintf(command, "drop table %s.%s%d;", db_name, tb_prefix, i);
        queryDB(dtaos, command);
      }
      break;
    case 2:
      // delete by stable
      if (!use_metric) {
        break;
      }
      else
      {
        sprintf(command, "drop table %s.meters;", db_name);
        queryDB(dtaos, command);
      }
      break;
    case 3:
      // delete by database
      sprintf(command, "drop database %s;", db_name);
      queryDB(dtaos, command);
      break;
    default:
      break;
    }

    printf("Table(s) droped!\n");
    taos_close(dtaos);

    double dt = getCurrentTime() - dts;
    printf("Spent %.4f seconds to drop %d tables\n", dt, ntables);

    FILE *fp = fopen(arguments.output_file, "a");
    fprintf(fp, "Spent %.4f seconds to drop %d tables\n", dt, ntables);
    fclose(fp);

  }
  

H
hzcheng 已提交
658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682
  if (!insert_only) {
    // query data
    pthread_t read_id;
    info *rInfo = malloc(sizeof(info));
    rInfo->start_time = 1500000000000;
    rInfo->start_table_id = 0;
    rInfo->end_table_id = ntables - 1;
    rInfo->do_aggreFunc = do_aggreFunc;
    rInfo->nrecords_per_table = nrecords_per_table;
    rInfo->taos = taos_connect(ip_addr, user, pass, db_name, port);
    strcpy(rInfo->tb_prefix, tb_prefix);
    strcpy(rInfo->fp, arguments.output_file);

    if (!use_metric) {
      pthread_create(&read_id, NULL, readTable, rInfo);
    } else {
      pthread_create(&read_id, NULL, readMetric, rInfo);
    }
    pthread_join(read_id, NULL);
    taos_close(rInfo->taos);
  }

  return 0;
}

H
Hui Li 已提交
683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869
#define MAX_SQL_SIZE       65536
void selectSql(TAOS* taos, char* sqlcmd)
{    
  TAOS_RES *pSql = taos_query(taos, sqlcmd);
  int32_t code = taos_errno(pSql);
  
  if (code != 0) {
    printf("Failed to sqlcmd:%s, reason:%s\n", sqlcmd, taos_errstr(pSql));
    taos_free_result(pSql);
    exit(1);
  }
  
  int count = 0;
  while (taos_fetch_row(pSql) != NULL) {
    count++;
  }
  
  taos_free_result(pSql);
  return;
}


/* Function to do regular expression check */
static int regexMatch(const char *s, const char *reg, int cflags) {
  regex_t regex;
  char    msgbuf[100] = {0};

  /* Compile regular expression */
  if (regcomp(&regex, reg, cflags) != 0) {
    printf("Fail to compile regex\n");
    exit(-1);
  }

  /* Execute regular expression */
  int reti = regexec(&regex, s, 0, NULL, 0);
  if (!reti) {
    regfree(&regex);
    return 1;
  } else if (reti == REG_NOMATCH) {
    regfree(&regex);
    return 0;
  } else {
    regerror(reti, &regex, msgbuf, sizeof(msgbuf));
    printf("Regex match failed: %s\n", msgbuf);
    regfree(&regex);
    exit(-1);
  }

  return 0;
}

static int isCommentLine(char *line) {
  if (line == NULL) return 1;

  return regexMatch(line, "^\\s*#.*", REG_EXTENDED);
}

void querySqlFile(TAOS* taos, char* sqlFile)
{
  FILE *fp = fopen(sqlFile, "r");
  if (fp == NULL) {
    printf("failed to open file %s, reason:%s\n", sqlFile, strerror(errno));
    exit(-1);
  }
  
  int       read_len = 0;
  char *    cmd = calloc(1, MAX_SQL_SIZE);
  size_t    cmd_len = 0;
  char *    line = NULL;
  size_t    line_len = 0;

  double t = getCurrentTime();
  
  while ((read_len = getline(&line, &line_len, fp)) != -1) {
    if (read_len >= MAX_SQL_SIZE) continue;
    line[--read_len] = '\0';

    if (read_len == 0 || isCommentLine(line)) {  // line starts with #
      continue;
    }

    if (line[read_len - 1] == '\\') {
      line[read_len - 1] = ' ';
      memcpy(cmd + cmd_len, line, read_len);
      cmd_len += read_len;
      continue;
    }

    memcpy(cmd + cmd_len, line, read_len);
    selectSql(taos, cmd);    
    memset(cmd, 0, MAX_SQL_SIZE);
    cmd_len = 0;
  }

  t = getCurrentTime() - t;
  printf("run %s took %.6f second(s)\n\n", sqlFile, t);

  free(cmd);
  if (line) free(line);
  fclose(fp);
  return;
}

void * createTable(void *sarg) 
{
  char command[BUFFER_SIZE] = "\0";
  
  info *winfo = (info *)sarg;  

  if (!winfo->use_metric) {
    /* Create all the tables; */
    printf("Creating table from %d to %d\n", winfo->start_table_id, winfo->end_table_id);
    for (int i = winfo->start_table_id; i <= winfo->end_table_id; i++) {
      snprintf(command, BUFFER_SIZE, "create table if not exists %s.%s%d (ts timestamp%s;", winfo->db_name, winfo->tb_prefix, i, winfo->cols);
      queryDB(winfo->taos, command);
    }
  } else {
    /* Create all the tables; */
    printf("Creating table from %d to %d\n", winfo->start_table_id, winfo->end_table_id);
    for (int i = winfo->start_table_id; i <= winfo->end_table_id; i++) {
      int j;
      if (i % 10 == 0) {
        j = 10;
      } else {
        j = i % 10;
      }
    if (j % 2 == 0) {
      snprintf(command, BUFFER_SIZE, "create table if not exists %s.%s%d using %s.meters tags (%d,\"%s\");", winfo->db_name, winfo->tb_prefix, i, winfo->db_name, j, "shanghai");
    } else {
      snprintf(command, BUFFER_SIZE, "create table if not exists %s.%s%d using %s.meters tags (%d,\"%s\");", winfo->db_name, winfo->tb_prefix, i, winfo->db_name, j, "beijing");
    }
      queryDB(winfo->taos, command);
    }
  }

  return NULL;
}

void multiThreadCreateTable(char* cols, bool use_metric, int threads, int ntables, char* db_name, char* tb_prefix, char *ip_addr, uint16_t port, char *user, char *pass) {
  double ts = getCurrentTime();
  printf("create table......\n");
  pthread_t *pids = malloc(threads * sizeof(pthread_t));
  info *infos = malloc(threads * sizeof(info));

  int a = ntables / threads;
  if (a < 1) {
    threads = ntables;
    a = 1;
  }

  int b = 0;
  if (threads != 0)
    b = ntables % threads;
  int last = 0;
  for (int i = 0; i < threads; i++) {
    info *t_info = infos + i;
    t_info->threadID = i;
    tstrncpy(t_info->db_name, db_name, MAX_DB_NAME_SIZE);
    tstrncpy(t_info->tb_prefix, tb_prefix, MAX_TB_NAME_SIZE);
    t_info->taos = taos_connect(ip_addr, user, pass, db_name, port);
    t_info->start_table_id = last;
    t_info->end_table_id = i < b ? last + a : last + a - 1;
    last = t_info->end_table_id + 1;
    t_info->use_metric = use_metric;
    t_info->cols = cols;
    pthread_create(pids + i, NULL, createTable, t_info);
  }
  
  for (int i = 0; i < threads; i++) {
    pthread_join(pids[i], NULL);
  }

  double t = getCurrentTime() - ts;
  printf("Spent %.4f seconds to create %d tables with %d connections\n", t, ntables, threads);

  for (int i = 0; i < threads; i++) {
    info *t_info = infos + i;
    sem_destroy(&(t_info->mutex_sem));
    sem_destroy(&(t_info->lock_sem));
  }

  free(pids);
  free(infos);  

  return ;
}

H
hzcheng 已提交
870 871 872 873
void *readTable(void *sarg) {
  info *rinfo = (info *)sarg;
  TAOS *taos = rinfo->taos;
  char command[BUFFER_SIZE] = "\0";
S
slguan 已提交
874
  int64_t sTime = rinfo->start_time;
H
hzcheng 已提交
875 876
  char *tb_prefix = rinfo->tb_prefix;
  FILE *fp = fopen(rinfo->fp, "a");
H
Hui Li 已提交
877 878 879 880 881
  if (NULL == fp) {
    printf("fopen %s fail, reason:%s.\n", rinfo->fp, strerror(errno));
    return NULL;
  }
  
H
hzcheng 已提交
882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897
  int num_of_DPT = rinfo->nrecords_per_table;
  int num_of_tables = rinfo->end_table_id - rinfo->start_table_id + 1;
  int totalData = num_of_DPT * num_of_tables;
  bool do_aggreFunc = rinfo->do_aggreFunc;

  int n = do_aggreFunc ? (sizeof(aggreFunc) / sizeof(aggreFunc[0])) : 2;
  if (!do_aggreFunc) {
    printf("\nThe first field is either Binary or Bool. Aggregation functions are not supported.\n");
  }
  printf("%d records:\n", totalData);
  fprintf(fp, "| QFunctions |    QRecords    |   QSpeed(R/s)   |  QLatency(ms) |\n");

  for (int j = 0; j < n; j++) {
    double totalT = 0;
    int count = 0;
    for (int i = 0; i < num_of_tables; i++) {
L
lihui 已提交
898
      sprintf(command, "select %s from %s%d where ts>= %" PRId64, aggreFunc[j], tb_prefix, i, sTime);
H
hzcheng 已提交
899 900

      double t = getCurrentTime();
S
Shuaiqiang Chang 已提交
901 902
      TAOS_RES *pSql = taos_query(taos, command);
      int32_t code = taos_errno(pSql);
H
hzcheng 已提交
903

S
Shuaiqiang Chang 已提交
904
      if (code != 0) {
H
Haojun Liao 已提交
905
        fprintf(stderr, "Failed to query:%s\n", taos_errstr(pSql));
S
Shuaiqiang Chang 已提交
906
        taos_free_result(pSql);
H
hzcheng 已提交
907
        taos_close(taos);
S
Shuaiqiang Chang 已提交
908
        exit(EXIT_FAILURE);
H
hzcheng 已提交
909 910
      }

S
Shuaiqiang Chang 已提交
911
      while (taos_fetch_row(pSql) != NULL) {
H
hzcheng 已提交
912 913 914 915 916 917
        count++;
      }

      t = getCurrentTime() - t;
      totalT += t;

S
Shuaiqiang Chang 已提交
918
      taos_free_result(pSql);
H
hzcheng 已提交
919 920 921
    }

    fprintf(fp, "|%10s  |   %10d   |  %12.2f   |   %10.2f  |\n",
S
slguan 已提交
922
            aggreFunc[j][0] == '*' ? "   *   " : aggreFunc[j], totalData,
H
hzcheng 已提交
923 924 925 926 927 928 929 930 931 932 933 934 935 936
            (double)(num_of_tables * num_of_DPT) / totalT, totalT * 1000);
    printf("select %10s took %.6f second(s)\n", aggreFunc[j], totalT);
  }
  fprintf(fp, "\n");

  fclose(fp);
  return NULL;
}

void *readMetric(void *sarg) {
  info *rinfo = (info *)sarg;
  TAOS *taos = rinfo->taos;
  char command[BUFFER_SIZE] = "\0";
  FILE *fp = fopen(rinfo->fp, "a");
H
Hui Li 已提交
937 938 939 940 941
  if (NULL == fp) {
    printf("fopen %s fail, reason:%s.\n", rinfo->fp, strerror(errno));
    return NULL;
  }
  
H
hzcheng 已提交
942 943 944 945 946 947 948 949 950 951 952 953 954
  int num_of_DPT = rinfo->nrecords_per_table;
  int num_of_tables = rinfo->end_table_id - rinfo->start_table_id + 1;
  int totalData = num_of_DPT * num_of_tables;
  bool do_aggreFunc = rinfo->do_aggreFunc;

  int n = do_aggreFunc ? (sizeof(aggreFunc) / sizeof(aggreFunc[0])) : 2;
  if (!do_aggreFunc) {
    printf("\nThe first field is either Binary or Bool. Aggregation functions are not supported.\n");
  }
  printf("%d records:\n", totalData);
  fprintf(fp, "Querying On %d records:\n", totalData);

  for (int j = 0; j < n; j++) {
sangshuduo's avatar
sangshuduo 已提交
955
    char condition[BUFFER_SIZE - 30] = "\0";
B
Bomin Zhang 已提交
956
    char tempS[64] = "\0";
H
hzcheng 已提交
957 958 959 960 961

    int m = 10 < num_of_tables ? 10 : num_of_tables;

    for (int i = 1; i <= m; i++) {
      if (i == 1) {
L
liu0x54 已提交
962
        sprintf(tempS, "areaid = %d", i);
H
hzcheng 已提交
963
      } else {
L
liu0x54 已提交
964
        sprintf(tempS, " or areaid = %d ", i);
H
hzcheng 已提交
965 966 967
      }
      strcat(condition, tempS);

L
liu0x54 已提交
968
      sprintf(command, "select %s from meters where %s", aggreFunc[j], condition);
H
hzcheng 已提交
969 970 971 972 973 974

      printf("Where condition: %s\n", condition);
      fprintf(fp, "%s\n", command);

      double t = getCurrentTime();

S
Shuaiqiang Chang 已提交
975 976 977 978
      TAOS_RES *pSql = taos_query(taos, command);
      int32_t code = taos_errno(pSql);

      if (code != 0) {
H
Haojun Liao 已提交
979
        fprintf(stderr, "Failed to query:%s\n", taos_errstr(pSql));
S
Shuaiqiang Chang 已提交
980
        taos_free_result(pSql);
H
hzcheng 已提交
981 982 983 984
        taos_close(taos);
        exit(1);
      }
      int count = 0;
S
Shuaiqiang Chang 已提交
985
      while (taos_fetch_row(pSql) != NULL) {
H
hzcheng 已提交
986 987 988 989 990 991 992
        count++;
      }
      t = getCurrentTime() - t;

      fprintf(fp, "| Speed: %12.2f(per s) | Latency: %.4f(ms) |\n", num_of_tables * num_of_DPT / t, t * 1000);
      printf("select %10s took %.6f second(s)\n\n", aggreFunc[j], t);

S
Shuaiqiang Chang 已提交
993
      taos_free_result(pSql);
H
hzcheng 已提交
994 995 996 997 998 999 1000 1001 1002
    }
    fprintf(fp, "\n");
  }

  fclose(fp);
  return NULL;
}

void queryDB(TAOS *taos, char *command) {
H
Hui Li 已提交
1003
  int i;
S
Shuaiqiang Chang 已提交
1004
  TAOS_RES *pSql = NULL;
H
Hui Li 已提交
1005 1006 1007 1008 1009 1010 1011 1012
  int32_t   code = -1;

  for (i = 0; i < 5; i++) {
    if (NULL != pSql) {
      taos_free_result(pSql);
      pSql = NULL;
    }
    
S
Shuaiqiang Chang 已提交
1013 1014
    pSql = taos_query(taos, command);
    code = taos_errno(pSql);
H
Hui Li 已提交
1015
    if (0 == code) {
S
Shuaiqiang Chang 已提交
1016
      break;
H
Hui Li 已提交
1017
    }    
1018
  }
S
Shuaiqiang Chang 已提交
1019

dengyihao's avatar
dengyihao 已提交
1020
  if (code != 0) {
H
Haojun Liao 已提交
1021 1022
    fprintf(stderr, "Failed to run %s, reason: %s\n", command, taos_errstr(pSql));
    taos_free_result(pSql);
H
hzcheng 已提交
1023 1024 1025
    taos_close(taos);
    exit(EXIT_FAILURE);
  }
H
Hui Li 已提交
1026 1027

  taos_free_result(pSql);
H
hzcheng 已提交
1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038
}

// sync insertion
void *syncWrite(void *sarg) {
  info *winfo = (info *)sarg;
  char buffer[BUFFER_SIZE] = "\0";
  char data[MAX_DATA_SIZE];
  char **data_type = winfo->datatype;
  int len_of_binary = winfo->len_of_binary;
  int ncols_per_record = winfo->ncols_per_record;
  srand(time(NULL));
S
slguan 已提交
1039
  int64_t time_counter = winfo->start_time;
H
hzcheng 已提交
1040 1041 1042
  for (int i = 0; i < winfo->nrecords_per_table;) {
    for (int tID = winfo->start_table_id; tID <= winfo->end_table_id; tID++) {
      int inserted = i;
S
slguan 已提交
1043
      int64_t tmp_time = time_counter;
H
hzcheng 已提交
1044 1045 1046 1047 1048

      char *pstr = buffer;
      pstr += sprintf(pstr, "insert into %s.%s%d values", winfo->db_name, winfo->tb_prefix, tID);
      int k;
      for (k = 0; k < winfo->nrecords_per_request;) {
1049
        int rand_num = rand() % 100;
1050 1051
        int len = -1;
        if (winfo->data_of_order ==1 && rand_num < winfo->data_of_rate) {
1052
          long d = tmp_time - rand() % 1000000 + rand_num;
1053 1054 1055
          len = generateData(data, data_type, ncols_per_record, d, len_of_binary);
        } else {
          len = generateData(data, data_type, ncols_per_record, tmp_time += 1000, len_of_binary);
S
Shuaiqiang Chang 已提交
1056
        }
1057 1058 1059 1060 1061 1062

        //assert(len + pstr - buffer < BUFFER_SIZE);
        if (len + pstr - buffer >= BUFFER_SIZE) { // too long
          break;
        }

H
hzcheng 已提交
1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100
        pstr += sprintf(pstr, " %s", data);
        inserted++;
        k++;

        if (inserted >= winfo->nrecords_per_table) break;
      }

      /* puts(buffer); */
      queryDB(winfo->taos, buffer);

      if (tID == winfo->end_table_id) {
        i = inserted;
        time_counter = tmp_time;
      }
    }
  }
  return NULL;
}

void *asyncWrite(void *sarg) {
  info *winfo = (info *)sarg;

  sTable *tb_infos = (sTable *)malloc(sizeof(sTable) * (winfo->end_table_id - winfo->start_table_id + 1));

  for (int tID = winfo->start_table_id; tID <= winfo->end_table_id; tID++) {
    sTable *tb_info = tb_infos + tID - winfo->start_table_id;
    tb_info->data_type = winfo->datatype;
    tb_info->ncols_per_record = winfo->ncols_per_record;
    tb_info->taos = winfo->taos;
    sprintf(tb_info->tb_name, "%s.%s%d", winfo->db_name, winfo->tb_prefix, tID);
    tb_info->timestamp = winfo->start_time;
    tb_info->counter = 0;
    tb_info->target = winfo->nrecords_per_table;
    tb_info->len_of_binary = winfo->len_of_binary;
    tb_info->nrecords_per_request = winfo->nrecords_per_request;
    tb_info->mutex_sem = &(winfo->mutex_sem);
    tb_info->notFinished = &(winfo->notFinished);
    tb_info->lock_sem = &(winfo->lock_sem);
S
Shuaiqiang Chang 已提交
1101 1102
    tb_info->data_of_order = winfo->data_of_order;
    tb_info->data_of_rate = winfo->data_of_rate;
H
hzcheng 已提交
1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121

    /* char buff[BUFFER_SIZE] = "\0"; */
    /* sprintf(buff, "insert into %s values (0, 0)", tb_info->tb_name); */
    /* queryDB(tb_info->taos,buff); */

    taos_query_a(winfo->taos, "show databases", callBack, tb_info);
  }

  sem_wait(&(winfo->lock_sem));
  free(tb_infos);

  return NULL;
}

void callBack(void *param, TAOS_RES *res, int code) {
  sTable *tb_info = (sTable *)param;
  char **datatype = tb_info->data_type;
  int ncols_per_record = tb_info->ncols_per_record;
  int len_of_binary = tb_info->len_of_binary;
S
slguan 已提交
1122
  int64_t tmp_time = tb_info->timestamp;
H
hzcheng 已提交
1123 1124

  if (code < 0) {
H
Haojun Liao 已提交
1125
    fprintf(stderr, "failed to insert data %d:reason; %s\n", code, taos_errstr(res));
H
hzcheng 已提交
1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143
    exit(EXIT_FAILURE);
  }

  // If finished;
  if (tb_info->counter >= tb_info->target) {
    sem_wait(tb_info->mutex_sem);
    (*(tb_info->notFinished))--;
    if (*(tb_info->notFinished) == 0) sem_post(tb_info->lock_sem);
    sem_post(tb_info->mutex_sem);
    return;
  }

  char buffer[BUFFER_SIZE] = "\0";
  char data[MAX_DATA_SIZE];
  char *pstr = buffer;
  pstr += sprintf(pstr, "insert into %s values", tb_info->tb_name);

  for (int i = 0; i < tb_info->nrecords_per_request; i++) {
1144
    int rand_num = rand() % 100;
S
Shuaiqiang Chang 已提交
1145 1146
    if (tb_info->data_of_order ==1 && rand_num < tb_info->data_of_rate)
    {
1147
      long d = tmp_time - rand() % 1000000 + rand_num;
S
Shuaiqiang Chang 已提交
1148 1149 1150 1151 1152
      generateData(data, datatype, ncols_per_record, d, len_of_binary);
    } else 
    {
      generateData(data, datatype, ncols_per_record, tmp_time += 1000, len_of_binary);
    }
H
hzcheng 已提交
1153 1154 1155 1156 1157 1158 1159
    pstr += sprintf(pstr, "%s", data);
    tb_info->counter++;

    if (tb_info->counter >= tb_info->target) {
      break;
    }
  }
S
Shuaiqiang Chang 已提交
1160
   tb_info->timestamp = tmp_time;
H
hzcheng 已提交
1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176

  taos_query_a(tb_info->taos, buffer, callBack, tb_info);

  taos_free_result(res);
}

double getCurrentTime() {
  struct timeval tv;
  if (gettimeofday(&tv, NULL) != 0) {
    perror("Failed to get current time in ms");
    exit(EXIT_FAILURE);
  }

  return tv.tv_sec + tv.tv_usec / 1E6;
}

L
Liu Tao 已提交
1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236
int32_t  randint[MAX_PREPARED_RAND];
int64_t  randbigint[MAX_PREPARED_RAND];
float  randfloat[MAX_PREPARED_RAND];
double  randdouble[MAX_PREPARED_RAND];

int32_t rand_tinyint(){
  static int cursor;
  cursor++;
  cursor = cursor % MAX_PREPARED_RAND;
  return randint[cursor] % 128;

}

int32_t rand_smallint(){
  static int cursor;
  cursor++;
  cursor = cursor % MAX_PREPARED_RAND;
  return randint[cursor] % 32767;
}

int32_t rand_int(){
  static int cursor;
  cursor++;
  cursor = cursor % MAX_PREPARED_RAND;
  return randint[cursor];
}

int64_t rand_bigint(){
  static int cursor;
  cursor++;
  cursor = cursor % MAX_PREPARED_RAND;
  return randbigint[cursor];
  
}

float rand_float(){
  static int cursor;
  cursor++;
  cursor = cursor % MAX_PREPARED_RAND;
  return randfloat[cursor];  
  
}

double rand_double() {
  static int cursor;
  cursor++;
  cursor = cursor % MAX_PREPARED_RAND;
  return randdouble[cursor];

}

void init_rand_data(){
  for (int i = 0; i < MAX_PREPARED_RAND; i++){
    randint[i] = (int)(rand() % 10);
    randbigint[i] = (int64_t)(rand() % 2147483648);
    randfloat[i] = (float)(rand() / 1000.0);
    randdouble[i] = (double)(rand() / 1000000.0);
  }
}

1237
int32_t generateData(char *res, char **data_type, int num_of_cols, int64_t timestamp, int len_of_binary) {
H
hzcheng 已提交
1238 1239
  memset(res, 0, MAX_DATA_SIZE);
  char *pstr = res;
L
lihui 已提交
1240
  pstr += sprintf(pstr, "(%" PRId64, timestamp);
H
hzcheng 已提交
1241 1242 1243
  int c = 0;

  for (; c < MAX_NUM_DATATYPE; c++) {
T
Tao Liu 已提交
1244
    if (data_type[c] == NULL) {
H
hzcheng 已提交
1245 1246 1247 1248
      break;
    }
  }

H
Hui Li 已提交
1249 1250 1251 1252 1253
  if (0 == c) {
    perror("data type error!");
    exit(-1);
  }

H
hzcheng 已提交
1254 1255
  for (int i = 0; i < num_of_cols; i++) {
    if (strcasecmp(data_type[i % c], "tinyint") == 0) {
L
Liu Tao 已提交
1256
      pstr += sprintf(pstr, ", %d", rand_tinyint() );
H
hzcheng 已提交
1257
    } else if (strcasecmp(data_type[i % c], "smallint") == 0) {
L
Liu Tao 已提交
1258
      pstr += sprintf(pstr, ", %d", rand_smallint());
H
hzcheng 已提交
1259
    } else if (strcasecmp(data_type[i % c], "int") == 0) {
L
Liu Tao 已提交
1260
      pstr += sprintf(pstr, ", %d", rand_int()); 
H
hzcheng 已提交
1261
    } else if (strcasecmp(data_type[i % c], "bigint") == 0) {
L
Liu Tao 已提交
1262
      pstr += sprintf(pstr, ", %" PRId64, rand_bigint());
H
hzcheng 已提交
1263
    } else if (strcasecmp(data_type[i % c], "float") == 0) {
L
Liu Tao 已提交
1264
      pstr += sprintf(pstr, ", %10.4f", rand_float());
H
hzcheng 已提交
1265
    } else if (strcasecmp(data_type[i % c], "double") == 0) {
L
Liu Tao 已提交
1266
      double t = rand_double();
H
hzcheng 已提交
1267 1268
      pstr += sprintf(pstr, ", %20.8f", t);
    } else if (strcasecmp(data_type[i % c], "bool") == 0) {
1269
      bool b = rand() & 1;
H
hzcheng 已提交
1270 1271 1272 1273
      pstr += sprintf(pstr, ", %s", b ? "true" : "false");
    } else if (strcasecmp(data_type[i % c], "binary") == 0) {
      char s[len_of_binary];
      rand_string(s, len_of_binary);
F
fang 已提交
1274
      pstr += sprintf(pstr, ", \"%s\"", s);
L
Liu Tao 已提交
1275 1276 1277 1278
    }else if (strcasecmp(data_type[i % c], "nchar") == 0) {
      char s[len_of_binary];
      rand_string(s, len_of_binary);
      pstr += sprintf(pstr, ", \"%s\"", s);
H
hzcheng 已提交
1279
    }
1280 1281 1282 1283 1284

    if (pstr - res > MAX_DATA_SIZE) {
      perror("column length too long, abort");
      exit(-1);
    }
H
hzcheng 已提交
1285 1286 1287
  }

  pstr += sprintf(pstr, ")");
1288 1289

  return pstr - res;
H
hzcheng 已提交
1290 1291
}

S
Shuaiqiang Chang 已提交
1292
static const char charset[] = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJK1234567890";
H
hzcheng 已提交
1293
void rand_string(char *str, int size) {
S
Shuaiqiang Chang 已提交
1294 1295
  str[0] = 0;
  if (size > 0) {
H
hzcheng 已提交
1296
    --size;
S
Shuaiqiang Chang 已提交
1297 1298
    int n;
    for (n = 0; n < size; n++) {
1299
      int key = rand() % (int)(sizeof charset - 1);
S
Shuaiqiang Chang 已提交
1300
      str[n] = charset[key];
H
hzcheng 已提交
1301
    }
S
Shuaiqiang Chang 已提交
1302
    str[n] = 0;
H
hzcheng 已提交
1303 1304
  }
}