taosdemo.c 39.7 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 45 46 47
#define BUFFER_SIZE      65536
#define MAX_DB_NAME_SIZE 64
#define MAX_TB_NAME_SIZE 64
#define MAX_DATA_SIZE    1024
#define MAX_NUM_DATATYPE 8
#define OPT_ABORT        1 /* –abort */
48
#define STRING_LEN       512
H
hzcheng 已提交
49 50 51 52 53 54

/* 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 已提交
55 56
  {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 已提交
57
  {0, 'm', "table_prefix",             0, "Table prefix name. Default is 't'.",                                                                               3},
H
Hui Li 已提交
58
  {0, 's', "sql file",                 0, "The select sql file.",                                                                                             3},
H
hzcheng 已提交
59 60 61 62 63
  {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},
64
  {0, 'l', "num_of_cols_per_record",   0, "The number of columns per record. Default is 3.",                                                                  8},
S
Shuaiqiang Chang 已提交
65
  {0, 'T', "num_of_threads",           0, "The number of threads. Default is 10.",                                                                            9},
S
slguan 已提交
66 67 68
  {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 已提交
69
  {0, 'c', "config_directory",         0, "Configuration directory. Default is '/etc/taos/'.",                                                                14},
H
hzcheng 已提交
70
  {0, 'x', 0,                          0, "Insert only flag.",                                                                                                13},
S
Shuaiqiang Chang 已提交
71 72 73
  {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 已提交
74 75 76
  {0}};

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

/* 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. */
107
  SDemoArguments *arguments = state->input;
H
hzcheng 已提交
108 109 110 111 112 113 114 115 116 117 118 119
  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 已提交
120
    case 'P':
H
hzcheng 已提交
121 122 123 124 125
      arguments->password = arg;
      break;
    case 'o':
      arguments->output_file = arg;
      break;
H
Hui Li 已提交
126 127 128
    case 's':
      arguments->sqlFile = arg;
      break;
H
hzcheng 已提交
129 130 131
    case 'q':
      arguments->mode = atoi(arg);
      break;
S
Shuaiqiang Chang 已提交
132 133
    case 'T':
      arguments->num_of_threads = atoi(arg);
H
hzcheng 已提交
134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187
      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 &&
            strcasecmp(arg, "BINARY")) {
          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 &&
              strcasecmp(token, "DOUBLE") != 0 && strcasecmp(token, "BINARY")) {
            argp_error(state, "Invalid data_type!");
          }
          sptr[index++] = token;
          token = strsep(&running, ", ");
        }
      }
      break;
    case 'w':
      arguments->len_of_binary = atoi(arg);
      break;
    case 'm':
      arguments->tb_prefix = arg;
      break;
    case 'M':
H
Hui Li 已提交
188
      arguments->use_metric = false;
H
hzcheng 已提交
189 190
      break;
    case 'x':
H
Hui Li 已提交
191
      arguments->insert_only = false;
H
hzcheng 已提交
192
      break;
S
Shuaiqiang Chang 已提交
193
    case 'c':
H
hzcheng 已提交
194 195 196 197
      if (wordexp(arg, &full_path, 0) != 0) {
        fprintf(stderr, "Invalid path %s\n", arg);
        return -1;
      }
198
      taos_options(TSDB_OPTION_CONFIGDIR, full_path.we_wordv[0]);
H
hzcheng 已提交
199 200
      wordfree(&full_path);
      break;
S
Shuaiqiang Chang 已提交
201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217
    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 已提交
218 219 220 221 222 223 224
    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 已提交
225 226 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
    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 已提交
258 259
  int data_of_order;
  int data_of_rate;
S
slguan 已提交
260
  int64_t start_time;
H
hzcheng 已提交
261
  bool do_aggreFunc;
H
Hui Li 已提交
262 263 264
  
  char* cols;  
  bool  use_metric;
H
hzcheng 已提交
265 266 267 268 269 270 271 272 273 274

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

typedef struct {
  TAOS  *taos;

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

  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 已提交
306 307
void *deleteTable();

H
hzcheng 已提交
308 309
void *asyncWrite(void *sarg);

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

void rand_string(char *str, int size);

double getCurrentTime();

void callBack(void *param, TAOS_RES *res, int code);
H
Hui Li 已提交
317 318
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 已提交
319 320

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

  /* Parse our arguments; every option seen by parse_opt will be
     reflected in arguments. */
357
  // For demo use, change default values for some parameters;
358
  arguments.num_of_tables = 10000;
359
  arguments.num_of_CPR = 3; 
S
Shuaiqiang Chang 已提交
360
  arguments.num_of_threads = 10;
361 362
  arguments.num_of_DPT = 100000;
  arguments.num_of_RPR = 1000;
363
  arguments.use_metric = true;
L
liu0x54 已提交
364
  arguments.insert_only = false;
S
slguan 已提交
365
  // end change
366

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

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

H
hzcheng 已提交
377 378
  enum MODE query_mode = arguments.mode;
  char *ip_addr = arguments.host;
L
lihui 已提交
379
  uint16_t port = arguments.port;
H
hzcheng 已提交
380 381 382 383 384 385
  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 已提交
386 387
  int order = arguments.order;
  int rate = arguments.rate;
S
Shuaiqiang Chang 已提交
388
  int method_of_delete = arguments.method_of_delete;
H
hzcheng 已提交
389
  int ntables = arguments.num_of_tables;
S
Shuaiqiang Chang 已提交
390
  int threads = arguments.num_of_threads;
H
hzcheng 已提交
391 392 393 394 395 396
  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;
397
  char dataString[STRING_LEN];
H
hzcheng 已提交
398
  bool do_aggreFunc = true;
399

H
Hui Li 已提交
400 401 402 403 404 405 406
  if (NULL != arguments.sqlFile) {
    TAOS* qtaos = taos_connect(ip_addr, user, pass, db_name, port);
    querySqlFile(qtaos, arguments.sqlFile);  
    taos_close(qtaos);
    return 0;
  }

407 408
  memset(dataString, 0, STRING_LEN);
  int len = 0;
409

H
hzcheng 已提交
410 411 412 413 414 415 416
  if (strcasecmp(data_type[0], "BINARY") == 0 || strcasecmp(data_type[0], "BOOL") == 0) {
    do_aggreFunc = false;
  }
  for (; count_data_type <= MAX_NUM_DATATYPE; count_data_type++) {
    if (strcasecmp(data_type[count_data_type], "") == 0) {
      break;
    }
417 418

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

  FILE *fp = fopen(arguments.output_file, "a");
422 423 424 425 426
  if (NULL == fp) {
    fprintf(stderr, "Failed to open %s for writing\n", arguments.output_file);
    return 1;
  };
  
H
hzcheng 已提交
427 428
  time_t tTime = time(NULL);
  struct tm tm = *localtime(&tTime);
S
Shuaiqiang Chang 已提交
429 430 431 432 433 434 435 436 437
  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",
          (strcasestr(dataString, "BINARY") != NULL) ? len_of_binary : -1);
  printf("# Number of Columns per record:      %d\n", ncols_per_record);
S
Shuaiqiang Chang 已提交
438
  printf("# Number of Threads:                 %d\n", threads);
S
Shuaiqiang Chang 已提交
439 440 441 442 443
  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 已提交
444 445
  if (order == 1)
  {
S
Shuaiqiang Chang 已提交
446 447
    printf("# Data order:                        %d\n", order);
    printf("# Data out of order rate:            %d\n", rate);
S
Shuaiqiang Chang 已提交
448 449

  }
S
Shuaiqiang Chang 已提交
450
  printf("# Delete method:                     %d\n", method_of_delete);
S
Shuaiqiang Chang 已提交
451 452 453 454
  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 已提交
455
  (void)getchar();
H
hzcheng 已提交
456 457

  fprintf(fp, "###################################################################\n");
L
lihui 已提交
458
  fprintf(fp, "# Server IP:                         %s:%hu\n", ip_addr == NULL ? "localhost" : ip_addr, port);
H
hzcheng 已提交
459 460 461 462 463 464 465
  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",
          (strcasestr(dataString, "BINARY") != NULL) ? len_of_binary : -1);
  fprintf(fp, "# Number of Columns per record:      %d\n", ncols_per_record);
S
Shuaiqiang Chang 已提交
466
  fprintf(fp, "# Number of Threads:                 %d\n", threads);
H
hzcheng 已提交
467 468 469 470 471
  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 已提交
472 473 474 475 476 477
  if (order == 1)
  {
    printf("# Data order:                        %d\n", order);
    printf("# Data out of order rate:            %d\n", rate);

  }
H
hzcheng 已提交
478 479 480 481 482 483 484 485
  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 已提交
486
    fprintf(stderr, "Failed to connect to TDengine, reason:%s\n", taos_errstr(NULL));
H
hzcheng 已提交
487 488 489 490 491
    return 1;
  }
  char command[BUFFER_SIZE] = "\0";

  sprintf(command, "drop database %s;", db_name);
492 493
  TAOS_RES* res = taos_query(taos, command);
  taos_free_result(res);
H
hzcheng 已提交
494 495

  sprintf(command, "create database %s;", db_name);
496 497
  res = taos_query(taos, command);
  taos_free_result(res);
H
hzcheng 已提交
498

499
  char cols[STRING_LEN] = "\0";
H
hzcheng 已提交
500
  int colIndex = 0;
501
  len = 0;
H
hzcheng 已提交
502 503 504

  for (; colIndex < ncols_per_record - 1; colIndex++) {
    if (strcasecmp(data_type[colIndex % count_data_type], "BINARY") != 0) {
505
      len += snprintf(cols + len, STRING_LEN - len, ",f%d %s", colIndex + 1, data_type[colIndex % count_data_type]);
H
hzcheng 已提交
506
    } else {
507
      len += snprintf(cols + len, STRING_LEN - len, ",f%d %s(%d)", colIndex + 1, data_type[colIndex % count_data_type], len_of_binary);
H
hzcheng 已提交
508 509 510 511
    }
  }

  if (strcasecmp(data_type[colIndex % count_data_type], "BINARY") != 0) {
512
    len += snprintf(cols + len, STRING_LEN - len, ",f%d %s)", colIndex + 1, data_type[colIndex % count_data_type]);
H
hzcheng 已提交
513
  } else {
514
    len += snprintf(cols + len, STRING_LEN - len, ",f%d %s(%d))", colIndex + 1, data_type[colIndex % count_data_type], len_of_binary);
H
hzcheng 已提交
515 516
  }

H
Hui Li 已提交
517
  if (use_metric) {
H
hzcheng 已提交
518
    /* Create metric table */
519
    printf("Creating meters super table...\n");
520
    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 已提交
521
    queryDB(taos, command);
522
    printf("meters created!\n");
H
hzcheng 已提交
523
  }
H
Hui Li 已提交
524
  taos_close(taos);
H
Hui Li 已提交
525
  
H
hzcheng 已提交
526
  /* Wait for table to create  */
H
Hui Li 已提交
527
  multiThreadCreateTable(cols, use_metric, threads, ntables, db_name, tb_prefix, ip_addr, port, user, pass);
S
Shuaiqiang Chang 已提交
528
  
H
hzcheng 已提交
529 530 531
  /* Insert data */
  double ts = getCurrentTime();
  printf("Inserting data......\n");
S
Shuaiqiang Chang 已提交
532 533
  pthread_t *pids = malloc(threads * sizeof(pthread_t));
  info *infos = malloc(threads * sizeof(info));
H
hzcheng 已提交
534

S
Shuaiqiang Chang 已提交
535
  int a = ntables / threads;
H
hzcheng 已提交
536
  if (a < 1) {
S
Shuaiqiang Chang 已提交
537
    threads = ntables;
H
hzcheng 已提交
538 539
    a = 1;
  }
S
Shuduo Sang 已提交
540 541 542 543

  int b = 0;
  if (threads != 0)
    b = ntables % threads;
H
hzcheng 已提交
544
  int last = 0;
S
Shuaiqiang Chang 已提交
545
  for (int i = 0; i < threads; i++) {
H
hzcheng 已提交
546 547
    info *t_info = infos + i;
    t_info->threadID = i;
H
Hui Li 已提交
548 549
    tstrncpy(t_info->db_name, db_name, MAX_DB_NAME_SIZE);
    tstrncpy(t_info->tb_prefix, tb_prefix, MAX_TB_NAME_SIZE);
H
hzcheng 已提交
550 551 552 553 554 555 556 557
    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 已提交
558 559
    t_info->data_of_order = order;
    t_info->data_of_rate = rate;
H
hzcheng 已提交
560 561 562 563 564 565 566 567 568 569 570 571 572
    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 已提交
573
  for (int i = 0; i < threads; i++) {
H
hzcheng 已提交
574 575 576 577 578
    pthread_join(pids[i], NULL);
  }

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

  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 已提交
593
  for (int i = 0; i < threads; i++) {
H
hzcheng 已提交
594 595 596 597 598 599 600 601 602 603
    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 已提交
604 605 606 607 608 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
  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 已提交
653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677
  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 已提交
678 679 680 681 682 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
#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 已提交
865 866 867 868
void *readTable(void *sarg) {
  info *rinfo = (info *)sarg;
  TAOS *taos = rinfo->taos;
  char command[BUFFER_SIZE] = "\0";
S
slguan 已提交
869
  int64_t sTime = rinfo->start_time;
H
hzcheng 已提交
870 871
  char *tb_prefix = rinfo->tb_prefix;
  FILE *fp = fopen(rinfo->fp, "a");
H
Hui Li 已提交
872 873 874 875 876
  if (NULL == fp) {
    printf("fopen %s fail, reason:%s.\n", rinfo->fp, strerror(errno));
    return NULL;
  }
  
H
hzcheng 已提交
877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892
  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 已提交
893
      sprintf(command, "select %s from %s%d where ts>= %" PRId64, aggreFunc[j], tb_prefix, i, sTime);
H
hzcheng 已提交
894 895

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

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

S
Shuaiqiang Chang 已提交
906
      while (taos_fetch_row(pSql) != NULL) {
H
hzcheng 已提交
907 908 909 910 911 912
        count++;
      }

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

S
Shuaiqiang Chang 已提交
913
      taos_free_result(pSql);
H
hzcheng 已提交
914 915 916
    }

    fprintf(fp, "|%10s  |   %10d   |  %12.2f   |   %10.2f  |\n",
S
slguan 已提交
917
            aggreFunc[j][0] == '*' ? "   *   " : aggreFunc[j], totalData,
H
hzcheng 已提交
918 919 920 921 922 923 924 925 926 927 928 929 930 931
            (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 已提交
932 933 934 935 936
  if (NULL == fp) {
    printf("fopen %s fail, reason:%s.\n", rinfo->fp, strerror(errno));
    return NULL;
  }
  
H
hzcheng 已提交
937 938 939 940 941 942 943 944 945 946 947 948 949
  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 已提交
950
    char condition[BUFFER_SIZE - 30] = "\0";
B
Bomin Zhang 已提交
951
    char tempS[64] = "\0";
H
hzcheng 已提交
952 953 954 955 956

    int m = 10 < num_of_tables ? 10 : num_of_tables;

    for (int i = 1; i <= m; i++) {
      if (i == 1) {
L
liu0x54 已提交
957
        sprintf(tempS, "areaid = %d", i);
H
hzcheng 已提交
958
      } else {
L
liu0x54 已提交
959
        sprintf(tempS, " or areaid = %d ", i);
H
hzcheng 已提交
960 961 962
      }
      strcat(condition, tempS);

L
liu0x54 已提交
963
      sprintf(command, "select %s from meters where %s", aggreFunc[j], condition);
H
hzcheng 已提交
964 965 966 967 968 969

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

      double t = getCurrentTime();

S
Shuaiqiang Chang 已提交
970 971 972 973
      TAOS_RES *pSql = taos_query(taos, command);
      int32_t code = taos_errno(pSql);

      if (code != 0) {
H
Haojun Liao 已提交
974
        fprintf(stderr, "Failed to query:%s\n", taos_errstr(pSql));
S
Shuaiqiang Chang 已提交
975
        taos_free_result(pSql);
H
hzcheng 已提交
976 977 978 979
        taos_close(taos);
        exit(1);
      }
      int count = 0;
S
Shuaiqiang Chang 已提交
980
      while (taos_fetch_row(pSql) != NULL) {
H
hzcheng 已提交
981 982 983 984 985 986 987
        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 已提交
988
      taos_free_result(pSql);
H
hzcheng 已提交
989 990 991 992 993 994 995 996 997
    }
    fprintf(fp, "\n");
  }

  fclose(fp);
  return NULL;
}

void queryDB(TAOS *taos, char *command) {
H
Hui Li 已提交
998
  int i;
S
Shuaiqiang Chang 已提交
999
  TAOS_RES *pSql = NULL;
H
Hui Li 已提交
1000 1001 1002 1003 1004 1005 1006 1007
  int32_t   code = -1;

  for (i = 0; i < 5; i++) {
    if (NULL != pSql) {
      taos_free_result(pSql);
      pSql = NULL;
    }
    
S
Shuaiqiang Chang 已提交
1008 1009
    pSql = taos_query(taos, command);
    code = taos_errno(pSql);
H
Hui Li 已提交
1010
    if (0 == code) {
S
Shuaiqiang Chang 已提交
1011
      break;
H
Hui Li 已提交
1012
    }    
1013
  }
S
Shuaiqiang Chang 已提交
1014

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

  taos_free_result(pSql);
H
hzcheng 已提交
1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033
}

// 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 已提交
1034
  int64_t time_counter = winfo->start_time;
H
hzcheng 已提交
1035 1036 1037
  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 已提交
1038
      int64_t tmp_time = time_counter;
H
hzcheng 已提交
1039 1040 1041 1042 1043

      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;) {
1044
        int rand_num = rand() % 100;
1045 1046
        int len = -1;
        if (winfo->data_of_order ==1 && rand_num < winfo->data_of_rate) {
1047
          long d = tmp_time - rand() % 1000000 + rand_num;
1048 1049 1050
          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 已提交
1051
        }
1052 1053 1054 1055 1056 1057

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

H
hzcheng 已提交
1058 1059 1060 1061 1062 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
        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 已提交
1096 1097
    tb_info->data_of_order = winfo->data_of_order;
    tb_info->data_of_rate = winfo->data_of_rate;
H
hzcheng 已提交
1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116

    /* 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 已提交
1117
  int64_t tmp_time = tb_info->timestamp;
H
hzcheng 已提交
1118 1119

  if (code < 0) {
H
Haojun Liao 已提交
1120
    fprintf(stderr, "failed to insert data %d:reason; %s\n", code, taos_errstr(res));
H
hzcheng 已提交
1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138
    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++) {
1139
    int rand_num = rand() % 100;
S
Shuaiqiang Chang 已提交
1140 1141
    if (tb_info->data_of_order ==1 && rand_num < tb_info->data_of_rate)
    {
1142
      long d = tmp_time - rand() % 1000000 + rand_num;
S
Shuaiqiang Chang 已提交
1143 1144 1145 1146 1147
      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 已提交
1148 1149 1150 1151 1152 1153 1154
    pstr += sprintf(pstr, "%s", data);
    tb_info->counter++;

    if (tb_info->counter >= tb_info->target) {
      break;
    }
  }
S
Shuaiqiang Chang 已提交
1155
   tb_info->timestamp = tmp_time;
H
hzcheng 已提交
1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171

  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;
}

1172
int32_t generateData(char *res, char **data_type, int num_of_cols, int64_t timestamp, int len_of_binary) {
H
hzcheng 已提交
1173 1174
  memset(res, 0, MAX_DATA_SIZE);
  char *pstr = res;
L
lihui 已提交
1175
  pstr += sprintf(pstr, "(%" PRId64, timestamp);
H
hzcheng 已提交
1176 1177 1178 1179 1180 1181 1182 1183
  int c = 0;

  for (; c < MAX_NUM_DATATYPE; c++) {
    if (strcasecmp(data_type[c], "") == 0) {
      break;
    }
  }

H
Hui Li 已提交
1184 1185 1186 1187 1188
  if (0 == c) {
    perror("data type error!");
    exit(-1);
  }

H
hzcheng 已提交
1189 1190
  for (int i = 0; i < num_of_cols; i++) {
    if (strcasecmp(data_type[i % c], "tinyint") == 0) {
1191
      pstr += sprintf(pstr, ", %d", (int)(rand() % 128));
H
hzcheng 已提交
1192
    } else if (strcasecmp(data_type[i % c], "smallint") == 0) {
1193
      pstr += sprintf(pstr, ", %d", (int)(rand() % 32767));
H
hzcheng 已提交
1194
    } else if (strcasecmp(data_type[i % c], "int") == 0) {
1195
      pstr += sprintf(pstr, ", %d", (int)(rand() % 10)); 
H
hzcheng 已提交
1196
    } else if (strcasecmp(data_type[i % c], "bigint") == 0) {
1197
      pstr += sprintf(pstr, ", %" PRId64, rand() % 2147483648);
H
hzcheng 已提交
1198
    } else if (strcasecmp(data_type[i % c], "float") == 0) {
1199
      pstr += sprintf(pstr, ", %10.4f", (float)(rand() / 1000.0));
H
hzcheng 已提交
1200
    } else if (strcasecmp(data_type[i % c], "double") == 0) {
1201
      double t = (double)(rand() / 1000000.0);
H
hzcheng 已提交
1202 1203
      pstr += sprintf(pstr, ", %20.8f", t);
    } else if (strcasecmp(data_type[i % c], "bool") == 0) {
1204
      bool b = rand() & 1;
H
hzcheng 已提交
1205 1206 1207 1208
      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 已提交
1209
      pstr += sprintf(pstr, ", \"%s\"", s);
H
hzcheng 已提交
1210
    }
1211 1212 1213 1214 1215

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

  pstr += sprintf(pstr, ")");
1219 1220

  return pstr - res;
H
hzcheng 已提交
1221 1222
}

S
Shuaiqiang Chang 已提交
1223
static const char charset[] = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJK1234567890";
H
hzcheng 已提交
1224
void rand_string(char *str, int size) {
S
Shuaiqiang Chang 已提交
1225 1226
  str[0] = 0;
  if (size > 0) {
H
hzcheng 已提交
1227
    --size;
S
Shuaiqiang Chang 已提交
1228 1229
    int n;
    for (n = 0; n < size; n++) {
1230
      int key = rand() % (int)(sizeof charset - 1);
S
Shuaiqiang Chang 已提交
1231
      str[n] = charset[key];
H
hzcheng 已提交
1232
    }
S
Shuaiqiang Chang 已提交
1233
    str[n] = 0;
H
hzcheng 已提交
1234 1235
  }
}