test: add test case
This commit is contained in:
parent
f324a50b2e
commit
2caa9fdf7d
|
@ -77,6 +77,22 @@ class TMQCom:
|
||||||
|
|
||||||
return resultList
|
return resultList
|
||||||
|
|
||||||
|
def selectConsumeMsgResult(self,expectRows,cdbName='cdb'):
|
||||||
|
resultList=[]
|
||||||
|
while 1:
|
||||||
|
tdSql.query("select * from %s.consumeresult"%cdbName)
|
||||||
|
#tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
|
||||||
|
if tdSql.getRows() == expectRows:
|
||||||
|
break
|
||||||
|
else:
|
||||||
|
time.sleep(5)
|
||||||
|
|
||||||
|
for i in range(expectRows):
|
||||||
|
tdLog.info ("consume id: %d, consume msgs: %d, consume rows: %d"%(tdSql.getData(i , 1), tdSql.getData(i , 2), tdSql.getData(i , 3)))
|
||||||
|
resultList.append(tdSql.getData(i , 2))
|
||||||
|
|
||||||
|
return resultList
|
||||||
|
|
||||||
def startTmqSimProcess(self,pollDelay,dbName,showMsg=1,showRow=1,cdbName='cdb',valgrind=0,alias=0):
|
def startTmqSimProcess(self,pollDelay,dbName,showMsg=1,showRow=1,cdbName='cdb',valgrind=0,alias=0):
|
||||||
buildPath = tdCom.getBuildPath()
|
buildPath = tdCom.getBuildPath()
|
||||||
cfgPath = tdCom.getClientCfgPath()
|
cfgPath = tdCom.getClientCfgPath()
|
||||||
|
|
|
@ -635,6 +635,9 @@ void loop_consume(SThreadInfo* pInfo) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
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) {
|
||||||
TAOS_RES* tmqMsg = tmq_consumer_poll(pInfo->tmq, consumeDelay);
|
TAOS_RES* tmqMsg = tmq_consumer_poll(pInfo->tmq, consumeDelay);
|
||||||
|
@ -647,6 +650,14 @@ void loop_consume(SThreadInfo* pInfo) {
|
||||||
|
|
||||||
totalMsgs++;
|
totalMsgs++;
|
||||||
|
|
||||||
|
int64_t currentPrintTime = taosGetTimestampMs();
|
||||||
|
if (currentPrintTime - lastPrintTime > 10 * 1000) {
|
||||||
|
taosFprintfFile(g_fp,
|
||||||
|
"consumer id %d has currently poll total msgs: %" PRId64 "\n",
|
||||||
|
pInfo->consumerId, totalMsgs);
|
||||||
|
lastPrintTime = currentPrintTime;
|
||||||
|
}
|
||||||
|
|
||||||
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);
|
||||||
|
@ -676,8 +687,6 @@ void loop_consume(SThreadInfo* pInfo) {
|
||||||
}
|
}
|
||||||
|
|
||||||
void* consumeThreadFunc(void* param) {
|
void* consumeThreadFunc(void* param) {
|
||||||
int32_t totalMsgs = 0;
|
|
||||||
|
|
||||||
SThreadInfo* pInfo = (SThreadInfo*)param;
|
SThreadInfo* pInfo = (SThreadInfo*)param;
|
||||||
|
|
||||||
pInfo->taos = taos_connect(NULL, "root", "taosdata", NULL, 0);
|
pInfo->taos = taos_connect(NULL, "root", "taosdata", NULL, 0);
|
||||||
|
@ -859,12 +868,27 @@ int main(int32_t argc, char* argv[]) {
|
||||||
(void*)(&(g_stConfInfo.stThreads[i])));
|
(void*)(&(g_stConfInfo.stThreads[i])));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
int64_t start = taosGetTimestampUs();
|
||||||
|
|
||||||
for (int32_t i = 0; i < g_stConfInfo.numOfThread; i++) {
|
for (int32_t i = 0; i < g_stConfInfo.numOfThread; i++) {
|
||||||
taosThreadJoin(g_stConfInfo.stThreads[i].thread, NULL);
|
taosThreadJoin(g_stConfInfo.stThreads[i].thread, NULL);
|
||||||
taosThreadClear(&g_stConfInfo.stThreads[i].thread);
|
taosThreadClear(&g_stConfInfo.stThreads[i].thread);
|
||||||
}
|
}
|
||||||
|
|
||||||
// printf("consumer: %d, cosumer1: %d\n", totalMsgs, pInfo->consumeMsgCnt);
|
int64_t end = taosGetTimestampUs();
|
||||||
|
|
||||||
|
int64_t totalMsgs = 0;
|
||||||
|
for (int32_t i = 0; i < g_stConfInfo.numOfThread; i++) {
|
||||||
|
totalMsgs += g_stConfInfo.stThreads[i].consumeMsgCnt;
|
||||||
|
}
|
||||||
|
|
||||||
|
int64_t t = end - start;
|
||||||
|
if (0 == t) t = 1;
|
||||||
|
|
||||||
|
double tInMs = (double)t / 1000000.0;
|
||||||
|
taosFprintfFile(g_fp,
|
||||||
|
"Spent %.4f seconds to poll msgs: %" PRIu64 " with %d thread(s), throughput: %.2f msgs/second\n\n",
|
||||||
|
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);
|
||||||
|
|
Loading…
Reference in New Issue