From 795503ccbe4f9ebc5e970e00414283822d209328 Mon Sep 17 00:00:00 2001 From: Shuduo Sang Date: Sat, 22 May 2021 16:17:03 +0800 Subject: [PATCH] [TD-4296]: taosdemo sub keepProgress. (#6201) Co-authored-by: Shuduo Sang --- src/kit/taosdemo/taosdemo.c | 35 +++++++++++++++++++++-------------- 1 file changed, 21 insertions(+), 14 deletions(-) diff --git a/src/kit/taosdemo/taosdemo.c b/src/kit/taosdemo/taosdemo.c index feab1aab90..aa7a022f91 100644 --- a/src/kit/taosdemo/taosdemo.c +++ b/src/kit/taosdemo/taosdemo.c @@ -403,7 +403,7 @@ typedef struct SuperQueryInfo_S { uint64_t 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[MAX_QUERY_SQL_COUNT]; + int resubAfterConsume; TAOS_SUB* tsub[MAX_QUERY_SQL_COUNT]; char* childTblName; @@ -4495,6 +4495,18 @@ static bool getMetaFromQueryJsonFile(cJSON* root) { g_queryInfo.superQueryInfo.subscribeKeepProgress = 0; } + cJSON* superResubAfterConsume = + cJSON_GetObjectItem(superQuery, "resubAfterConsume"); + if (superResubAfterConsume + && superResubAfterConsume->type == cJSON_Number) { + g_queryInfo.superQueryInfo.resubAfterConsume = + superResubAfterConsume->valueint; + } else if (!superResubAfterConsume) { + //printf("failed to read json, subscribe interval no found\n"); + ////goto PARSE_OVER; + g_queryInfo.superQueryInfo.resubAfterConsume = 1; + } + // sqls cJSON* subsqls = cJSON_GetObjectItem(superQuery, "sqls"); if (!subsqls) { @@ -4526,18 +4538,6 @@ static bool getMetaFromQueryJsonFile(cJSON* root) { tstrncpy(g_queryInfo.superQueryInfo.sql[j], sqlStr->valuestring, MAX_QUERY_SQL_LENGTH); - cJSON* superResubAfterConsume = - cJSON_GetObjectItem(sql, "resubAfterConsume"); - if (superResubAfterConsume - && superResubAfterConsume->type == cJSON_Number) { - g_queryInfo.superQueryInfo.resubAfterConsume[j] = - superResubAfterConsume->valueint; - } else if (!superResubAfterConsume) { - //printf("failed to read json, subscribe interval no found\n"); - //goto PARSE_OVER; - g_queryInfo.superQueryInfo.resubAfterConsume[j] = 1; - } - cJSON *result = cJSON_GetObjectItem(sql, "result"); if (result != NULL && result->type == cJSON_String && result->valuestring != NULL){ @@ -6855,6 +6855,8 @@ static void *superSubscribe(void *sarg) { } TAOS_RES* res = NULL; + uint64_t st = 0, et = 0; + while(1) { for (uint64_t i = pThreadInfo->start_table_from; i <= pThreadInfo->end_table_to; i++) { @@ -6863,7 +6865,12 @@ static void *superSubscribe(void *sarg) { continue; } + st = taosGetTimestampMs(); + performancePrint("st: %"PRIu64" et: %"PRIu64" st-et: %"PRIu64"\n", st, et, (st - et)); res = taos_consume(tsub[tsubSeq]); + et = taosGetTimestampMs(); + performancePrint("st: %"PRIu64" et: %"PRIu64" delta: %"PRIu64"\n", st, et, (et - st)); + if (res) { if (g_queryInfo.superQueryInfo.result[pThreadInfo->querySeq][0] != 0) { sprintf(pThreadInfo->fp, "%s-%d", @@ -6881,7 +6888,7 @@ static void *superSubscribe(void *sarg) { if ((g_queryInfo.superQueryInfo.subscribeKeepProgress) && (consumed[tsubSeq] >= - g_queryInfo.superQueryInfo.resubAfterConsume[pThreadInfo->querySeq])) { + g_queryInfo.superQueryInfo.resubAfterConsume)) { printf("keepProgress:%d, resub super table query: %"PRIu64"\n", g_queryInfo.superQueryInfo.subscribeKeepProgress, pThreadInfo->querySeq); -- GitLab