xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

[TD5065]<feature>: taosdemo nano second support (#6902)

This commit is contained in:
Shuduo Sang 2021-07-17 18:40:11 +08:00 committed by GitHub
parent 91774fda49
commit 93396bca69
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 1537 additions and 1520 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
src/kit/taosdemo/taosdemo.c
View File

@ -54,6 +54,7 @@
#include "tutil.h" #include "tutil.h"
#define STMT_IFACE_ENABLED 1 #define STMT_IFACE_ENABLED 1
#define NANO_SECOND_ENABLED 1
#define REQ_EXTRA_BUF_LEN 1024 #define REQ_EXTRA_BUF_LEN 1024
#define RESP_BUF_LEN 4096 #define RESP_BUF_LEN 4096
@ -66,13 +67,6 @@ extern char configDir[];
#define STR_INSERT_INTO "INSERT INTO " #define STR_INSERT_INTO "INSERT INTO "
enum TEST_MODE {
INSERT_TEST, // 0
QUERY_TEST, // 1
SUBSCRIBE_TEST, // 2
INVAID_TEST
};
#define MAX_RECORDS_PER_REQ 32766 #define MAX_RECORDS_PER_REQ 32766
#define HEAD_BUFF_LEN TSDB_MAX_COLUMNS*24 // 16*MAX_COLUMNS + (192+32)*2 + insert into .. #define HEAD_BUFF_LEN TSDB_MAX_COLUMNS*24 // 16*MAX_COLUMNS + (192+32)*2 + insert into ..
@ -105,6 +99,13 @@ enum TEST_MODE {
#define DEFAULT_TIMESTAMP_STEP 1 #define DEFAULT_TIMESTAMP_STEP 1
enum TEST_MODE {
INSERT_TEST, // 0
QUERY_TEST, // 1
SUBSCRIBE_TEST, // 2
INVAID_TEST
};
typedef enum CREATE_SUB_TALBE_MOD_EN { typedef enum CREATE_SUB_TALBE_MOD_EN {
PRE_CREATE_SUBTBL, PRE_CREATE_SUBTBL,
AUTO_CREATE_SUBTBL, AUTO_CREATE_SUBTBL,
@ -228,7 +229,7 @@ typedef struct SArguments_S {
int64_t num_of_DPT; int64_t num_of_DPT;
int abort; int abort;
uint32_t disorderRatio; // 0: no disorder, >0: x% uint32_t disorderRatio; // 0: no disorder, >0: x%
int disorderRange; // ms or us by database precision int disorderRange; // ms, us or ns. accordig to database precision
uint32_t method_of_delete; uint32_t method_of_delete;
char ** arg_list; char ** arg_list;
uint64_t totalInsertRows; uint64_t totalInsertRows;
@ -256,10 +257,10 @@ typedef struct SSuperTable_S {
int64_t childTblLimit; int64_t childTblLimit;
uint64_t childTblOffset; uint64_t childTblOffset;
// int multiThreadWriteOneTbl; // 0: no, 1: yes // int multiThreadWriteOneTbl; // 0: no, 1: yes
uint32_t interlaceRows; // uint32_t interlaceRows; //
int disorderRatio; // 0: no disorder, >0: x% int disorderRatio; // 0: no disorder, >0: x%
int disorderRange; // ms or us by database precision int disorderRange; // ms, us or ns. according to database precision
uint64_t maxSqlLen; // uint64_t maxSqlLen; //
uint64_t insertInterval; // insert interval, will override global insert interval uint64_t insertInterval; // insert interval, will override global insert interval
@ -317,7 +318,7 @@ typedef struct {
} SDbInfo; } SDbInfo;
typedef struct SDbCfg_S { typedef struct SDbCfg_S {
// int maxtablesPerVnode; // int maxtablesPerVnode;
uint32_t minRows; // 0 means default uint32_t minRows; // 0 means default
uint32_t maxRows; // 0 means default uint32_t maxRows; // 0 means default
int comp; int comp;
@ -1513,7 +1514,10 @@ static int printfInsertMeta() {
} }
if (g_Dbs.db[i].dbCfg.precision[0] != 0) { if (g_Dbs.db[i].dbCfg.precision[0] != 0) {
if ((0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "ms", 2)) if ((0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "ms", 2))
|| (0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "us", 2))) { #if NANO_SECOND_ENABLED == 1
|| (0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "us", 2))
#endif
|| (0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "ns", 2))) {
printf(" precision: \033[33m%s\033[0m\n", printf(" precision: \033[33m%s\033[0m\n",
g_Dbs.db[i].dbCfg.precision); g_Dbs.db[i].dbCfg.precision);
} else { } else {
@ -1703,6 +1707,9 @@ static void printfInsertMetaToFile(FILE* fp) {
} }
if (g_Dbs.db[i].dbCfg.precision[0] != 0) { if (g_Dbs.db[i].dbCfg.precision[0] != 0) {
if ((0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "ms", 2)) if ((0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "ms", 2))
#if NANO_SECOND_ENABLED == 1
|| (0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "ns", 2))
#endif
|| (0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "us", 2))) { || (0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "us", 2))) {
fprintf(fp, " precision: %s\n", fprintf(fp, " precision: %s\n",
g_Dbs.db[i].dbCfg.precision); g_Dbs.db[i].dbCfg.precision);
@ -1903,10 +1910,12 @@ static void printfQueryMeta() {
static char* formatTimestamp(char* buf, int64_t val, int precision) { static char* formatTimestamp(char* buf, int64_t val, int precision) {
time_t tt; time_t tt;
if (precision == TSDB_TIME_PRECISION_NANO) { if (precision == TSDB_TIME_PRECISION_MICRO) {
tt = (time_t)(val / 1000000000);
} else if (precision == TSDB_TIME_PRECISION_MICRO) {
tt = (time_t)(val / 1000000); tt = (time_t)(val / 1000000);
#if NANO_SECOND_ENABLED == 1
} if (precision == TSDB_TIME_PRECISION_NANO) {
tt = (time_t)(val / 1000000000);
#endif
} else { } else {
tt = (time_t)(val / 1000); tt = (time_t)(val / 1000);
} }
@ -1926,10 +1935,12 @@ static char* formatTimestamp(char* buf, int64_t val, int precision) {
struct tm* ptm = localtime(&tt); struct tm* ptm = localtime(&tt);
size_t pos = strftime(buf, 32, "%Y-%m-%d %H:%M:%S", ptm); size_t pos = strftime(buf, 32, "%Y-%m-%d %H:%M:%S", ptm);
if (precision == TSDB_TIME_PRECISION_NANO) { if (precision == TSDB_TIME_PRECISION_MICRO) {
sprintf(buf + pos, ".%09d", (int)(val % 1000000000));
} else if (precision == TSDB_TIME_PRECISION_MICRO) {
sprintf(buf + pos, ".%06d", (int)(val % 1000000)); sprintf(buf + pos, ".%06d", (int)(val % 1000000));
#if NANO_SECOND_ENABLED == 1
} else if (precision == TSDB_TIME_PRECISION_NANO) {
sprintf(buf + pos, ".%09d", (int)(val % 1000000000));
#endif
} else { } else {
sprintf(buf + pos, ".%03d", (int)(val % 1000)); sprintf(buf + pos, ".%03d", (int)(val % 1000));
} }
@ -2953,6 +2964,10 @@ static int createDatabasesAndStables() {
" fsync %d", g_Dbs.db[i].dbCfg.fsync); " fsync %d", g_Dbs.db[i].dbCfg.fsync);
} }
if ((0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "ms", strlen("ms"))) if ((0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "ms", strlen("ms")))
#if NANO_SECOND_ENABLED == 1
|| (0 == strncasecmp(g_Dbs.db[i].dbCfg.precision,
"ns", strlen("ns")))
#endif
|| (0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, || (0 == strncasecmp(g_Dbs.db[i].dbCfg.precision,
"us", strlen("us")))) { "us", strlen("us")))) {
dataLen += snprintf(command + dataLen, BUFFER_SIZE - dataLen, dataLen += snprintf(command + dataLen, BUFFER_SIZE - dataLen,
@ -4140,7 +4155,7 @@ static bool getMetaFromInsertJsonFile(cJSON* root) {
__func__, __LINE__); __func__, __LINE__);
goto PARSE_OVER; goto PARSE_OVER;
} }
/* /*
cJSON *multiThreadWriteOneTbl = cJSON *multiThreadWriteOneTbl =
cJSON_GetObjectItem(stbInfo, "multi_thread_write_one_tbl"); // no , yes cJSON_GetObjectItem(stbInfo, "multi_thread_write_one_tbl"); // no , yes
if (multiThreadWriteOneTbl if (multiThreadWriteOneTbl
@ -4157,7 +4172,7 @@ static bool getMetaFromInsertJsonFile(cJSON* root) {
printf("ERROR: failed to read json, multiThreadWriteOneTbl not found\n"); printf("ERROR: failed to read json, multiThreadWriteOneTbl not found\n");
goto PARSE_OVER; goto PARSE_OVER;
} }
*/ */
cJSON* insertRows = cJSON_GetObjectItem(stbInfo, "insert_rows"); cJSON* insertRows = cJSON_GetObjectItem(stbInfo, "insert_rows");
if (insertRows && insertRows->type == cJSON_Number) { if (insertRows && insertRows->type == cJSON_Number) {
if (insertRows->valueint < 0) { if (insertRows->valueint < 0) {
@ -6541,8 +6556,10 @@ static void startMultiThreadInsertData(int threads, char* db_name,
timePrec = TSDB_TIME_PRECISION_MILLI; timePrec = TSDB_TIME_PRECISION_MILLI;
} else if (0 == strncasecmp(precision, "us", 2)) { } else if (0 == strncasecmp(precision, "us", 2)) {
timePrec = TSDB_TIME_PRECISION_MICRO; timePrec = TSDB_TIME_PRECISION_MICRO;
#if NANO_SECOND_ENABLED == 1
} else if (0 == strncasecmp(precision, "ns", 2)) { } else if (0 == strncasecmp(precision, "ns", 2)) {
timePrec = TSDB_TIME_PRECISION_NANO; timePrec = TSDB_TIME_PRECISION_NANO;
#endif
} else { } else {
errorPrint("Not support precision: %s\n", precision); errorPrint("Not support precision: %s\n", precision);
exit(-1); exit(-1);
@ -6761,19 +6778,19 @@ static void startMultiThreadInsertData(int threads, char* db_name,
pThreadInfo->taos = NULL; pThreadInfo->taos = NULL;
} }
/* if ((NULL == superTblInfo) /* if ((NULL == superTblInfo)
|| (0 == superTblInfo->multiThreadWriteOneTbl)) { || (0 == superTblInfo->multiThreadWriteOneTbl)) {
*/ */
pThreadInfo->start_table_from = tableFrom; pThreadInfo->start_table_from = tableFrom;
pThreadInfo->ntables = i<b?a+1:a; pThreadInfo->ntables = i<b?a+1:a;
pThreadInfo->end_table_to = i < b ? tableFrom + a : tableFrom + a - 1; pThreadInfo->end_table_to = i < b ? tableFrom + a : tableFrom + a - 1;
tableFrom = pThreadInfo->end_table_to + 1; tableFrom = pThreadInfo->end_table_to + 1;
/* } else { /* } else {
pThreadInfo->start_table_from = 0; pThreadInfo->start_table_from = 0;
pThreadInfo->ntables = superTblInfo->childTblCount; pThreadInfo->ntables = superTblInfo->childTblCount;
pThreadInfo->start_time = pThreadInfo->start_time + rand_int() % 10000 - rand_tinyint(); pThreadInfo->start_time = pThreadInfo->start_time + rand_int() % 10000 - rand_tinyint();
} }
*/ */
tsem_init(&(pThreadInfo->lock_sem), 0, 0); tsem_init(&(pThreadInfo->lock_sem), 0, 0);
if (ASYNC_MODE == g_Dbs.asyncMode) { if (ASYNC_MODE == g_Dbs.asyncMode) {
pthread_create(pids + i, NULL, asyncWrite, pThreadInfo); pthread_create(pids + i, NULL, asyncWrite, pThreadInfo);