diff --git a/src/kit/taosdemo/insert-interlace.json b/src/kit/taosdemo/insert-interlace.json index 344db4fd00fb1fd6616915f7ed1e71b5ee7368d8..cf3e1de2f4a76f5cc242399b9a268c95c2dca878 100644 --- a/src/kit/taosdemo/insert-interlace.json +++ b/src/kit/taosdemo/insert-interlace.json @@ -40,7 +40,6 @@ "data_source": "rand", "insert_mode": "taosc", "insert_rows": 1000, - "multi_thread_write_one_tbl": "no", "interlace_rows": 20, "max_sql_len": 1024000, "disorder_ratio": 0, diff --git a/src/kit/taosdemo/insert.json b/src/kit/taosdemo/insert.json index f0e3ab1d50dd6cd268568fa7ed81c6ff0a2351c2..5f152a2d0c257931db4a5c69c007e524247f8fb8 100644 --- a/src/kit/taosdemo/insert.json +++ b/src/kit/taosdemo/insert.json @@ -40,7 +40,6 @@ "data_source": "rand", "insert_mode": "taosc", "insert_rows": 100000, - "multi_thread_write_one_tbl": "no", "interlace_rows": 0, "max_sql_len": 1024000, "disorder_ratio": 0, diff --git a/src/kit/taosdemo/query.json b/src/kit/taosdemo/query.json index 33ac120bdae283f64c4419d2885a93d36a7c8093..d84f997c329f005e62642ac32856b8face1c8048 100644 --- a/src/kit/taosdemo/query.json +++ b/src/kit/taosdemo/query.json @@ -1,5 +1,5 @@ { - "filetype":"query", + "filetype": "query", "cfgdir": "/etc/taos", "host": "127.0.0.1", "port": 6030, @@ -7,13 +7,30 @@ "password": "taosdata", "confirm_parameter_prompt": "yes", "databases": "dbx", - "specified_table_query": - {"query_interval":1, "concurrent":4, - "sqls": [{"sql": "select last_row(*) from stb where color='red'", "result": "./query_res0.txt"}, - {"sql": "select count(*) from stb_01", "result": "./query_res1.txt"}] - }, - "super_table_query": - {"stblname": "stb", "query_interval":1, "threads":4, - "sqls": [{"sql": "select last_row(*) from xxxx", "result": "./query_res2.txt"}] - } + "query_times": 1, + "specified_table_query": { + "query_interval": 1, + "concurrent": 4, + "sqls": [ + { + "sql": "select last_row(*) from stb where color='red'", + "result": "./query_res0.txt" + }, + { + "sql": "select count(*) from stb_01", + "result": "./query_res1.txt" + } + ] + }, + "super_table_query": { + "stblname": "stb", + "query_interval": 1, + "threads": 4, + "sqls": [ + { + "sql": "select last_row(*) from xxxx", + "result": "./query_res2.txt" + } + ] + } } diff --git a/src/kit/taosdemo/taosdemo.c b/src/kit/taosdemo/taosdemo.c index 815a842d60b3c5ec3633db205e842fbd9a4e7eb3..dd2891fd329e3ea1d1cd2fdc14859e62fb938639 100644 --- a/src/kit/taosdemo/taosdemo.c +++ b/src/kit/taosdemo/taosdemo.c @@ -366,6 +366,7 @@ typedef struct SpecifiedQueryInfo_S { char sql[MAX_QUERY_SQL_COUNT][MAX_QUERY_SQL_LENGTH+1]; char result[MAX_QUERY_SQL_COUNT][MAX_FILE_NAME_LEN+1]; TAOS_SUB* tsub[MAX_QUERY_SQL_COUNT]; + int totalQueried; } SpecifiedQueryInfo; typedef struct SuperQueryInfo_S { @@ -385,6 +386,7 @@ typedef struct SuperQueryInfo_S { TAOS_SUB* tsub[MAX_QUERY_SQL_COUNT]; char* childTblName; + int totalQueried; } SuperQueryInfo; typedef struct SQueryMetaInfo_S { @@ -398,6 +400,7 @@ typedef struct SQueryMetaInfo_S { SpecifiedQueryInfo specifiedQueryInfo; SuperQueryInfo superQueryInfo; + int totalQueried; } SQueryMetaInfo; typedef struct SThreadInfo_S { @@ -2602,8 +2605,8 @@ static int createDatabasesAndStables() { static void* createTable(void *sarg) { - threadInfo *winfo = (threadInfo *)sarg; - SSuperTable* superTblInfo = winfo->superTblInfo; + threadInfo *pThreadInfo = (threadInfo *)sarg; + SSuperTable* superTblInfo = pThreadInfo->superTblInfo; int64_t lastPrintTime = taosGetTimestampMs(); @@ -2621,15 +2624,15 @@ static void* createTable(void *sarg) verbosePrint("%s() LN%d: Creating table from %d to %d\n", __func__, __LINE__, - winfo->start_table_from, winfo->end_table_to); + pThreadInfo->start_table_from, pThreadInfo->end_table_to); - for (int i = winfo->start_table_from; i <= winfo->end_table_to; i++) { + for (int i = pThreadInfo->start_table_from; i <= pThreadInfo->end_table_to; i++) { if (0 == g_Dbs.use_metric) { snprintf(buffer, buff_len, "create table if not exists %s.%s%d %s;", - winfo->db_name, + pThreadInfo->db_name, g_args.tb_prefix, i, - winfo->cols); + pThreadInfo->cols); } else { if (superTblInfo == NULL) { errorPrint("%s() LN%d, use metric, but super table info is NULL\n", @@ -2658,8 +2661,8 @@ static void* createTable(void *sarg) len += snprintf(buffer + len, buff_len - len, "if not exists %s.%s%d using %s.%s tags %s ", - winfo->db_name, superTblInfo->childTblPrefix, - i, winfo->db_name, + pThreadInfo->db_name, superTblInfo->childTblPrefix, + i, pThreadInfo->db_name, superTblInfo->sTblName, tagsValBuf); free(tagsValBuf); batchNum++; @@ -2673,7 +2676,7 @@ 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, false)){ + if (0 != queryDbExec(pThreadInfo->taos, buffer, NO_INSERT_TYPE, false)){ errorPrint( "queryDbExec() failed. buffer:\n%s\n", buffer); free(buffer); return NULL; @@ -2682,14 +2685,14 @@ static void* createTable(void *sarg) int64_t currentPrintTime = taosGetTimestampMs(); if (currentPrintTime - lastPrintTime > 30*1000) { printf("thread[%d] already create %d - %d tables\n", - winfo->threadID, winfo->start_table_from, i); + pThreadInfo->threadID, pThreadInfo->start_table_from, i); lastPrintTime = currentPrintTime; } } if (0 != len) { verbosePrint("%s() %d buffer: %s\n", __func__, __LINE__, buffer); - if (0 != queryDbExec(winfo->taos, buffer, NO_INSERT_TYPE, false)) { + if (0 != queryDbExec(pThreadInfo->taos, buffer, NO_INSERT_TYPE, false)) { errorPrint( "queryDbExec() failed. buffer:\n%s\n", buffer); } } @@ -3694,7 +3697,7 @@ static bool getMetaFromInsertJsonFile(cJSON* root) { __func__, __LINE__); goto PARSE_OVER; } - +/* cJSON *multiThreadWriteOneTbl = cJSON_GetObjectItem(stbInfo, "multi_thread_write_one_tbl"); // no , yes if (multiThreadWriteOneTbl @@ -3711,7 +3714,7 @@ static bool getMetaFromInsertJsonFile(cJSON* root) { printf("ERROR: failed to read json, multiThreadWriteOneTbl not found\n"); goto PARSE_OVER; } - +*/ cJSON* interlaceRows = cJSON_GetObjectItem(stbInfo, "interlace_rows"); if (interlaceRows && interlaceRows->type == cJSON_Number) { g_Dbs.db[i].superTbls[j].interlaceRows = interlaceRows->valueint; @@ -5159,45 +5162,45 @@ free_and_statistics_2: static void* syncWrite(void *sarg) { - threadInfo *winfo = (threadInfo *)sarg; - SSuperTable* superTblInfo = winfo->superTblInfo; + threadInfo *pThreadInfo = (threadInfo *)sarg; + SSuperTable* superTblInfo = pThreadInfo->superTblInfo; int interlaceRows = superTblInfo?superTblInfo->interlaceRows:g_args.interlace_rows; if (interlaceRows > 0) { // interlace mode - return syncWriteInterlace(winfo); + return syncWriteInterlace(pThreadInfo); } else { // progressive mode - return syncWriteProgressive(winfo); + return syncWriteProgressive(pThreadInfo); } } static void callBack(void *param, TAOS_RES *res, int code) { - threadInfo* winfo = (threadInfo*)param; - SSuperTable* superTblInfo = winfo->superTblInfo; + threadInfo* pThreadInfo = (threadInfo*)param; + SSuperTable* superTblInfo = pThreadInfo->superTblInfo; int insert_interval = superTblInfo?superTblInfo->insertInterval:g_args.insert_interval; if (insert_interval) { - winfo->et = taosGetTimestampUs(); - if (((winfo->et - winfo->st)/1000) < insert_interval) { - taosMsleep(insert_interval - (winfo->et - winfo->st)/1000); // ms + pThreadInfo->et = taosGetTimestampUs(); + if (((pThreadInfo->et - pThreadInfo->st)/1000) < insert_interval) { + taosMsleep(insert_interval - (pThreadInfo->et - pThreadInfo->st)/1000); // ms } } - char *buffer = calloc(1, winfo->superTblInfo->maxSqlLen); + char *buffer = calloc(1, pThreadInfo->superTblInfo->maxSqlLen); char data[MAX_DATA_SIZE]; char *pstr = buffer; - pstr += sprintf(pstr, "insert into %s.%s%d values", winfo->db_name, winfo->tb_prefix, - winfo->start_table_from); -// if (winfo->counter >= winfo->superTblInfo->insertRows) { - if (winfo->counter >= g_args.num_of_RPR) { - winfo->start_table_from++; - winfo->counter = 0; - } - if (winfo->start_table_from > winfo->end_table_to) { - tsem_post(&winfo->lock_sem); + pstr += sprintf(pstr, "insert into %s.%s%d values", pThreadInfo->db_name, pThreadInfo->tb_prefix, + pThreadInfo->start_table_from); +// if (pThreadInfo->counter >= pThreadInfo->superTblInfo->insertRows) { + if (pThreadInfo->counter >= g_args.num_of_RPR) { + pThreadInfo->start_table_from++; + pThreadInfo->counter = 0; + } + if (pThreadInfo->start_table_from > pThreadInfo->end_table_to) { + tsem_post(&pThreadInfo->lock_sem); free(buffer); taos_free_result(res); return; @@ -5205,46 +5208,46 @@ static void callBack(void *param, TAOS_RES *res, int code) { for (int i = 0; i < g_args.num_of_RPR; i++) { int rand_num = taosRandom() % 100; - if (0 != winfo->superTblInfo->disorderRatio - && rand_num < winfo->superTblInfo->disorderRatio) { - int64_t d = winfo->lastTs - (taosRandom() % winfo->superTblInfo->disorderRange + 1); - generateRowData(data, d, winfo->superTblInfo); + if (0 != pThreadInfo->superTblInfo->disorderRatio + && rand_num < pThreadInfo->superTblInfo->disorderRatio) { + int64_t d = pThreadInfo->lastTs - (taosRandom() % pThreadInfo->superTblInfo->disorderRange + 1); + generateRowData(data, d, pThreadInfo->superTblInfo); } else { - generateRowData(data, winfo->lastTs += 1000, winfo->superTblInfo); + generateRowData(data, pThreadInfo->lastTs += 1000, pThreadInfo->superTblInfo); } pstr += sprintf(pstr, "%s", data); - winfo->counter++; + pThreadInfo->counter++; - if (winfo->counter >= winfo->superTblInfo->insertRows) { + if (pThreadInfo->counter >= pThreadInfo->superTblInfo->insertRows) { break; } } if (insert_interval) { - winfo->st = taosGetTimestampUs(); + pThreadInfo->st = taosGetTimestampUs(); } - taos_query_a(winfo->taos, buffer, callBack, winfo); + taos_query_a(pThreadInfo->taos, buffer, callBack, pThreadInfo); free(buffer); taos_free_result(res); } static void *asyncWrite(void *sarg) { - threadInfo *winfo = (threadInfo *)sarg; - SSuperTable* superTblInfo = winfo->superTblInfo; + threadInfo *pThreadInfo = (threadInfo *)sarg; + SSuperTable* superTblInfo = pThreadInfo->superTblInfo; - winfo->st = 0; - winfo->et = 0; - winfo->lastTs = winfo->start_time; + pThreadInfo->st = 0; + pThreadInfo->et = 0; + pThreadInfo->lastTs = pThreadInfo->start_time; int insert_interval = superTblInfo?superTblInfo->insertInterval:g_args.insert_interval; if (insert_interval) { - winfo->st = taosGetTimestampUs(); + pThreadInfo->st = taosGetTimestampUs(); } - taos_query_a(winfo->taos, "show databases", callBack, winfo); + taos_query_a(pThreadInfo->taos, "show databases", callBack, pThreadInfo); - tsem_wait(&(winfo->lock_sem)); + tsem_wait(&(pThreadInfo->lock_sem)); return NULL; } @@ -5774,10 +5777,10 @@ static int insertTestProcess() { return 0; } -static void *specifiedQueryProcess(void *sarg) { - threadInfo *winfo = (threadInfo *)sarg; +static void *specifiedTableQuery(void *sarg) { + threadInfo *pThreadInfo = (threadInfo *)sarg; - if (winfo->taos == NULL) { + if (pThreadInfo->taos == NULL) { TAOS * taos = NULL; taos = taos_connect(g_queryInfo.host, g_queryInfo.user, @@ -5786,17 +5789,17 @@ static void *specifiedQueryProcess(void *sarg) { g_queryInfo.port); if (taos == NULL) { errorPrint("[%d] Failed to connect to TDengine, reason:%s\n", - winfo->threadID, taos_errstr(NULL)); + pThreadInfo->threadID, taos_errstr(NULL)); return NULL; } else { - winfo->taos = taos; + pThreadInfo->taos = taos; } } char sqlStr[MAX_DB_NAME_SIZE + 5]; sprintf(sqlStr, "use %s", g_queryInfo.dbName); - if (0 != queryDbExec(winfo->taos, sqlStr, NO_INSERT_TYPE, false)) { - taos_close(winfo->taos); + if (0 != queryDbExec(pThreadInfo->taos, sqlStr, NO_INSERT_TYPE, false)) { + taos_close(pThreadInfo->taos); errorPrint( "use database %s failed!\n\n", g_queryInfo.dbName); return NULL; @@ -5807,11 +5810,15 @@ static void *specifiedQueryProcess(void *sarg) { int queryTimes = g_queryInfo.specifiedQueryInfo.queryTimes; + int totalQueried = 0; + int64_t lastPrintTime = taosGetTimestampMs(); + int64_t startTs = taosGetTimestampMs(); + while(queryTimes --) { if (g_queryInfo.specifiedQueryInfo.rate && (et - st) < (int64_t)g_queryInfo.specifiedQueryInfo.rate*1000) { taosMsleep(g_queryInfo.specifiedQueryInfo.rate*1000 - (et - st)); // ms - //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, winfo->start_table_from, winfo->end_table_to); + //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, pThreadInfo->start_table_from, pThreadInfo->end_table_to); } st = taosGetTimestampUs(); @@ -5819,13 +5826,13 @@ static void *specifiedQueryProcess(void *sarg) { if (0 == strncasecmp(g_queryInfo.queryMode, "taosc", 5)) { int64_t t1 = taosGetTimestampUs(); char tmpFile[MAX_FILE_NAME_LEN*2] = {0}; - if (g_queryInfo.specifiedQueryInfo.result[winfo->querySeq][0] != 0) { + if (g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq][0] != 0) { sprintf(tmpFile, "%s-%d", - g_queryInfo.specifiedQueryInfo.result[winfo->querySeq], - winfo->threadID); + g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq], + pThreadInfo->threadID); } - selectAndGetResult(winfo->taos, - g_queryInfo.specifiedQueryInfo.sql[winfo->querySeq], tmpFile); + selectAndGetResult(pThreadInfo->taos, + g_queryInfo.specifiedQueryInfo.sql[pThreadInfo->querySeq], tmpFile); int64_t t2 = taosGetTimestampUs(); printf("=[taosc] thread[%"PRId64"] complete one sql, Spent %f s\n", taosGetSelfPthreadId(), (t2 - t1)/1000000.0); @@ -5833,25 +5840,37 @@ static void *specifiedQueryProcess(void *sarg) { int64_t t1 = taosGetTimestampUs(); int retCode = postProceSql(g_queryInfo.host, g_queryInfo.port, - g_queryInfo.specifiedQueryInfo.sql[winfo->querySeq]); + g_queryInfo.specifiedQueryInfo.sql[pThreadInfo->querySeq]); + if (0 != retCode) { + printf("====restful return fail, threadID[%d]\n", pThreadInfo->threadID); + return NULL; + } int64_t t2 = taosGetTimestampUs(); printf("=[restful] thread[%"PRId64"] complete one sql, Spent %f s\n", taosGetSelfPthreadId(), (t2 - t1)/1000000.0); - if (0 != retCode) { - printf("====restful return fail, threadID[%d]\n", winfo->threadID); - return NULL; - } } + totalQueried ++; + g_queryInfo.specifiedQueryInfo.totalQueried ++; et = taosGetTimestampUs(); printf("==thread[%"PRId64"] complete all sqls to specify tables once queries duration:%.6fs\n\n", taosGetSelfPthreadId(), (double)(et - st)/1000.0); + + int64_t currentPrintTime = taosGetTimestampMs(); + int64_t endTs = taosGetTimestampMs(); + if (currentPrintTime - lastPrintTime > 30*1000) { + printf("thread[%d] has currently completed queries: %d, QPS: %10.2f\n", + pThreadInfo->threadID, + totalQueried, + totalQueried/((endTs-startTs)/1000.0)); + } + lastPrintTime = currentPrintTime; } return NULL; } -static void replaceSubTblName(char* inSql, char* outSql, int tblIndex) { +static void replaceChildTblName(char* inSql, char* outSql, int tblIndex) { char sourceString[32] = "xxxx"; char subTblName[MAX_TB_NAME_SIZE*3]; sprintf(subTblName, "%s.%s", @@ -5873,11 +5892,11 @@ static void replaceSubTblName(char* inSql, char* outSql, int tblIndex) { //printf("3: %s\n", outSql); } -static void *superQueryProcess(void *sarg) { +static void *superTableQuery(void *sarg) { char sqlstr[1024]; - threadInfo *winfo = (threadInfo *)sarg; + threadInfo *pThreadInfo = (threadInfo *)sarg; - if (winfo->taos == NULL) { + if (pThreadInfo->taos == NULL) { TAOS * taos = NULL; taos = taos_connect(g_queryInfo.host, g_queryInfo.user, @@ -5886,10 +5905,10 @@ static void *superQueryProcess(void *sarg) { g_queryInfo.port); if (taos == NULL) { errorPrint("[%d] Failed to connect to TDengine, reason:%s\n", - winfo->threadID, taos_errstr(NULL)); + pThreadInfo->threadID, taos_errstr(NULL)); return NULL; } else { - winfo->taos = taos; + pThreadInfo->taos = taos; } } @@ -5897,33 +5916,49 @@ static void *superQueryProcess(void *sarg) { int64_t et = (int64_t)g_queryInfo.superQueryInfo.rate*1000; int queryTimes = g_queryInfo.superQueryInfo.queryTimes; + int totalQueried = 0; + int64_t startTs = taosGetTimestampMs(); + int64_t lastPrintTime = taosGetTimestampMs(); while(queryTimes --) { if (g_queryInfo.superQueryInfo.rate && (et - st) < (int64_t)g_queryInfo.superQueryInfo.rate*1000) { taosMsleep(g_queryInfo.superQueryInfo.rate*1000 - (et - st)); // ms - //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, winfo->start_table_from, winfo->end_table_to); + //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, pThreadInfo->start_table_from, pThreadInfo->end_table_to); } st = taosGetTimestampUs(); - for (int i = winfo->start_table_from; i <= winfo->end_table_to; i++) { + for (int i = pThreadInfo->start_table_from; i <= pThreadInfo->end_table_to; i++) { for (int j = 0; j < g_queryInfo.superQueryInfo.sqlCount; j++) { memset(sqlstr,0,sizeof(sqlstr)); - replaceSubTblName(g_queryInfo.superQueryInfo.sql[j], sqlstr, i); + replaceChildTblName(g_queryInfo.superQueryInfo.sql[j], sqlstr, i); char tmpFile[MAX_FILE_NAME_LEN*2] = {0}; if (g_queryInfo.superQueryInfo.result[j][0] != 0) { sprintf(tmpFile, "%s-%d", g_queryInfo.superQueryInfo.result[j], - winfo->threadID); + pThreadInfo->threadID); } - selectAndGetResult(winfo->taos, sqlstr, tmpFile); + selectAndGetResult(pThreadInfo->taos, sqlstr, tmpFile); + + totalQueried++; + g_queryInfo.superQueryInfo.totalQueried ++; + + int64_t currentPrintTime = taosGetTimestampMs(); + int64_t endTs = taosGetTimestampMs(); + if (currentPrintTime - lastPrintTime > 30*1000) { + printf("thread[%d] has currently completed queries: %d, QPS: %10.2f\n", + pThreadInfo->threadID, + totalQueried, + totalQueried/((endTs-startTs)/1000.0)); + } + lastPrintTime = currentPrintTime; } } et = taosGetTimestampUs(); printf("####thread[%"PRId64"] complete all sqls to allocate all sub-tables[%d - %d] once queries duration:%.4fs\n\n", taosGetSelfPthreadId(), - winfo->start_table_from, - winfo->end_table_to, + pThreadInfo->start_table_from, + pThreadInfo->end_table_to, (double)(et - st)/1000000.0); } @@ -5969,6 +6004,8 @@ static int queryTestProcess() { int nConcurrent = g_queryInfo.specifiedQueryInfo.concurrent; int nSqlCount = g_queryInfo.specifiedQueryInfo.sqlCount; + int64_t startTs = taosGetTimestampMs(); + if ((nSqlCount > 0) && (nConcurrent > 0)) { pids = malloc(nConcurrent * nSqlCount * sizeof(pthread_t)); @@ -6002,7 +6039,7 @@ static int queryTestProcess() { t_info->taos = NULL;// TODO: workaround to use separate taos connection; - pthread_create(pids + i * nSqlCount + j, NULL, specifiedQueryProcess, + pthread_create(pids + i * nSqlCount + j, NULL, specifiedTableQuery, t_info); } } @@ -6051,7 +6088,7 @@ static int queryTestProcess() { t_info->end_table_to = i < b ? startFrom + a : startFrom + a - 1; startFrom = t_info->end_table_to + 1; t_info->taos = NULL; // TODO: workaround to use separate taos connection; - pthread_create(pidsOfSub + i, NULL, superQueryProcess, t_info); + pthread_create(pidsOfSub + i, NULL, superTableQuery, t_info); } g_queryInfo.superQueryInfo.threadCnt = threads; @@ -6078,6 +6115,14 @@ static int queryTestProcess() { tmfree((char*)infosOfSub); // taos_close(taos);// TODO: workaround to use separate taos connection; + int64_t endTs = taosGetTimestampMs(); + + int totalQueried = g_queryInfo.specifiedQueryInfo.totalQueried + + g_queryInfo.superQueryInfo.totalQueried; + + printf("==== completed total queries: %d, the QPS of all threads: %10.2f====\n", + totalQueried, + totalQueried/((endTs-startTs)/1000.0)); return 0; } @@ -6114,12 +6159,12 @@ static TAOS_SUB* subscribeImpl(TAOS *taos, char *sql, char* topic, char* resultF return tsub; } -static void *subSubscribeProcess(void *sarg) { - threadInfo *winfo = (threadInfo *)sarg; +static void *superSubscribe(void *sarg) { + threadInfo *pThreadInfo = (threadInfo *)sarg; char subSqlstr[1024]; TAOS_SUB* tsub[MAX_QUERY_SQL_COUNT] = {0}; - if (winfo->taos == NULL) { + if (pThreadInfo->taos == NULL) { TAOS * taos = NULL; taos = taos_connect(g_queryInfo.host, g_queryInfo.user, @@ -6128,17 +6173,17 @@ static void *subSubscribeProcess(void *sarg) { g_queryInfo.port); if (taos == NULL) { errorPrint("[%d] Failed to connect to TDengine, reason:%s\n", - winfo->threadID, taos_errstr(NULL)); + pThreadInfo->threadID, taos_errstr(NULL)); return NULL; } else { - winfo->taos = taos; + pThreadInfo->taos = taos; } } char sqlStr[MAX_TB_NAME_SIZE*2]; sprintf(sqlStr, "use %s", g_queryInfo.dbName); - if (0 != queryDbExec(winfo->taos, sqlStr, NO_INSERT_TYPE, false)) { - taos_close(winfo->taos); + if (0 != queryDbExec(pThreadInfo->taos, sqlStr, NO_INSERT_TYPE, false)) { + taos_close(pThreadInfo->taos); errorPrint( "use database %s failed!\n\n", g_queryInfo.dbName); return NULL; @@ -6149,7 +6194,7 @@ static void *subSubscribeProcess(void *sarg) { do { //if (g_queryInfo.specifiedQueryInfo.rate && (et - st) < g_queryInfo.specifiedQueryInfo.rate*1000) { // taosMsleep(g_queryInfo.specifiedQueryInfo.rate*1000 - (et - st)); // ms - // //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, winfo->start_table_from, winfo->end_table_to); + // //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, pThreadInfo->start_table_from, pThreadInfo->end_table_to); //} //st = taosGetTimestampMs(); @@ -6157,15 +6202,15 @@ static void *subSubscribeProcess(void *sarg) { for (int i = 0; i < g_queryInfo.superQueryInfo.sqlCount; i++) { sprintf(topic, "taosdemo-subscribe-%d", i); memset(subSqlstr,0,sizeof(subSqlstr)); - replaceSubTblName(g_queryInfo.superQueryInfo.sql[i], subSqlstr, i); + replaceChildTblName(g_queryInfo.superQueryInfo.sql[i], subSqlstr, i); char tmpFile[MAX_FILE_NAME_LEN*2] = {0}; if (g_queryInfo.superQueryInfo.result[i][0] != 0) { sprintf(tmpFile, "%s-%d", - g_queryInfo.superQueryInfo.result[i], winfo->threadID); + g_queryInfo.superQueryInfo.result[i], pThreadInfo->threadID); } - tsub[i] = subscribeImpl(winfo->taos, subSqlstr, topic, tmpFile); + tsub[i] = subscribeImpl(pThreadInfo->taos, subSqlstr, topic, tmpFile); if (NULL == tsub[i]) { - taos_close(winfo->taos); + taos_close(pThreadInfo->taos); return NULL; } } @@ -6187,7 +6232,7 @@ static void *subSubscribeProcess(void *sarg) { if (g_queryInfo.superQueryInfo.result[i][0] != 0) { sprintf(tmpFile, "%s-%d", g_queryInfo.superQueryInfo.result[i], - winfo->threadID); + pThreadInfo->threadID); } getResult(res, tmpFile); } @@ -6199,15 +6244,15 @@ static void *subSubscribeProcess(void *sarg) { taos_unsubscribe(tsub[i], g_queryInfo.superQueryInfo.subscribeKeepProgress); } - taos_close(winfo->taos); + taos_close(pThreadInfo->taos); return NULL; } -static void *superSubscribeProcess(void *sarg) { - threadInfo *winfo = (threadInfo *)sarg; +static void *specifiedSubscribe(void *sarg) { + threadInfo *pThreadInfo = (threadInfo *)sarg; TAOS_SUB* tsub[MAX_QUERY_SQL_COUNT] = {0}; - if (winfo->taos == NULL) { + if (pThreadInfo->taos == NULL) { TAOS * taos = NULL; taos = taos_connect(g_queryInfo.host, g_queryInfo.user, @@ -6216,18 +6261,18 @@ static void *superSubscribeProcess(void *sarg) { g_queryInfo.port); if (taos == NULL) { errorPrint("[%d] Failed to connect to TDengine, reason:%s\n", - winfo->threadID, taos_errstr(NULL)); + pThreadInfo->threadID, taos_errstr(NULL)); return NULL; } else { - winfo->taos = taos; + pThreadInfo->taos = taos; } } char sqlStr[MAX_TB_NAME_SIZE*2]; sprintf(sqlStr, "use %s", g_queryInfo.dbName); debugPrint("%s() %d sqlStr: %s\n", __func__, __LINE__, sqlStr); - if (0 != queryDbExec(winfo->taos, sqlStr, NO_INSERT_TYPE, false)) { - taos_close(winfo->taos); + if (0 != queryDbExec(pThreadInfo->taos, sqlStr, NO_INSERT_TYPE, false)) { + taos_close(pThreadInfo->taos); return NULL; } @@ -6236,7 +6281,7 @@ static void *superSubscribeProcess(void *sarg) { do { //if (g_queryInfo.specifiedQueryInfo.rate && (et - st) < g_queryInfo.specifiedQueryInfo.rate*1000) { // taosMsleep(g_queryInfo.specifiedQueryInfo.rate*1000 - (et - st)); // ms - // //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, winfo->start_table_from, winfo->end_table_to); + // //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, pThreadInfo->start_table_from, pThreadInfo->end_table_to); //} //st = taosGetTimestampMs(); @@ -6246,12 +6291,12 @@ static void *superSubscribeProcess(void *sarg) { char tmpFile[MAX_FILE_NAME_LEN*2] = {0}; if (g_queryInfo.superQueryInfo.result[i][0] != 0) { sprintf(tmpFile, "%s-%d", - g_queryInfo.specifiedQueryInfo.result[i], winfo->threadID); + g_queryInfo.specifiedQueryInfo.result[i], pThreadInfo->threadID); } - tsub[i] = subscribeImpl(winfo->taos, + tsub[i] = subscribeImpl(pThreadInfo->taos, g_queryInfo.specifiedQueryInfo.sql[i], topic, tmpFile); if (NULL == g_queryInfo.specifiedQueryInfo.tsub[i]) { - taos_close(winfo->taos); + taos_close(pThreadInfo->taos); return NULL; } } @@ -6272,7 +6317,7 @@ static void *superSubscribeProcess(void *sarg) { char tmpFile[MAX_FILE_NAME_LEN*2] = {0}; if (g_queryInfo.specifiedQueryInfo.result[i][0] != 0) { sprintf(tmpFile, "%s-%d", - g_queryInfo.specifiedQueryInfo.result[i], winfo->threadID); + g_queryInfo.specifiedQueryInfo.result[i], pThreadInfo->threadID); } getResult(res, tmpFile); } @@ -6285,7 +6330,7 @@ static void *superSubscribeProcess(void *sarg) { g_queryInfo.specifiedQueryInfo.subscribeKeepProgress); } - taos_close(winfo->taos); + taos_close(pThreadInfo->taos); return NULL; } @@ -6343,7 +6388,7 @@ static int subscribeTestProcess() { threadInfo *t_info = infos + i; t_info->threadID = i; t_info->taos = NULL; // TODO: workaround to use separate taos connection; - pthread_create(pids + i, NULL, superSubscribeProcess, t_info); + pthread_create(pids + i, NULL, specifiedSubscribe, t_info); } //==== create sub threads for query from sub table @@ -6386,7 +6431,7 @@ static int subscribeTestProcess() { t_info->end_table_to = i < b ? startFrom + a : startFrom + a - 1; startFrom = t_info->end_table_to + 1; t_info->taos = NULL; // TODO: workaround to use separate taos connection; - pthread_create(pidsOfSub + i, NULL, subSubscribeProcess, t_info); + pthread_create(pidsOfSub + i, NULL, superSubscribe, t_info); } g_queryInfo.superQueryInfo.threadCnt = threads; diff --git a/tests/examples/JDBC/taosdemo/src/main/resources/insert.json b/tests/examples/JDBC/taosdemo/src/main/resources/insert.json index 35c77731758c7c8de92f050803a537c776578f3d..7578083d33c73829ecce1678358e04d2a50d528f 100644 --- a/tests/examples/JDBC/taosdemo/src/main/resources/insert.json +++ b/tests/examples/JDBC/taosdemo/src/main/resources/insert.json @@ -36,8 +36,6 @@ "insert_mode": "taosc", "insert_rate": 0, "insert_rows": 100, - "multi_thread_write_one_tbl": "no", - "number_of_tbl_in_one_sql": 0, "interlace_rows": 3, "max_sql_len": 1024, "disorder_ratio": 0, diff --git a/tests/pytest/cluster/clusterEnvSetup/insert.json b/tests/pytest/cluster/clusterEnvSetup/insert.json index 4548ef74e6f2b56e3fdf78e46792065330ee359e..2f3cf0f0d9c98abdb31c19ad833098e23e0541f2 100644 --- a/tests/pytest/cluster/clusterEnvSetup/insert.json +++ b/tests/pytest/cluster/clusterEnvSetup/insert.json @@ -37,8 +37,6 @@ "insert_mode": "taosc", "insert_rate": 0, "insert_rows": 100000, - "multi_thread_write_one_tbl": "no", - "number_of_tbl_in_one_sql": 1, "interlace_rows": 100, "max_sql_len": 1024000, "disorder_ratio": 0, diff --git a/tests/pytest/perfbenchmark/bug3433.py b/tests/pytest/perfbenchmark/bug3433.py index 80e9cce0dc60937b3eaf7c122a8ae5951fd1ad6b..e4480df6b6753df88e3526930ac4bee087264d35 100644 --- a/tests/pytest/perfbenchmark/bug3433.py +++ b/tests/pytest/perfbenchmark/bug3433.py @@ -91,8 +91,6 @@ class TDTestCase: "data_source": "rand", "insert_mode": "taosc", "insert_rows": 1, - "multi_thread_write_one_tbl": "no", - "number_of_tbl_in_one_sql": 0, "rows_per_tbl": 1, "max_sql_len": 65480, "disorder_ratio": 0, @@ -135,8 +133,6 @@ class TDTestCase: "data_source": "rand", "insert_mode": "taosc", "insert_rows": 1, - "multi_thread_write_one_tbl": "no", - "number_of_tbl_in_one_sql": 0, "rows_per_tbl": 1, "max_sql_len": 65480, "disorder_ratio": 0, diff --git a/tests/pytest/query/query1970YearsAf.py b/tests/pytest/query/query1970YearsAf.py index e66a79f5e6d4938bef62eccc93f4cec3de6fe712..93404afd5967e772b383b8e2439ec4e89f5a7029 100644 --- a/tests/pytest/query/query1970YearsAf.py +++ b/tests/pytest/query/query1970YearsAf.py @@ -82,8 +82,6 @@ class TDTestCase: "data_source": "rand", "insert_mode": "taosc", "insert_rows": 5000, - "multi_thread_write_one_tbl": "no", - "number_of_tbl_in_one_sql": 0, "rows_per_tbl": 1, "max_sql_len": 65480, "disorder_ratio": 0, @@ -255,4 +253,4 @@ class TDTestCase: tdCases.addLinux(__file__, TDTestCase()) -tdCases.addWindows(__file__, TDTestCase()) \ No newline at end of file +tdCases.addWindows(__file__, TDTestCase()) diff --git a/tests/pytest/tools/insert-interlace.json b/tests/pytest/tools/insert-interlace.json index d4767ad0640c8b2a1528fc24e681c359b719a4b9..a5c545d1599ee742cf94a4bc592bf76abe792ae5 100644 --- a/tests/pytest/tools/insert-interlace.json +++ b/tests/pytest/tools/insert-interlace.json @@ -41,7 +41,6 @@ "data_source": "rand", "insert_mode": "taosc", "insert_rows": 250, - "multi_thread_write_one_tbl": "no", "interlace_rows": 80, "max_sql_len": 1024000, "disorder_ratio": 0, diff --git a/tests/pytest/tools/insert-tblimit-tboffset-createdb.json b/tests/pytest/tools/insert-tblimit-tboffset-createdb.json index 8c3b39fb0bf997806deafc554a5df9a3da74ffb6..9220bc1d17a0ead401c2adaf1e9d3a1455e2db00 100644 --- a/tests/pytest/tools/insert-tblimit-tboffset-createdb.json +++ b/tests/pytest/tools/insert-tblimit-tboffset-createdb.json @@ -40,8 +40,6 @@ "data_source": "rand", "insert_mode": "taosc", "insert_rows": 0, - "multi_thread_write_one_tbl": "no", - "number_of_tbl_in_one_sql": 0, "max_sql_len": 1024000, "disorder_ratio": 0, "disorder_range": 1000, diff --git a/tests/pytest/tools/insert-tblimit-tboffset-insertrec.json b/tests/pytest/tools/insert-tblimit-tboffset-insertrec.json index a9efa7c31c53f70cd65ec798ce64cd99673885f4..164d4fe8be99720e291ab3cf745765af92c1f23f 100644 --- a/tests/pytest/tools/insert-tblimit-tboffset-insertrec.json +++ b/tests/pytest/tools/insert-tblimit-tboffset-insertrec.json @@ -42,8 +42,6 @@ "insert_rows": 1000, "childtable_limit": 33, "childtable_offset": 33, - "multi_thread_write_one_tbl": "no", - "number_of_tbl_in_one_sql": 0, "max_sql_len": 1024000, "disorder_ratio": 0, "disorder_range": 1000, diff --git a/tests/pytest/tools/insert-tblimit-tboffset.json b/tests/pytest/tools/insert-tblimit-tboffset.json index f3d3e864baaa5429248f57a6dd1ea6b5bc314bfb..0b8e0bd6c550a163bcfe0500a43e88b84e2d27ae 100644 --- a/tests/pytest/tools/insert-tblimit-tboffset.json +++ b/tests/pytest/tools/insert-tblimit-tboffset.json @@ -42,8 +42,6 @@ "insert_rows": 1000, "childtable_limit": 33, "childtable_offset": 33, - "multi_thread_write_one_tbl": "no", - "number_of_tbl_in_one_sql": 0, "max_sql_len": 1024000, "disorder_ratio": 0, "disorder_range": 1000, diff --git a/tests/pytest/tools/insert-tblimit-tboffset0.json b/tests/pytest/tools/insert-tblimit-tboffset0.json index 6bf3783cb2b62442edb63b3cb07f22156ac33ddb..55d9e1905592e8e93d6d32a5fc159461c8b0fcb2 100644 --- a/tests/pytest/tools/insert-tblimit-tboffset0.json +++ b/tests/pytest/tools/insert-tblimit-tboffset0.json @@ -42,8 +42,6 @@ "insert_rows": 1000, "childtable_limit": 20, "childtable_offset": 0, - "multi_thread_write_one_tbl": "no", - "number_of_tbl_in_one_sql": 0, "max_sql_len": 1024000, "disorder_ratio": 0, "disorder_range": 1000, diff --git a/tests/pytest/tools/insert-tblimit1-tboffset.json b/tests/pytest/tools/insert-tblimit1-tboffset.json index 292902093deffb2e2085312d27f8a4c23e6a0c46..3a886656617be1e0d38cfef262fae9159eee5227 100644 --- a/tests/pytest/tools/insert-tblimit1-tboffset.json +++ b/tests/pytest/tools/insert-tblimit1-tboffset.json @@ -42,8 +42,6 @@ "insert_rows": 1000, "childtable_limit": 1, "childtable_offset": 50, - "multi_thread_write_one_tbl": "no", - "number_of_tbl_in_one_sql": 0, "max_sql_len": 1024000, "disorder_ratio": 0, "disorder_range": 1000, diff --git a/tests/pytest/tools/taosdemo-sampledata.json b/tests/pytest/tools/taosdemo-sampledata.json index d543b7e08672775433da8f789c4229d36b45653d..ce1c68b3939e3905fcfba8924e25f6b09bdd53f6 100644 --- a/tests/pytest/tools/taosdemo-sampledata.json +++ b/tests/pytest/tools/taosdemo-sampledata.json @@ -22,8 +22,6 @@ "insert_mode": "taosc", "insert_rate": 0, "insert_rows": 20, - "multi_thread_write_one_tbl": "no", - "number_of_tbl_in_one_sql": 0, "max_sql_len": 1048000, "timestamp_step": 1000, "start_timestamp": "2020-1-1 00:00:00", diff --git a/tests/pytest/tools/taosdemoPerformance.py b/tests/pytest/tools/taosdemoPerformance.py index 8ce99d096902215a4180c9431e166848028d359e..b50180e2b3452b8f6f4d0701fa595582cc5a7937 100644 --- a/tests/pytest/tools/taosdemoPerformance.py +++ b/tests/pytest/tools/taosdemoPerformance.py @@ -49,8 +49,6 @@ class taosdemoPerformace: "batch_create_tbl_num": 10, "insert_mode": "taosc", "insert_rows": 100000, - "multi_thread_write_one_tbl": "no", - "number_of_tbl_in_one_sql": 0, "interlace_rows": 100, "max_sql_len": 1024000, "disorder_ratio": 0,