From e9b3f5d35b75ff5f11ee08ae9ffeb59fa5d297e7 Mon Sep 17 00:00:00 2001 From: Shuduo Sang Date: Thu, 25 Mar 2021 21:35:13 +0800 Subject: [PATCH] =?UTF-8?q?[TD-3478]=20:=20increase=20thread=20num=20?= =?UTF-8?q?of=20table=20creation=20same=20as=20insert=E2=80=A6=20(#5577)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * [TD-3478] : increase thread num of table creation same as insertion. * [TD-3478] : increase thread num of table creation same as insertion. fix potential refer to derefed buffer. * [TD-3478] : increase thread num of table creation same as insertion. change rowsPerTbl to interlaceRows Co-authored-by: Shuduo Sang --- src/kit/taosdemo/taosdemo.c | 125 +++++++++++++++++++----------------- 1 file changed, 67 insertions(+), 58 deletions(-) diff --git a/src/kit/taosdemo/taosdemo.c b/src/kit/taosdemo/taosdemo.c index a981129bcd..2f1ff9656d 100644 --- a/src/kit/taosdemo/taosdemo.c +++ b/src/kit/taosdemo/taosdemo.c @@ -237,7 +237,7 @@ typedef struct SSuperTable_S { int childTblOffset; int multiThreadWriteOneTbl; // 0: no, 1: yes - int rowsPerTbl; // + int interlaceRows; // int disorderRatio; // 0: no disorder, >0: x% int disorderRange; // ms or us by database precision int maxSqlLen; // @@ -1199,8 +1199,8 @@ static int printfInsertMeta() { }else { printf(" multiThreadWriteOneTbl: \033[33myes\033[0m\n"); } - printf(" rowsPerTbl: \033[33m%d\033[0m\n", - g_Dbs.db[i].superTbls[j].rowsPerTbl); + printf(" interlaceRows: \033[33m%d\033[0m\n", + g_Dbs.db[i].superTbls[j].interlaceRows); printf(" disorderRange: \033[33m%d\033[0m\n", g_Dbs.db[i].superTbls[j].disorderRange); printf(" disorderRatio: \033[33m%d\033[0m\n", @@ -1361,7 +1361,7 @@ static void printfInsertMetaToFile(FILE* fp) { }else { fprintf(fp, " multiThreadWriteOneTbl: yes\n"); } - fprintf(fp, " rowsPerTbl: %d\n", g_Dbs.db[i].superTbls[j].rowsPerTbl); + fprintf(fp, " interlaceRows: %d\n", g_Dbs.db[i].superTbls[j].interlaceRows); fprintf(fp, " disorderRange: %d\n", g_Dbs.db[i].superTbls[j].disorderRange); fprintf(fp, " disorderRatio: %d\n", g_Dbs.db[i].superTbls[j].disorderRatio); fprintf(fp, " maxSqlLen: %d\n", g_Dbs.db[i].superTbls[j].maxSqlLen); @@ -2573,8 +2573,8 @@ static void* createTable(void *sarg) len = 0; verbosePrint("%s() LN%d %s\n", __func__, __LINE__, buffer); if (0 != queryDbExec(winfo->taos, buffer, NO_INSERT_TYPE)){ - free(buffer); errorPrint( "queryDbExec() failed. buffer:\n%s\n", buffer); + free(buffer); return NULL; } @@ -3061,9 +3061,10 @@ static bool getMetaFromInsertJsonFile(cJSON* root) { if (threads2 && threads2->type == cJSON_Number) { g_Dbs.threadCountByCreateTbl = threads2->valueint; } else if (!threads2) { - g_Dbs.threadCountByCreateTbl = 1; + g_Dbs.threadCountByCreateTbl = g_args.num_of_threads; } else { - printf("ERROR: failed to read json, threads2 not found\n"); + errorPrint("%s() LN%d, failed to read json, threads2 not found\n", + __func__, __LINE__); goto PARSE_OVER; } @@ -3578,13 +3579,15 @@ static bool getMetaFromInsertJsonFile(cJSON* root) { goto PARSE_OVER; } - cJSON* rowsPerTbl = cJSON_GetObjectItem(stbInfo, "interlace_rows"); - if (rowsPerTbl && rowsPerTbl->type == cJSON_Number) { - g_Dbs.db[i].superTbls[j].rowsPerTbl = rowsPerTbl->valueint; - } else if (!rowsPerTbl) { - g_Dbs.db[i].superTbls[j].rowsPerTbl = 0; // 0 means progressive mode, > 0 mean interlace mode. max value is less or equ num_of_records_per_req + cJSON* interlaceRows = cJSON_GetObjectItem(stbInfo, "interlace_rows"); + if (interlaceRows && interlaceRows->type == cJSON_Number) { + g_Dbs.db[i].superTbls[j].interlaceRows = interlaceRows->valueint; + } else if (!interlaceRows) { + g_Dbs.db[i].superTbls[j].interlaceRows = 0; // 0 means progressive mode, > 0 mean interlace mode. max value is less or equ num_of_records_per_req } else { - errorPrint("%s() LN%d, failed to read json, rowsPerTbl input mistake\n", __func__, __LINE__); + errorPrint( + "%s() LN%d, failed to read json, interlace rows input mistake\n", + __func__, __LINE__); goto PARSE_OVER; } @@ -4505,17 +4508,17 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) { int insertMode; char tableName[TSDB_TABLE_NAME_LEN]; - int rowsPerTbl = superTblInfo?superTblInfo->rowsPerTbl:g_args.interlace_rows; + int interlaceRows = superTblInfo?superTblInfo->interlaceRows:g_args.interlace_rows; - if (rowsPerTbl > 0) { + if (interlaceRows > 0) { insertMode = INTERLACE_INSERT_MODE; } else { insertMode = PROGRESSIVE_INSERT_MODE; } // rows per table need be less than insert batch - if (rowsPerTbl > g_args.num_of_RPR) - rowsPerTbl = g_args.num_of_RPR; + if (interlaceRows > g_args.num_of_RPR) + interlaceRows = g_args.num_of_RPR; pThreadInfo->totalInsertRows = 0; pThreadInfo->totalAffectedRows = 0; @@ -4543,13 +4546,13 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) { assert(pThreadInfo->ntables > 0); - if (rowsPerTbl > g_args.num_of_RPR) - rowsPerTbl = g_args.num_of_RPR; + if (interlaceRows > g_args.num_of_RPR) + interlaceRows = g_args.num_of_RPR; - batchPerTbl = rowsPerTbl; - if ((rowsPerTbl > 0) && (pThreadInfo->ntables > 1)) { + batchPerTbl = interlaceRows; + if ((interlaceRows > 0) && (pThreadInfo->ntables > 1)) { batchPerTblTimes = - (g_args.num_of_RPR / (rowsPerTbl * pThreadInfo->ntables)) + 1; + (g_args.num_of_RPR / (interlaceRows * pThreadInfo->ntables)) + 1; } else { batchPerTblTimes = 1; } @@ -4830,9 +4833,9 @@ static void* syncWrite(void *sarg) { threadInfo *winfo = (threadInfo *)sarg; SSuperTable* superTblInfo = winfo->superTblInfo; - int rowsPerTbl = superTblInfo?superTblInfo->rowsPerTbl:g_args.interlace_rows; + int interlaceRows = superTblInfo?superTblInfo->interlaceRows:g_args.interlace_rows; - if (rowsPerTbl > 0) { + if (interlaceRows > 0) { // interlace mode return syncWriteInterlace(winfo); } else { @@ -4974,7 +4977,7 @@ static void startMultiThreadInsertData(int threads, char* db_name, } else if (0 == strncasecmp(precision, "us", 2)) { timePrec = TSDB_TIME_PRECISION_MICRO; } else { - errorPrint( "No support precision: %s\n", precision); + errorPrint("Not support precision: %s\n", precision); exit(-1); } } @@ -5009,7 +5012,8 @@ static void startMultiThreadInsertData(int threads, char* db_name, if ((superTblInfo) && (0 == strncasecmp(superTblInfo->dataSource, "sample", strlen("sample")))) { if (0 != prepareSampleDataForSTable(superTblInfo)) { - errorPrint("%s() LN%d, prepare sample data for stable failed!\n", __func__, __LINE__); + errorPrint("%s() LN%d, prepare sample data for stable failed!\n", + __func__, __LINE__); exit(-1); } } @@ -5018,7 +5022,8 @@ static void startMultiThreadInsertData(int threads, char* db_name, if ((superTblInfo) && (0 == strncasecmp(superTblInfo->dataSource, "sample", strlen("sample")))) { if (0 != prepareSampleDataForSTable(superTblInfo)) { - errorPrint("%s() LN%d, prepare sample data for stable failed!\n", __func__, __LINE__); + errorPrint("%s() LN%d, prepare sample data for stable failed!\n", + __func__, __LINE__); exit(-1); } } @@ -5078,7 +5083,8 @@ static void startMultiThreadInsertData(int threads, char* db_name, g_Dbs.host, g_Dbs.user, g_Dbs.password, db_name, g_Dbs.port); if (NULL == t_info->taos) { - errorPrint( "connect to server fail from insert sub thread, reason: %s\n", + errorPrint( + "connect to server fail from insert sub thread, reason: %s\n", taos_errstr(NULL)); exit(-1); } @@ -5374,10 +5380,10 @@ static int insertTestProcess() { if (g_totalChildTables > 0) { printf("Spent %.4f seconds to create %d tables with %d thread(s)\n\n", - end - start, g_totalChildTables, g_Dbs.threadCount); + end - start, g_totalChildTables, g_Dbs.threadCountByCreateTbl); fprintf(g_fpOfInsertResult, "Spent %.4f seconds to create %d tables with %d thread(s)\n\n", - end - start, g_totalChildTables, g_Dbs.threadCount); + end - start, g_totalChildTables, g_Dbs.threadCountByCreateTbl); } taosMsleep(1000); @@ -5546,7 +5552,8 @@ static int queryTestProcess() { NULL, g_queryInfo.port); if (taos == NULL) { - errorPrint( "Failed to connect to TDengine, reason:%s\n", taos_errstr(NULL)); + errorPrint( "Failed to connect to TDengine, reason:%s\n", + taos_errstr(NULL)); exit(-1); } @@ -5867,7 +5874,8 @@ static int subscribeTestProcess() { g_queryInfo.dbName, g_queryInfo.port); if (taos == NULL) { - errorPrint( "Failed to connect to TDengine, reason:%s\n", taos_errstr(NULL)); + errorPrint( "Failed to connect to TDengine, reason:%s\n", + taos_errstr(NULL)); exit(-1); } @@ -5998,15 +6006,15 @@ static void setParaFromArg(){ if (g_args.user) { strcpy(g_Dbs.user, g_args.user); - } + } if (g_args.password) { strcpy(g_Dbs.password, g_args.password); - } - + } + if (g_args.port) { g_Dbs.port = g_args.port; - } + } g_Dbs.threadCount = g_args.num_of_threads; g_Dbs.threadCountByCreateTbl = g_args.num_of_threads; @@ -6027,11 +6035,11 @@ static void setParaFromArg(){ char dataString[STRING_LEN]; char **data_type = g_args.datatype; - + memset(dataString, 0, STRING_LEN); - if (strcasecmp(data_type[0], "BINARY") == 0 - || strcasecmp(data_type[0], "BOOL") == 0 + if (strcasecmp(data_type[0], "BINARY") == 0 + || strcasecmp(data_type[0], "BOOL") == 0 || strcasecmp(data_type[0], "NCHAR") == 0 ) { g_Dbs.do_aggreFunc = false; } @@ -6041,7 +6049,7 @@ static void setParaFromArg(){ tstrncpy(g_Dbs.db[0].superTbls[0].sTblName, "meters", MAX_TB_NAME_SIZE); g_Dbs.db[0].superTbls[0].childTblCount = g_args.num_of_tables; g_Dbs.threadCount = g_args.num_of_threads; - g_Dbs.threadCountByCreateTbl = 1; + g_Dbs.threadCountByCreateTbl = g_args.num_of_threads; g_Dbs.queryMode = g_args.mode; g_Dbs.db[0].superTbls[0].autoCreateTable = PRE_CREATE_SUBTBL; @@ -6049,14 +6057,14 @@ static void setParaFromArg(){ g_Dbs.db[0].superTbls[0].childTblExists = TBL_NO_EXISTS; g_Dbs.db[0].superTbls[0].disorderRange = g_args.disorderRange; g_Dbs.db[0].superTbls[0].disorderRatio = g_args.disorderRatio; - tstrncpy(g_Dbs.db[0].superTbls[0].childTblPrefix, + tstrncpy(g_Dbs.db[0].superTbls[0].childTblPrefix, g_args.tb_prefix, MAX_TB_NAME_SIZE); tstrncpy(g_Dbs.db[0].superTbls[0].dataSource, "rand", MAX_TB_NAME_SIZE); tstrncpy(g_Dbs.db[0].superTbls[0].insertMode, "taosc", MAX_TB_NAME_SIZE); - tstrncpy(g_Dbs.db[0].superTbls[0].startTimestamp, + tstrncpy(g_Dbs.db[0].superTbls[0].startTimestamp, "2017-07-14 10:40:00.000", MAX_TB_NAME_SIZE); g_Dbs.db[0].superTbls[0].timeStampStep = DEFAULT_TIMESTAMP_STEP; - + g_Dbs.db[0].superTbls[0].insertRows = g_args.num_of_DPT; g_Dbs.db[0].superTbls[0].maxSqlLen = TSDB_PAYLOAD_SIZE; @@ -6066,31 +6074,31 @@ static void setParaFromArg(){ break; } - tstrncpy(g_Dbs.db[0].superTbls[0].columns[i].dataType, + tstrncpy(g_Dbs.db[0].superTbls[0].columns[i].dataType, data_type[i], MAX_TB_NAME_SIZE); - g_Dbs.db[0].superTbls[0].columns[i].dataLen = g_args.len_of_binary; + g_Dbs.db[0].superTbls[0].columns[i].dataLen = g_args.len_of_binary; g_Dbs.db[0].superTbls[0].columnCount++; } - + if (g_Dbs.db[0].superTbls[0].columnCount > g_args.num_of_CPR) { g_Dbs.db[0].superTbls[0].columnCount = g_args.num_of_CPR; } else { for (int i = g_Dbs.db[0].superTbls[0].columnCount; i < g_args.num_of_CPR; i++) { tstrncpy(g_Dbs.db[0].superTbls[0].columns[i].dataType, "INT", MAX_TB_NAME_SIZE); - g_Dbs.db[0].superTbls[0].columns[i].dataLen = 0; + g_Dbs.db[0].superTbls[0].columns[i].dataLen = 0; g_Dbs.db[0].superTbls[0].columnCount++; } } tstrncpy(g_Dbs.db[0].superTbls[0].tags[0].dataType, "INT", MAX_TB_NAME_SIZE); - g_Dbs.db[0].superTbls[0].tags[0].dataLen = 0; + g_Dbs.db[0].superTbls[0].tags[0].dataLen = 0; tstrncpy(g_Dbs.db[0].superTbls[0].tags[1].dataType, "BINARY", MAX_TB_NAME_SIZE); - g_Dbs.db[0].superTbls[0].tags[1].dataLen = g_args.len_of_binary; - g_Dbs.db[0].superTbls[0].tagCount = 2; + g_Dbs.db[0].superTbls[0].tags[1].dataLen = g_args.len_of_binary; + g_Dbs.db[0].superTbls[0].tagCount = 2; } else { - g_Dbs.threadCountByCreateTbl = 1; - g_Dbs.db[0].superTbls[0].tagCount = 0; + g_Dbs.threadCountByCreateTbl = g_args.num_of_threads; + g_Dbs.db[0].superTbls[0].tagCount = 0; } } @@ -6222,7 +6230,7 @@ static void queryResult() { rInfo->ntables = g_Dbs.db[0].superTbls[0].childTblCount; rInfo->end_table_to = g_Dbs.db[0].superTbls[0].childTblCount - 1; rInfo->superTblInfo = &g_Dbs.db[0].superTbls[0]; - strcpy(rInfo->tb_prefix, + strcpy(rInfo->tb_prefix, g_Dbs.db[0].superTbls[0].childTblPrefix); } else { rInfo->ntables = g_args.num_of_tables; @@ -6231,13 +6239,14 @@ static void queryResult() { } rInfo->taos = taos_connect( - g_Dbs.host, - g_Dbs.user, - g_Dbs.password, - g_Dbs.db[0].dbName, + g_Dbs.host, + g_Dbs.user, + g_Dbs.password, + g_Dbs.db[0].dbName, g_Dbs.port); if (rInfo->taos == NULL) { - errorPrint( "Failed to connect to TDengine, reason:%s\n", taos_errstr(NULL)); + errorPrint( "Failed to connect to TDengine, reason:%s\n", + taos_errstr(NULL)); free(rInfo); exit(-1); } -- GitLab