From 41d412fba061ea9be3683b80c7202f00926b9532 Mon Sep 17 00:00:00 2001 From: plum-lihui Date: Wed, 1 Jun 2022 14:21:08 +0800 Subject: [PATCH] test: modify consume processor --- tests/test/c/tmqSim.c | 99 +++++++++++++++++++++++++++++++++++++++---- 1 file changed, 91 insertions(+), 8 deletions(-) diff --git a/tests/test/c/tmqSim.c b/tests/test/c/tmqSim.c index accd1dd080..13c64b68f9 100644 --- a/tests/test/c/tmqSim.c +++ b/tests/test/c/tmqSim.c @@ -61,6 +61,10 @@ typedef struct { tmq_t* tmq; tmq_list_t* topicList; + + int32_t numOfVgroups; + int32_t rowsOfPerVgroups[32][2]; // [i][0]: vgroup id, [i][1]: rows of consume + int64_t ts; } SThreadInfo; @@ -69,7 +73,8 @@ typedef struct { char cdbName[32]; char dbName[32]; int32_t showMsgFlag; - int32_t showRowFlag; + int32_t showRowFlag; + int32_t saveRowFlag; int32_t consumeDelay; // unit s int32_t numOfThread; SThreadInfo stThreads[MAX_CONSUMER_THREAD_CNT]; @@ -77,6 +82,7 @@ typedef struct { static SConfInfo g_stConfInfo; TdFilePtr g_fp = NULL; +static int running = 1; // char* g_pRowValue = NULL; // TdFilePtr g_fp = NULL; @@ -93,6 +99,8 @@ static void printHelp() { printf("%s%s%s%d\n", indent, indent, "showMsgFlag, default is ", g_stConfInfo.showMsgFlag); printf("%s%s\n", indent, "-r"); printf("%s%s%s%d\n", indent, indent, "showRowFlag, default is ", g_stConfInfo.showRowFlag); + printf("%s%s\n", indent, "-s"); + printf("%s%s%s%d\n", indent, indent, "saveRowFlag, default is ", g_stConfInfo.saveRowFlag); printf("%s%s\n", indent, "-y"); printf("%s%s%s%d\n", indent, indent, "consume delay, default is s", g_stConfInfo.consumeDelay); exit(EXIT_SUCCESS); @@ -135,6 +143,7 @@ void saveConfigToLogFile() { taosFprintfFile(g_fp, "# cdbName: %s\n", g_stConfInfo.cdbName); taosFprintfFile(g_fp, "# showMsgFlag: %d\n", g_stConfInfo.showMsgFlag); taosFprintfFile(g_fp, "# showRowFlag: %d\n", g_stConfInfo.showRowFlag); + taosFprintfFile(g_fp, "# saveRowFlag: %d\n", g_stConfInfo.saveRowFlag); taosFprintfFile(g_fp, "# consumeDelay: %d\n", g_stConfInfo.consumeDelay); taosFprintfFile(g_fp, "# numOfThread: %d\n", g_stConfInfo.numOfThread); @@ -165,6 +174,7 @@ void parseArgument(int32_t argc, char* argv[]) { memset(&g_stConfInfo, 0, sizeof(SConfInfo)); g_stConfInfo.showMsgFlag = 0; g_stConfInfo.showRowFlag = 0; + g_stConfInfo.saveRowFlag = 0; g_stConfInfo.consumeDelay = 5; for (int32_t i = 1; i < argc; i++) { @@ -181,6 +191,8 @@ void parseArgument(int32_t argc, char* argv[]) { g_stConfInfo.showMsgFlag = atol(argv[++i]); } else if (strcmp(argv[i], "-r") == 0) { g_stConfInfo.showRowFlag = atol(argv[++i]); + } else if (strcmp(argv[i], "-s") == 0) { + g_stConfInfo.saveRowFlag = atol(argv[++i]); } else if (strcmp(argv[i], "-y") == 0) { g_stConfInfo.consumeDelay = atol(argv[++i]); } else { @@ -200,6 +212,7 @@ void parseArgument(int32_t argc, char* argv[]) { pPrint("%s consumeDelay:%d %s", GREEN, g_stConfInfo.consumeDelay, NC); pPrint("%s showMsgFlag:%d %s", GREEN, g_stConfInfo.showMsgFlag, NC); pPrint("%s showRowFlag:%d %s", GREEN, g_stConfInfo.showRowFlag, NC); + pPrint("%s saveRowFlag:%d %s", GREEN, g_stConfInfo.saveRowFlag, NC); #endif } @@ -225,15 +238,52 @@ void ltrim(char* str) { // return str; } -static int running = 1; -static int32_t msg_process(TAOS_RES* msg, int64_t msgIndex, int32_t threadLable) { +void addRowsToVgroupId(SThreadInfo* pInfo, int32_t vgroupId, int32_t rows) { + int32_t i; + for (i = 0; i < pInfo->numOfVgroups; i++) { + if (vgroupId == pInfo->rowsOfPerVgroups[i][0]) { + pInfo->rowsOfPerVgroups[i][1] += rows; + } + } + + if (i == pInfo->numOfVgroups) { + pInfo->rowsOfPerVgroups[i][0] = vgroupId; + pInfo->rowsOfPerVgroups[i][1] += rows; + pInfo->numOfVgroups++; + } +} + + +int32_t saveConsumeContentToTbl(SThreadInfo* pInfo, char* buf) { + char sqlStr[1024] = {0}; + + TAOS* pConn = taos_connect(NULL, "root", "taosdata", NULL, 0); + assert(pConn != NULL); + + sprintf(sqlStr, "insert into %s.content_%d values (%"PRId64", \'%s\')", g_stConfInfo.cdbName, pInfo->consumerId, pInfo->ts++, buf); + TAOS_RES* pRes = taos_query(pConn, sqlStr); + if (taos_errno(pRes) != 0) { + pError("error in insert consume result, reason:%s\n", taos_errstr(pRes)); + taosFprintfFile(g_fp, "error in insert consume result, reason:%s\n", taos_errstr(pRes)); + taosCloseFile(&g_fp); + taos_free_result(pRes); + exit(-1); + } + + taos_free_result(pRes); + + return 0; +} + +static int32_t msg_process(TAOS_RES* msg, SThreadInfo* pInfo, int32_t threadLable) { char buf[1024]; int32_t totalRows = 0; - + // printf("topic: %s\n", tmq_get_topic_name(msg)); - // printf("vg:%d\n", tmq_get_vgroup_id(msg)); - taosFprintfFile(g_fp, "msg index:%" PRId64 ", threadLable: %d\n", msgIndex, threadLable); - taosFprintfFile(g_fp, "topic: %s, vgroupId: %d\n", tmq_get_topic_name(msg), tmq_get_vgroup_id(msg)); + int32_t vgroupId = tmq_get_vgroup_id(msg); + + taosFprintfFile(g_fp, "msg index:%" PRId64 ", threadLable: %d\n", threadLable, pInfo->consumerId); + taosFprintfFile(g_fp, "topic: %s, vgroupId: %d, tableName: %s\n", tmq_get_topic_name(msg), vgroupId, tmq_get_table_name(msg)); while (1) { TAOS_ROW row = taos_fetch_row(msg); @@ -247,11 +297,16 @@ static int32_t msg_process(TAOS_RES* msg, int64_t msgIndex, int32_t threadLable) if (0 != g_stConfInfo.showRowFlag) { taosFprintfFile(g_fp, "rows[%d]: %s\n", totalRows, buf); + if (0 != g_stConfInfo.saveRowFlag) { + saveConsumeContentToTbl(pInfo, buf); + } } totalRows++; } + addRowsToVgroupId(pInfo, vgroupId, totalRows); + return totalRows; } @@ -344,6 +399,32 @@ int32_t saveConsumeResult(SThreadInfo* pInfo) { taos_free_result(pRes); + #if 0 + // vgroups + for (i = 0; i < pInfo->numOfVgroups; i++) { + // schema: ts timestamp, consumerid int, consummsgcnt bigint, checkresult int + sprintf(sqlStr, "insert into %s.vgroup_%d values (%"PRId64", %d, %" PRId64 ", %" PRId64 ", %d)", + g_stConfInfo.cdbName, + now, + pInfo->consumerId, + pInfo->consumeMsgCnt, + pInfo->consumeRowCnt, + pInfo->checkresult); + + char tmpString[128]; + taosFprintfFile(g_fp, "%s, consume id %d result: %s\n", getCurrentTimeString(tmpString), pInfo->consumerId ,sqlStr); + + TAOS_RES* pRes = taos_query(pConn, sqlStr); + if (taos_errno(pRes) != 0) { + pError("error in save consumeinfo, reason:%s\n", taos_errstr(pRes)); + taos_free_result(pRes); + exit(-1); + } + + taos_free_result(pRes); + } + #endif + return 0; } @@ -356,11 +437,13 @@ void loop_consume(SThreadInfo* pInfo) { char tmpString[128]; taosFprintfFile(g_fp, "%s consumer id %d start to loop pull msg\n", getCurrentTimeString(tmpString), pInfo->consumerId); + pInfo->ts = taosGetTimestampMs(); + while (running) { TAOS_RES* tmqMsg = tmq_consumer_poll(pInfo->tmq, g_stConfInfo.consumeDelay * 1000); if (tmqMsg) { if (0 != g_stConfInfo.showMsgFlag) { - totalRows += msg_process(tmqMsg, totalMsgs, pInfo->consumerId); + totalRows += msg_process(tmqMsg, pInfo, totalMsgs); } taos_free_result(tmqMsg); -- GitLab