xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #14412 from taosdata/test3.0/lihui 

Test3.0/lihui
This commit is contained in:
Hui Li 2022-07-01 09:48:07 +08:00 committed by GitHub
parent 126d884ab3 4b6ab7fe26
commit f421ab09dd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 18 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
tests/test/c/tmqSim.c
View File

@ -633,8 +633,9 @@ void loop_consume(SThreadInfo* pInfo) {
} }
} }
uint64_t lastPrintTime = taosGetTimestampMs(); int64_t lastTotalMsgs = 0;
uint64_t startTs = taosGetTimestampMs(); uint64_t lastPrintTime = taosGetTimestampMs();
uint64_t startTs = taosGetTimestampMs();
int32_t consumeDelay = g_stConfInfo.consumeDelay == -1 ? -1 : (g_stConfInfo.consumeDelay * 1000); int32_t consumeDelay = g_stConfInfo.consumeDelay == -1 ? -1 : (g_stConfInfo.consumeDelay * 1000);
while (running) { while (running) {
@ -647,20 +648,22 @@ void loop_consume(SThreadInfo* pInfo) {
taos_free_result(tmqMsg); taos_free_result(tmqMsg);
totalMsgs++; totalMsgs++;
int64_t currentPrintTime = taosGetTimestampMs(); int64_t currentPrintTime = taosGetTimestampMs();
if (currentPrintTime - lastPrintTime > 10 * 1000) { if (currentPrintTime - lastPrintTime > 10 * 1000) {
taosFprintfFile(g_fp, "consumer id %d has currently poll total msgs: %" PRId64 "\n", pInfo->consumerId, taosFprintfFile(g_fp,
totalMsgs); "consumer id %d has currently poll total msgs: %" PRId64 ", period rate: %.3f msgs/second\n",
lastPrintTime = currentPrintTime; pInfo->consumerId, totalMsgs, (totalMsgs - lastTotalMsgs) * 1000.0/(currentPrintTime - lastPrintTime));
} lastPrintTime = currentPrintTime;
lastTotalMsgs = totalMsgs;
}
if (0 == once_flag) { if (0 == once_flag) {
once_flag = 1; once_flag = 1;
notifyMainScript(pInfo, NOTIFY_CMD_START_CONSUM); notifyMainScript(pInfo, NOTIFY_CMD_START_CONSUM);
} }
if (totalRows >= pInfo->expectMsgCnt) { if ((totalRows >= pInfo->expectMsgCnt) || (totalMsgs >= pInfo->expectMsgCnt)) {
char tmpString[128]; char tmpString[128];
taosFprintfFile(g_fp, "%s over than expect rows, so break consume\n", getCurrentTimeString(tmpString)); taosFprintfFile(g_fp, "%s over than expect rows, so break consume\n", getCurrentTimeString(tmpString));
break; break;
@ -671,7 +674,7 @@ void loop_consume(SThreadInfo* pInfo) {
break; break;
} }
} }
if (0 == running) { if (0 == running) {
taosFprintfFile(g_fp, "receive stop signal and not continue consume\n"); taosFprintfFile(g_fp, "receive stop signal and not continue consume\n");
} }
@ -881,11 +884,11 @@ int main(int32_t argc, char* argv[]) {
int64_t t = end - start; int64_t t = end - start;
if (0 == t) t = 1; if (0 == t) t = 1;
double tInMs = (double)t / 1000000.0; double tInMs = (double)t / 1000000.0;
taosFprintfFile(g_fp, taosFprintfFile(g_fp,
"Spent %.4f seconds to poll msgs: %" PRIu64 " with %d thread(s), throughput: %.2f msgs/second\n\n", "Spent %.3f seconds to poll msgs: %" PRIu64 " with %d thread(s), throughput: %.3f msgs/second\n\n",
tInMs, totalMsgs, g_stConfInfo.numOfThread, (double)(totalMsgs / tInMs)); tInMs, totalMsgs, g_stConfInfo.numOfThread, (double)(totalMsgs / tInMs));
taosFprintfFile(g_fp, "==== close tmqlog ====\n"); taosFprintfFile(g_fp, "==== close tmqlog ====\n");
taosCloseFile(&g_fp); taosCloseFile(&g_fp);