diff --git a/src/kit/taosdemo/taosdemo.c b/src/kit/taosdemo/taosdemo.c index a8af72d36af21cf1aa820d45c93296f6d104fc78..e7b2d4f16b830c05be9ebef8f14768f7d0974444 100644 --- a/src/kit/taosdemo/taosdemo.c +++ b/src/kit/taosdemo/taosdemo.c @@ -380,7 +380,7 @@ typedef struct SDbs_S { typedef struct SpecifiedQueryInfo_S { uint64_t queryInterval; // 0: unlimit > 0 loop/s uint32_t concurrent; - uint64_t sqlCount; + int sqlCount; uint32_t asyncMode; // 0: sync, 1: async uint64_t subscribeInterval; // ms uint64_t queryTimes; @@ -389,6 +389,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]; int resubAfterConsume[MAX_QUERY_SQL_COUNT]; + int endAfterConsume[MAX_QUERY_SQL_COUNT]; TAOS_SUB* tsub[MAX_QUERY_SQL_COUNT]; char topic[MAX_QUERY_SQL_COUNT][32]; int consumed[MAX_QUERY_SQL_COUNT]; @@ -407,10 +408,11 @@ typedef struct SuperQueryInfo_S { uint64_t queryTimes; int64_t childTblCount; char childTblPrefix[MAX_TB_NAME_SIZE]; - uint64_t sqlCount; + int sqlCount; char sql[MAX_QUERY_SQL_COUNT][MAX_QUERY_SQL_LENGTH+1]; char result[MAX_QUERY_SQL_COUNT][MAX_FILE_NAME_LEN+1]; int resubAfterConsume; + int endAfterConsume; TAOS_SUB* tsub[MAX_QUERY_SQL_COUNT]; char* childTblName; @@ -1775,7 +1777,7 @@ static void printfQueryMeta() { if ((SUBSCRIBE_TEST == g_args.test_mode) || (QUERY_TEST == g_args.test_mode)) { printf("specified table query info: \n"); - printf("sqlCount: \033[33m%"PRIu64"\033[0m\n", + printf("sqlCount: \033[33m%d\033[0m\n", g_queryInfo.specifiedQueryInfo.sqlCount); if (g_queryInfo.specifiedQueryInfo.sqlCount > 0) { printf("specified tbl query times:\n"); @@ -1795,15 +1797,15 @@ static void printfQueryMeta() { printf("keepProgress: \033[33m%d\033[0m\n", g_queryInfo.specifiedQueryInfo.subscribeKeepProgress); - for (uint64_t i = 0; i < g_queryInfo.specifiedQueryInfo.sqlCount; i++) { - printf(" sql[%"PRIu64"]: \033[33m%s\033[0m\n", + for (int i = 0; i < g_queryInfo.specifiedQueryInfo.sqlCount; i++) { + printf(" sql[%d]: \033[33m%s\033[0m\n", i, g_queryInfo.specifiedQueryInfo.sql[i]); } printf("\n"); } printf("super table query info:\n"); - printf("sqlCount: \033[33m%"PRIu64"\033[0m\n", + printf("sqlCount: \033[33m%d\033[0m\n", g_queryInfo.superQueryInfo.sqlCount); if (g_queryInfo.superQueryInfo.sqlCount > 0) { @@ -4277,7 +4279,7 @@ static bool getMetaFromQueryJsonFile(cJSON* root) { if (concurrent && concurrent->type == cJSON_Number) { if (concurrent->valueint <= 0) { errorPrint( - "%s() LN%d, query sqlCount %"PRIu64" or concurrent %d is not correct.\n", + "%s() LN%d, query sqlCount %d or concurrent %d is not correct.\n", __func__, __LINE__, g_queryInfo.specifiedQueryInfo.sqlCount, g_queryInfo.specifiedQueryInfo.concurrent); @@ -4376,6 +4378,17 @@ static bool getMetaFromQueryJsonFile(cJSON* root) { tstrncpy(g_queryInfo.specifiedQueryInfo.sql[j], sqlStr->valuestring, MAX_QUERY_SQL_LENGTH); + cJSON* endAfterConsume = + cJSON_GetObjectItem(specifiedQuery, "endAfterConsume"); + if (endAfterConsume + && endAfterConsume->type == cJSON_Number) { + g_queryInfo.specifiedQueryInfo.endAfterConsume[j] + = endAfterConsume->valueint; + } else if (!endAfterConsume) { + // default value is -1, which mean infinite loop + g_queryInfo.specifiedQueryInfo.endAfterConsume[j] = -1; + } + cJSON* resubAfterConsume = cJSON_GetObjectItem(specifiedQuery, "resubAfterConsume"); if (resubAfterConsume @@ -4383,9 +4396,8 @@ static bool getMetaFromQueryJsonFile(cJSON* root) { g_queryInfo.specifiedQueryInfo.resubAfterConsume[j] = resubAfterConsume->valueint; } else if (!resubAfterConsume) { - //printf("failed to read json, subscribe interval no found\n"); - //goto PARSE_OVER; - g_queryInfo.specifiedQueryInfo.resubAfterConsume[j] = 1; + // default value is -1, which mean do not resub + g_queryInfo.specifiedQueryInfo.resubAfterConsume[j] = -1; } cJSON *result = cJSON_GetObjectItem(sql, "result"); @@ -4529,16 +4541,26 @@ static bool getMetaFromQueryJsonFile(cJSON* root) { g_queryInfo.superQueryInfo.subscribeKeepProgress = 0; } + cJSON* superEndAfterConsume = + cJSON_GetObjectItem(superQuery, "endAfterConsume"); + if (superEndAfterConsume + && superEndAfterConsume->type == cJSON_Number) { + g_queryInfo.superQueryInfo.endAfterConsume = + superEndAfterConsume->valueint; + } else if (!superEndAfterConsume) { + // default value is -1, which mean do not resub + g_queryInfo.superQueryInfo.endAfterConsume = -1; + } + cJSON* superResubAfterConsume = - cJSON_GetObjectItem(superQuery, "resubAfterConsume"); + cJSON_GetObjectItem(superQuery, "endAfterConsume"); if (superResubAfterConsume && superResubAfterConsume->type == cJSON_Number) { - g_queryInfo.superQueryInfo.resubAfterConsume = + g_queryInfo.superQueryInfo.endAfterConsume = superResubAfterConsume->valueint; } else if (!superResubAfterConsume) { - //printf("failed to read json, subscribe interval no found\n"); - ////goto PARSE_OVER; - g_queryInfo.superQueryInfo.resubAfterConsume = 1; + // default value is -1, which mean do not resub + g_queryInfo.superQueryInfo.endAfterConsume = -1; } // supert table sqls @@ -7262,7 +7284,10 @@ static void *superSubscribe(void *sarg) { uint64_t st = 0, et = 0; - while(1) { + while ((g_queryInfo.superQueryInfo.endAfterConsume == -1) + || (g_queryInfo.superQueryInfo.endAfterConsume < + consumed[pThreadInfo->end_table_to - pThreadInfo->start_table_from])) { + for (uint64_t i = pThreadInfo->start_table_from; i <= pThreadInfo->end_table_to; i++) { tsubSeq = i - pThreadInfo->start_table_from; @@ -7291,7 +7316,7 @@ static void *superSubscribe(void *sarg) { } consumed[tsubSeq] ++; - if ((g_queryInfo.superQueryInfo.subscribeKeepProgress) + if ((g_queryInfo.superQueryInfo.resubAfterConsume != -1) && (consumed[tsubSeq] >= g_queryInfo.superQueryInfo.resubAfterConsume)) { printf("keepProgress:%d, resub super table query: %"PRIu64"\n", @@ -7373,7 +7398,10 @@ static void *specifiedSubscribe(void *sarg) { // start loop to consume result g_queryInfo.specifiedQueryInfo.consumed[pThreadInfo->threadID] = 0; - while(1) { + while((g_queryInfo.specifiedQueryInfo.endAfterConsume[pThreadInfo->querySeq] == -1) + || (g_queryInfo.specifiedQueryInfo.consumed[pThreadInfo->threadID] < + g_queryInfo.specifiedQueryInfo.endAfterConsume[pThreadInfo->querySeq])) { + if (ASYNC_MODE == g_queryInfo.specifiedQueryInfo.asyncMode) { continue; } @@ -7392,7 +7420,7 @@ static void *specifiedSubscribe(void *sarg) { } g_queryInfo.specifiedQueryInfo.consumed[pThreadInfo->threadID] ++; - if ((g_queryInfo.specifiedQueryInfo.subscribeKeepProgress) + if ((g_queryInfo.specifiedQueryInfo.resubAfterConsume[pThreadInfo->querySeq] != -1) && (g_queryInfo.specifiedQueryInfo.consumed[pThreadInfo->threadID] >= g_queryInfo.specifiedQueryInfo.resubAfterConsume[pThreadInfo->querySeq])) { printf("keepProgress:%d, resub specified query: %"PRIu64"\n", @@ -7460,12 +7488,12 @@ static int subscribeTestProcess() { //==== create threads for query for specified table if (g_queryInfo.specifiedQueryInfo.sqlCount <= 0) { - debugPrint("%s() LN%d, sepcified query sqlCount %"PRIu64".\n", + debugPrint("%s() LN%d, sepcified query sqlCount %d.\n", __func__, __LINE__, g_queryInfo.specifiedQueryInfo.sqlCount); } else { if (g_queryInfo.specifiedQueryInfo.concurrent <= 0) { - errorPrint("%s() LN%d, sepcified query sqlCount %"PRIu64".\n", + errorPrint("%s() LN%d, sepcified query sqlCount %d.\n", __func__, __LINE__, g_queryInfo.specifiedQueryInfo.sqlCount); exit(-1); @@ -7498,7 +7526,7 @@ static int subscribeTestProcess() { //==== create threads for super table query if (g_queryInfo.superQueryInfo.sqlCount <= 0) { - debugPrint("%s() LN%d, super table query sqlCount %"PRIu64".\n", + debugPrint("%s() LN%d, super table query sqlCount %d.\n", __func__, __LINE__, g_queryInfo.superQueryInfo.sqlCount); } else {