From 7adb395b850ad6572c93c6c46300d118ffdb28c1 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Thu, 14 Apr 2022 14:42:51 +0800 Subject: [PATCH] enh: add connection type for tmq --- example/src/tmq.c | 7 +- include/client/taos.h | 4 +- include/common/tmsg.h | 2 + source/client/inc/clientInt.h | 9 +- source/client/src/clientHb.c | 17 +- source/client/src/clientImpl.c | 17 +- source/client/src/clientMain.c | 8 +- source/client/src/clientMsgHandler.c | 94 ++--- source/client/src/tmq.c | 10 +- source/common/src/tmsg.c | 4 + source/dnode/mnode/impl/src/mndProfile.c | 28 +- tests/test/c/tmqDemo.c | 460 ++++++++++++----------- tests/test/c/tmqSim.c | 193 +++++----- 13 files changed, 450 insertions(+), 403 deletions(-) diff --git a/example/src/tmq.c b/example/src/tmq.c index 832e389a13..67ab719bab 100644 --- a/example/src/tmq.c +++ b/example/src/tmq.c @@ -144,6 +144,7 @@ void tmq_commit_cb_print(tmq_t* tmq, tmq_resp_err_t resp, tmq_topic_vgroup_list_ } tmq_t* build_consumer() { +#if 0 TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); assert(pConn != NULL); @@ -152,11 +153,15 @@ tmq_t* build_consumer() { printf("error in use db, reason:%s\n", taos_errstr(pRes)); } taos_free_result(pRes); +#endif tmq_conf_t* conf = tmq_conf_new(); tmq_conf_set(conf, "group.id", "tg2"); + tmq_conf_set(conf, "td.connect.user", "root"); + tmq_conf_set(conf, "td.connect.pass", "taosdata"); + tmq_conf_set(conf, "td.connect.db", "abc1"); tmq_conf_set_offset_commit_cb(conf, tmq_commit_cb_print); - tmq_t* tmq = tmq_consumer_new(pConn, conf, NULL, 0); + tmq_t* tmq = tmq_consumer_new1(conf, NULL, 0); return tmq; } diff --git a/include/client/taos.h b/include/client/taos.h index 87948e7824..2180903633 100644 --- a/include/client/taos.h +++ b/include/client/taos.h @@ -247,10 +247,10 @@ DLL_EXPORT tmq_list_t *tmq_list_new(); DLL_EXPORT int32_t tmq_list_append(tmq_list_t *, const char *); DLL_EXPORT void tmq_list_destroy(tmq_list_t *); -// will be removed in 3.0 +#if 1 DLL_EXPORT tmq_t *tmq_consumer_new(void *conn, tmq_conf_t *conf, char *errstr, int32_t errstrLen); +#endif -// will replace last one DLL_EXPORT tmq_t *tmq_consumer_new1(tmq_conf_t *conf, char *errstr, int32_t errstrLen); DLL_EXPORT const char *tmq_err2str(tmq_resp_err_t); diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 1ded61eb3f..d7e86bab4f 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -332,6 +332,7 @@ int32_t taosEncodeSEpSet(void** buf, const SEpSet* pEp); void* taosDecodeSEpSet(void* buf, SEpSet* pEp); typedef struct { + int8_t connType; int32_t pid; char app[TSDB_APP_NAME_LEN]; char db[TSDB_DB_NAME_LEN]; @@ -346,6 +347,7 @@ typedef struct { int64_t clusterId; int32_t connId; int8_t superUser; + int8_t connType; SEpSet epSet; char sVersion[128]; } SConnectRsp; diff --git a/source/client/inc/clientInt.h b/source/client/inc/clientInt.h index b7d88c277b..142c4ee5c3 100644 --- a/source/client/inc/clientInt.h +++ b/source/client/inc/clientInt.h @@ -45,6 +45,11 @@ extern "C" { #define HEARTBEAT_INTERVAL 1500 // ms +enum { + CONN_TYPE__QUERY = 1, + CONN_TYPE__TMQ, +}; + typedef struct SAppInstInfo SAppInstInfo; typedef struct { @@ -132,9 +137,9 @@ typedef struct STscObj { char pass[TSDB_PASSWORD_LEN]; char db[TSDB_DB_FNAME_LEN]; char ver[128]; + int8_t connType; int32_t acctId; uint32_t connId; - int32_t connType; uint64_t id; // ref ID returned by taosAddRef TdThreadMutex mutex; // used to protect the operation on db int32_t numOfReqs; // number of sqlObj bound to this connection @@ -272,7 +277,7 @@ void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet); void initMsgHandleFp(); TAOS* taos_connect_internal(const char* ip, const char* user, const char* pass, const char* auth, const char* db, - uint16_t port); + uint16_t port, int connType); int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery); int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArray* pNodeList); diff --git a/source/client/src/clientHb.c b/source/client/src/clientHb.c index d389fc34c6..82788b2e11 100644 --- a/source/client/src/clientHb.c +++ b/source/client/src/clientHb.c @@ -23,6 +23,8 @@ static SClientHbMgr clientHbMgr = {0}; static int32_t hbCreateThread(); static void hbStopThread(); +static int32_t hbMqHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq *req) { return 0; } + static int32_t hbMqHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) { return 0; } static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog *pCatalog) { @@ -297,11 +299,10 @@ int32_t hbQueryHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq *req return TSDB_CODE_SUCCESS; } -int32_t hbMqHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq *req) { return 0; } - void hbMgrInitMqHbHandle() { clientHbMgr.reqHandle[HEARTBEAT_TYPE_QUERY] = hbQueryHbReqHandle; clientHbMgr.reqHandle[HEARTBEAT_TYPE_MQ] = hbMqHbReqHandle; + clientHbMgr.rspHandle[HEARTBEAT_TYPE_QUERY] = hbQueryHbRspHandle; clientHbMgr.rspHandle[HEARTBEAT_TYPE_MQ] = hbMqHbRspHandle; } @@ -438,7 +439,7 @@ static int32_t hbCreateThread() { if (taosThreadCreate(&clientHbMgr.thread, &thAttr, hbThreadFunc, NULL) != 0) { terrno = TAOS_SYSTEM_ERROR(errno); return -1; - } + } taosThreadAttrDestroy(&thAttr); return 0; } @@ -568,7 +569,7 @@ int hbRegisterConnImpl(SAppHbMgr *pAppHbMgr, SClientHbKey connKey, SHbConnInfo * int hbRegisterConn(SAppHbMgr *pAppHbMgr, int32_t connId, int64_t clusterId, int32_t hbType) { SClientHbKey connKey = { .connId = connId, - .hbType = HEARTBEAT_TYPE_QUERY, + .hbType = hbType, }; SHbConnInfo info = {0}; @@ -578,16 +579,14 @@ int hbRegisterConn(SAppHbMgr *pAppHbMgr, int32_t connId, int64_t clusterId, int3 *pClusterId = clusterId; info.param = pClusterId; - break; + return hbRegisterConnImpl(pAppHbMgr, connKey, &info); } case HEARTBEAT_TYPE_MQ: { - break; + return 0; } default: - break; + return 0; } - - return hbRegisterConnImpl(pAppHbMgr, connKey, &info); } void hbDeregisterConn(SAppHbMgr *pAppHbMgr, SClientHbKey connKey) { diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c index 63bb3637cf..d09205d088 100644 --- a/source/client/src/clientImpl.c +++ b/source/client/src/clientImpl.c @@ -11,7 +11,7 @@ #include "tref.h" static int32_t initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSet); -static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest); +static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest, int8_t connType); static void destroySendMsgInfo(SMsgSendInfo* pMsgBody); static bool stringLengthCheck(const char* str, size_t maxsize) { @@ -40,10 +40,10 @@ static char* getClusterKey(const char* user, const char* auth, const char* ip, i } static STscObj* taosConnectImpl(const char* user, const char* auth, const char* db, __taos_async_fn_t fp, void* param, - SAppInstInfo* pAppInfo); + SAppInstInfo* pAppInfo, int connType); TAOS* taos_connect_internal(const char* ip, const char* user, const char* pass, const char* auth, const char* db, - uint16_t port) { + uint16_t port, int connType) { if (taos_init() != TSDB_CODE_SUCCESS) { return NULL; } @@ -111,7 +111,7 @@ TAOS* taos_connect_internal(const char* ip, const char* user, const char* pass, taosThreadMutexUnlock(&appInfo.mutex); taosMemoryFreeClear(key); - return taosConnectImpl(user, &secretEncrypt[0], localDb, NULL, NULL, *pInst); + return taosConnectImpl(user, &secretEncrypt[0], localDb, NULL, NULL, *pInst, connType); } int32_t buildRequest(STscObj* pTscObj, const char* sql, int sqlLen, SRequestObj** pRequest) { @@ -425,7 +425,7 @@ int initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSe } STscObj* taosConnectImpl(const char* user, const char* auth, const char* db, __taos_async_fn_t fp, void* param, - SAppInstInfo* pAppInfo) { + SAppInstInfo* pAppInfo, int connType) { STscObj* pTscObj = createTscObj(user, auth, db, pAppInfo); if (NULL == pTscObj) { terrno = TSDB_CODE_TSC_OUT_OF_MEMORY; @@ -439,7 +439,7 @@ STscObj* taosConnectImpl(const char* user, const char* auth, const char* db, __t return NULL; } - SMsgSendInfo* body = buildConnectMsg(pRequest); + SMsgSendInfo* body = buildConnectMsg(pRequest, connType); int64_t transporterId = 0; asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &transporterId, body); @@ -462,7 +462,7 @@ STscObj* taosConnectImpl(const char* user, const char* auth, const char* db, __t return pTscObj; } -static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest) { +static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest, int8_t connType) { SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo)); if (pMsgSendInfo == NULL) { terrno = TSDB_CODE_TSC_OUT_OF_MEMORY; @@ -485,6 +485,7 @@ static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest) { } taosMemoryFreeClear(db); + connectReq.connType = connType; connectReq.pid = htonl(appInfo.pid); connectReq.startTime = htobe64(appInfo.startTime); tstrncpy(connectReq.app, appInfo.appName, sizeof(connectReq.app)); @@ -570,7 +571,7 @@ TAOS* taos_connect_auth(const char* ip, const char* user, const char* auth, cons return NULL; } - return taos_connect_internal(ip, user, NULL, auth, db, port); + return taos_connect_internal(ip, user, NULL, auth, db, port, CONN_TYPE__QUERY); } TAOS* taos_connect_l(const char* ip, int ipLen, const char* user, int userLen, const char* pass, int passLen, diff --git a/source/client/src/clientMain.c b/source/client/src/clientMain.c index fa55e9c295..7bc29e2427 100644 --- a/source/client/src/clientMain.c +++ b/source/client/src/clientMain.c @@ -87,7 +87,7 @@ TAOS *taos_connect(const char *ip, const char *user, const char *pass, const cha pass = TSDB_DEFAULT_PASS; } - return taos_connect_internal(ip, user, pass, NULL, db, port); + return taos_connect_internal(ip, user, pass, NULL, db, port, CONN_TYPE__QUERY); } void taos_close(TAOS *taos) { @@ -124,8 +124,10 @@ const char *taos_errstr(TAOS_RES *res) { } void taos_free_result(TAOS_RES *res) { - SRequestObj *pRequest = (SRequestObj *)res; - destroyRequest(pRequest); + if (TD_RES_QUERY(res)) { + SRequestObj *pRequest = (SRequestObj *)res; + destroyRequest(pRequest); + } } int taos_field_count(TAOS_RES *res) { diff --git a/source/client/src/clientMsgHandler.c b/source/client/src/clientMsgHandler.c index 99c6e81551..960cf1620a 100644 --- a/source/client/src/clientMsgHandler.c +++ b/source/client/src/clientMsgHandler.c @@ -13,13 +13,13 @@ * along with this program. If not, see . */ -#include "os.h" -#include "tdef.h" -#include "tname.h" +#include "catalog.h" #include "clientInt.h" #include "clientLog.h" -#include "catalog.h" +#include "os.h" #include "query.h" +#include "tdef.h" +#include "tname.h" int32_t (*handleRequestRspFp[TDMT_MAX])(void*, const SDataBuf* pMsg, int32_t code); @@ -69,9 +69,9 @@ int32_t processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) { pTscObj->pAppInfo->clusterId = connectRsp.clusterId; atomic_add_fetch_64(&pTscObj->pAppInfo->numOfConns, 1); - pTscObj->connType = HEARTBEAT_TYPE_QUERY; + pTscObj->connType = connectRsp.connType; - hbRegisterConn(pTscObj->pAppInfo->pAppHbMgr, connectRsp.connId, connectRsp.clusterId, HEARTBEAT_TYPE_QUERY); + hbRegisterConn(pTscObj->pAppInfo->pAppHbMgr, connectRsp.connId, connectRsp.clusterId, connectRsp.connType); // pRequest->body.resInfo.pRspMsg = pMsg->pData; tscDebug("0x%" PRIx64 " clusterId:%" PRId64 ", totalConn:%" PRId64, pRequest->requestId, connectRsp.clusterId, @@ -82,13 +82,13 @@ int32_t processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) { return 0; } -SMsgSendInfo* buildMsgInfoImpl(SRequestObj *pRequest) { +SMsgSendInfo* buildMsgInfoImpl(SRequestObj* pRequest) { SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo)); pMsgSendInfo->requestObjRefId = pRequest->self; - pMsgSendInfo->requestId = pRequest->requestId; - pMsgSendInfo->param = pRequest; - pMsgSendInfo->msgType = pRequest->type; + pMsgSendInfo->requestId = pRequest->requestId; + pMsgSendInfo->param = pRequest; + pMsgSendInfo->msgType = pRequest->type; if (pRequest->type == TDMT_MND_SHOW_RETRIEVE || pRequest->type == TDMT_VND_SHOW_TABLES_FETCH) { if (pRequest->type == TDMT_MND_SHOW_RETRIEVE) { @@ -119,7 +119,9 @@ SMsgSendInfo* buildMsgInfoImpl(SRequestObj *pRequest) { pMsgSendInfo->msgInfo = pRequest->body.requestMsg; } - pMsgSendInfo->fp = (handleRequestRspFp[TMSG_INDEX(pRequest->type)] == NULL)? genericRspCallback:handleRequestRspFp[TMSG_INDEX(pRequest->type)]; + pMsgSendInfo->fp = (handleRequestRspFp[TMSG_INDEX(pRequest->type)] == NULL) + ? genericRspCallback + : handleRequestRspFp[TMSG_INDEX(pRequest->type)]; return pMsgSendInfo; } @@ -133,7 +135,7 @@ int32_t processShowRsp(void* param, const SDataBuf* pMsg, int32_t code) { SShowRsp showRsp = {0}; tDeserializeSShowRsp(pMsg->pData, pMsg->len, &showRsp); - STableMetaRsp *pMetaMsg = &showRsp.tableMeta; + STableMetaRsp* pMetaMsg = &showRsp.tableMeta; taosMemoryFreeClear(pRequest->body.resInfo.pRspMsg); pRequest->body.resInfo.pRspMsg = pMsg->pData; @@ -159,7 +161,7 @@ int32_t processShowRsp(void* param, const SDataBuf* pMsg, int32_t code) { if (pRequest->type == TDMT_VND_SHOW_TABLES) { SShowReqInfo* pShowInfo = &pRequest->body.showInfo; - int32_t index = pShowInfo->currentIndex; + int32_t index = pShowInfo->currentIndex; SVgroupInfo* pInfo = taosArrayGet(pShowInfo->pArray, index); pShowInfo->vgId = pInfo->vgId; } @@ -169,8 +171,8 @@ int32_t processShowRsp(void* param, const SDataBuf* pMsg, int32_t code) { } int32_t processRetrieveMnodeRsp(void* param, const SDataBuf* pMsg, int32_t code) { - SRequestObj *pRequest = param; - SReqResultInfo *pResInfo = &pRequest->body.resInfo; + SRequestObj* pRequest = param; + SReqResultInfo* pResInfo = &pRequest->body.resInfo; taosMemoryFreeClear(pResInfo->pRspMsg); if (code != TSDB_CODE_SUCCESS) { @@ -181,19 +183,19 @@ int32_t processRetrieveMnodeRsp(void* param, const SDataBuf* pMsg, int32_t code) assert(pMsg->len >= sizeof(SRetrieveTableRsp)); - SRetrieveTableRsp *pRetrieve = (SRetrieveTableRsp *) pMsg->pData; - pRetrieve->numOfRows = htonl(pRetrieve->numOfRows); - pRetrieve->precision = htons(pRetrieve->precision); + SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)pMsg->pData; + pRetrieve->numOfRows = htonl(pRetrieve->numOfRows); + pRetrieve->precision = htons(pRetrieve->precision); - pResInfo->pRspMsg = pMsg->pData; + pResInfo->pRspMsg = pMsg->pData; pResInfo->numOfRows = pRetrieve->numOfRows; - pResInfo->pData = pRetrieve->data; + pResInfo->pData = pRetrieve->data; pResInfo->completed = pRetrieve->completed; pResInfo->current = 0; -// setResultDataPtr(pResInfo, pResInfo->fields, pResInfo->numOfCols, pResInfo->numOfRows); + // setResultDataPtr(pResInfo, pResInfo->fields, pResInfo->numOfCols, pResInfo->numOfRows); - tscDebug("0x%"PRIx64" numOfRows:%d, complete:%d, qId:0x%"PRIx64, pRequest->self, pRetrieve->numOfRows, + tscDebug("0x%" PRIx64 " numOfRows:%d, complete:%d, qId:0x%" PRIx64, pRequest->self, pRetrieve->numOfRows, pRetrieve->completed, pRequest->body.showInfo.execId); tsem_post(&pRequest->body.rspSem); @@ -214,20 +216,20 @@ int32_t processRetrieveVndRsp(void* param, const SDataBuf* pMsg, int32_t code) { assert(pMsg->len >= sizeof(SRetrieveTableRsp)); - pResInfo->pRspMsg = pMsg->pData; + pResInfo->pRspMsg = pMsg->pData; - SVShowTablesFetchRsp *pFetchRsp = (SVShowTablesFetchRsp *) pMsg->pData; - pFetchRsp->numOfRows = htonl(pFetchRsp->numOfRows); - pFetchRsp->precision = htons(pFetchRsp->precision); + SVShowTablesFetchRsp* pFetchRsp = (SVShowTablesFetchRsp*)pMsg->pData; + pFetchRsp->numOfRows = htonl(pFetchRsp->numOfRows); + pFetchRsp->precision = htons(pFetchRsp->precision); - pResInfo->pRspMsg = pMsg->pData; + pResInfo->pRspMsg = pMsg->pData; pResInfo->numOfRows = pFetchRsp->numOfRows; - pResInfo->pData = pFetchRsp->data; + pResInfo->pData = pFetchRsp->data; pResInfo->current = 0; -// setResultDataPtr(pResInfo, pResInfo->fields, pResInfo->numOfCols, pResInfo->numOfRows); + // setResultDataPtr(pResInfo, pResInfo->fields, pResInfo->numOfCols, pResInfo->numOfRows); - tscDebug("0x%"PRIx64" numOfRows:%d, complete:%d, qId:0x%"PRIx64, pRequest->self, pFetchRsp->numOfRows, + tscDebug("0x%" PRIx64 " numOfRows:%d, complete:%d, qId:0x%" PRIx64, pRequest->self, pFetchRsp->numOfRows, pFetchRsp->completed, pRequest->body.showInfo.execId); tsem_post(&pRequest->body.rspSem); @@ -251,18 +253,19 @@ int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) { if (TSDB_CODE_MND_DB_NOT_EXIST == code) { SUseDbRsp usedbRsp = {0}; tDeserializeSUseDbRsp(pMsg->pData, pMsg->len, &usedbRsp); - struct SCatalog *pCatalog = NULL; + struct SCatalog* pCatalog = NULL; if (usedbRsp.vgVersion >= 0) { int32_t code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog); if (code != TSDB_CODE_SUCCESS) { - tscWarn("catalogGetHandle failed, clusterId:%"PRIx64", error:%s", pRequest->pTscObj->pAppInfo->clusterId, tstrerror(code)); + tscWarn("catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", pRequest->pTscObj->pAppInfo->clusterId, + tstrerror(code)); } else { catalogRemoveDB(pCatalog, usedbRsp.db, usedbRsp.uid); } } - tFreeSUsedbRsp(&usedbRsp); + tFreeSUsedbRsp(&usedbRsp); } if (code != TSDB_CODE_SUCCESS) { @@ -276,7 +279,7 @@ int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) { tDeserializeSUseDbRsp(pMsg->pData, pMsg->len, &usedbRsp); SName name = {0}; - tNameFromString(&name, usedbRsp.db, T_NAME_ACCT|T_NAME_DB); + tNameFromString(&name, usedbRsp.db, T_NAME_ACCT | T_NAME_DB); SUseDbOutput output = {0}; code = queryBuildUseDbOutput(&output, &usedbRsp); @@ -288,11 +291,12 @@ int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) { tscError("failed to build use db output since %s", terrstr()); } else { - struct SCatalog *pCatalog = NULL; - + struct SCatalog* pCatalog = NULL; + int32_t code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog); if (code != TSDB_CODE_SUCCESS) { - tscWarn("catalogGetHandle failed, clusterId:%"PRIx64", error:%s", pRequest->pTscObj->pAppInfo->clusterId, tstrerror(code)); + tscWarn("catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", pRequest->pTscObj->pAppInfo->clusterId, + tstrerror(code)); } else { catalogUpdateDBVgInfo(pCatalog, output.db, output.dbId, output.dbVgroup); } @@ -419,14 +423,14 @@ void initMsgHandleFp() { tscProcessMsgRsp[TSDB_SQL_SHOW_CREATE_DATABASE] = tscProcessShowCreateRsp; #endif - handleRequestRspFp[TMSG_INDEX(TDMT_MND_CONNECT)] = processConnectRsp; - handleRequestRspFp[TMSG_INDEX(TDMT_MND_SHOW)] = processShowRsp; + handleRequestRspFp[TMSG_INDEX(TDMT_MND_CONNECT)] = processConnectRsp; + handleRequestRspFp[TMSG_INDEX(TDMT_MND_SHOW)] = processShowRsp; handleRequestRspFp[TMSG_INDEX(TDMT_MND_SHOW_RETRIEVE)] = processRetrieveMnodeRsp; - handleRequestRspFp[TMSG_INDEX(TDMT_MND_CREATE_DB)] = processCreateDbRsp; - handleRequestRspFp[TMSG_INDEX(TDMT_MND_USE_DB)] = processUseDbRsp; - handleRequestRspFp[TMSG_INDEX(TDMT_MND_CREATE_STB)] = processCreateTableRsp; - handleRequestRspFp[TMSG_INDEX(TDMT_MND_DROP_DB)] = processDropDbRsp; + handleRequestRspFp[TMSG_INDEX(TDMT_MND_CREATE_DB)] = processCreateDbRsp; + handleRequestRspFp[TMSG_INDEX(TDMT_MND_USE_DB)] = processUseDbRsp; + handleRequestRspFp[TMSG_INDEX(TDMT_MND_CREATE_STB)] = processCreateTableRsp; + handleRequestRspFp[TMSG_INDEX(TDMT_MND_DROP_DB)] = processDropDbRsp; - handleRequestRspFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES)] = processShowRsp; - handleRequestRspFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES_FETCH)] = processRetrieveVndRsp; + handleRequestRspFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES)] = processShowRsp; + handleRequestRspFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES_FETCH)] = processRetrieveVndRsp; } diff --git a/source/client/src/tmq.c b/source/client/src/tmq.c index 2b69b47865..c271e7810f 100644 --- a/source/client/src/tmq.c +++ b/source/client/src/tmq.c @@ -357,7 +357,15 @@ tmq_t* tmq_consumer_new1(tmq_conf_t* conf, char* errstr, int32_t errstrLen) { if (pTmq == NULL) { return NULL; } - pTmq->pTscObj = taos_connect(conf->ip, conf->user, conf->pass, conf->db, conf->port); + const char* user = conf->user == NULL ? TSDB_DEFAULT_USER : conf->user; + const char* pass = conf->pass == NULL ? TSDB_DEFAULT_PASS : conf->pass; + + ASSERT(user); + ASSERT(pass); + ASSERT(conf->db); + + pTmq->pTscObj = taos_connect_internal(conf->ip, user, pass, NULL, conf->db, conf->port, CONN_TYPE__TMQ); + if (pTmq->pTscObj == NULL) return NULL; pTmq->inWaiting = 0; pTmq->status = 0; diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index 08010db1d8..4d0a2b5e0d 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -2532,6 +2532,7 @@ int32_t tSerializeSConnectReq(void *buf, int32_t bufLen, SConnectReq *pReq) { tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER); if (tStartEncode(&encoder) < 0) return -1; + if (tEncodeI8(&encoder, pReq->connType) < 0) return -1; if (tEncodeI32(&encoder, pReq->pid) < 0) return -1; if (tEncodeCStr(&encoder, pReq->app) < 0) return -1; if (tEncodeCStr(&encoder, pReq->db) < 0) return -1; @@ -2548,6 +2549,7 @@ int32_t tDeserializeSConnectReq(void *buf, int32_t bufLen, SConnectReq *pReq) { tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_DECODER); if (tStartDecode(&decoder) < 0) return -1; + if (tDecodeI8(&decoder, &pReq->connType) < 0) return -1; if (tDecodeI32(&decoder, &pReq->pid) < 0) return -1; if (tDecodeCStrTo(&decoder, pReq->app) < 0) return -1; if (tDecodeCStrTo(&decoder, pReq->db) < 0) return -1; @@ -2567,6 +2569,7 @@ int32_t tSerializeSConnectRsp(void *buf, int32_t bufLen, SConnectRsp *pRsp) { if (tEncodeI64(&encoder, pRsp->clusterId) < 0) return -1; if (tEncodeI32(&encoder, pRsp->connId) < 0) return -1; if (tEncodeI8(&encoder, pRsp->superUser) < 0) return -1; + if (tEncodeI8(&encoder, pRsp->connType) < 0) return -1; if (tEncodeSEpSet(&encoder, &pRsp->epSet) < 0) return -1; if (tEncodeCStr(&encoder, pRsp->sVersion) < 0) return -1; tEndEncode(&encoder); @@ -2585,6 +2588,7 @@ int32_t tDeserializeSConnectRsp(void *buf, int32_t bufLen, SConnectRsp *pRsp) { if (tDecodeI64(&decoder, &pRsp->clusterId) < 0) return -1; if (tDecodeI32(&decoder, &pRsp->connId) < 0) return -1; if (tDecodeI8(&decoder, &pRsp->superUser) < 0) return -1; + if (tDecodeI8(&decoder, &pRsp->connType) < 0) return -1; if (tDecodeSEpSet(&decoder, &pRsp->epSet) < 0) return -1; if (tDecodeCStrTo(&decoder, pRsp->sVersion) < 0) return -1; tEndDecode(&decoder); diff --git a/source/dnode/mnode/impl/src/mndProfile.c b/source/dnode/mnode/impl/src/mndProfile.c index 320671c332..974c88a752 100644 --- a/source/dnode/mnode/impl/src/mndProfile.c +++ b/source/dnode/mnode/impl/src/mndProfile.c @@ -23,13 +23,14 @@ #include "tglobal.h" #include "version.h" -#define QUERY_ID_SIZE 20 -#define QUERY_OBJ_ID_SIZE 18 +#define QUERY_ID_SIZE 20 +#define QUERY_OBJ_ID_SIZE 18 #define SUBQUERY_INFO_SIZE 6 -#define QUERY_SAVE_SIZE 20 +#define QUERY_SAVE_SIZE 20 typedef struct { int32_t id; + int8_t connType; char user[TSDB_USER_LEN]; char app[TSDB_APP_NAME_LEN]; // app name that invokes taosc int64_t appStartTimeMs; // app start time @@ -44,8 +45,8 @@ typedef struct { SQueryDesc *pQueries; } SConnObj; -static SConnObj *mndCreateConn(SMnode *pMnode, const char *user, uint32_t ip, uint16_t port, int32_t pid, - const char *app, int64_t startTime); +static SConnObj *mndCreateConn(SMnode *pMnode, const char *user, int8_t connType, uint32_t ip, uint16_t port, + int32_t pid, const char *app, int64_t startTime); static void mndFreeConn(SConnObj *pConn); static SConnObj *mndAcquireConn(SMnode *pMnode, int32_t connId); static void mndReleaseConn(SMnode *pMnode, SConnObj *pConn); @@ -93,8 +94,8 @@ void mndCleanupProfile(SMnode *pMnode) { } } -static SConnObj *mndCreateConn(SMnode *pMnode, const char *user, uint32_t ip, uint16_t port, int32_t pid, - const char *app, int64_t startTime) { +static SConnObj *mndCreateConn(SMnode *pMnode, const char *user, int8_t connType, uint32_t ip, uint16_t port, + int32_t pid, const char *app, int64_t startTime) { SProfileMgmt *pMgmt = &pMnode->profileMgmt; int32_t connId = atomic_add_fetch_32(&pMgmt->connId, 1); @@ -102,6 +103,7 @@ static SConnObj *mndCreateConn(SMnode *pMnode, const char *user, uint32_t ip, ui if (startTime == 0) startTime = taosGetTimestampMs(); SConnObj connObj = {.id = connId, + .connType = connType, .appStartTimeMs = startTime, .pid = pid, .ip = ip, @@ -159,8 +161,8 @@ static void mndReleaseConn(SMnode *pMnode, SConnObj *pConn) { } void *mndGetNextConn(SMnode *pMnode, SCacheIter *pIter) { - SConnObj* pConn = NULL; - bool hasNext = taosCacheIterNext(pIter); + SConnObj *pConn = NULL; + bool hasNext = taosCacheIterNext(pIter); if (hasNext) { size_t dataLen = 0; pConn = taosCacheIterGetData(pIter, &dataLen); @@ -210,8 +212,8 @@ static int32_t mndProcessConnectReq(SNodeMsg *pReq) { } } - pConn = - mndCreateConn(pMnode, pReq->user, pReq->clientIp, pReq->clientPort, connReq.pid, connReq.app, connReq.startTime); + pConn = mndCreateConn(pMnode, pReq->user, connReq.connType, pReq->clientIp, pReq->clientPort, connReq.pid, + connReq.app, connReq.startTime); if (pConn == NULL) { mError("user:%s, failed to login from %s while create connection since %s", pReq->user, ip, terrstr()); goto CONN_OVER; @@ -222,6 +224,7 @@ static int32_t mndProcessConnectReq(SNodeMsg *pReq) { connectRsp.superUser = pUser->superUser; connectRsp.clusterId = pMnode->clusterId; connectRsp.connId = pConn->id; + connectRsp.connType = connReq.connType; snprintf(connectRsp.sVersion, sizeof(connectRsp.sVersion), "ver:%s\nbuild:%s\ngitinfo:%s", version, buildinfo, gitinfo); @@ -343,7 +346,6 @@ static int32_t mndProcessHeartBeatReq(SNodeMsg *pReq) { return -1; } - SClientHbBatchRsp batchRsp = {0}; batchRsp.rsps = taosArrayInit(0, sizeof(SClientHbRsp)); @@ -917,4 +919,4 @@ static void mndCancelGetNextQuery(SMnode *pMnode, void *pIter) { int32_t mndGetNumOfConnections(SMnode *pMnode) { SProfileMgmt *pMgmt = &pMnode->profileMgmt; return taosCacheGetNumOfObj(pMgmt->cache); -} \ No newline at end of file +} diff --git a/tests/test/c/tmqDemo.c b/tests/test/c/tmqDemo.c index 756893f217..3824471ea1 100644 --- a/tests/test/c/tmqDemo.c +++ b/tests/test/c/tmqDemo.c @@ -13,76 +13,69 @@ * along with this program. If not, see . */ -// clang-format off - #include +#include #include +#include #include -#include #include #include +#include #include -#include -#include #include "taos.h" #include "taoserror.h" #include "tlog.h" -#define GREEN "\033[1;32m" -#define NC "\033[0m" +#define GREEN "\033[1;32m" +#define NC "\033[0m" #define min(a, b) (((a) < (b)) ? (a) : (b)) -#define MAX_SQL_STR_LEN (1024 * 1024) -#define MAX_ROW_STR_LEN (16 * 1024) +#define MAX_SQL_STR_LEN (1024 * 1024) +#define MAX_ROW_STR_LEN (16 * 1024) -enum _RUN_MODE { - TMQ_RUN_INSERT_AND_CONSUME, - TMQ_RUN_ONLY_INSERT, - TMQ_RUN_ONLY_CONSUME, - TMQ_RUN_MODE_BUTT -}; +enum _RUN_MODE { TMQ_RUN_INSERT_AND_CONSUME, TMQ_RUN_ONLY_INSERT, TMQ_RUN_ONLY_CONSUME, TMQ_RUN_MODE_BUTT }; typedef struct { - char dbName[32]; - char stbName[64]; - char resultFileName[256]; - char vnodeWalPath[256]; - int32_t numOfThreads; - int32_t numOfTables; - int32_t numOfVgroups; - int32_t runMode; - int32_t numOfColumn; - double ratio; - int32_t batchNumOfRow; - int32_t totalRowsOfPerTbl; - int64_t startTimestamp; - int32_t showMsgFlag; - int32_t simCase; - - int32_t totalRowsOfT2; + char dbName[32]; + char stbName[64]; + char resultFileName[256]; + char vnodeWalPath[256]; + int32_t numOfThreads; + int32_t numOfTables; + int32_t numOfVgroups; + int32_t runMode; + int32_t numOfColumn; + double ratio; + int32_t batchNumOfRow; + int32_t totalRowsOfPerTbl; + int64_t startTimestamp; + int32_t showMsgFlag; + int32_t simCase; + + int32_t totalRowsOfT2; } SConfInfo; static SConfInfo g_stConfInfo = { "tmqdb", "stb", - "./tmqResult.txt", // output_file + "./tmqResult.txt", // output_file "", // /data2/dnode/data/vnode/vnode2/wal", - 1, // threads - 1, // tables - 1, // vgroups - 0, // run mode - 1, // columns - 1, // ratio - 1, // batch size - 10000, // total rows for per table - 0, // 2020-01-01 00:00:00.000 - 0, // show consume msg switch - 0, // if run in sim case + 1, // threads + 1, // tables + 1, // vgroups + 0, // run mode + 1, // columns + 1, // ratio + 1, // batch size + 10000, // total rows for per table + 0, // 2020-01-01 00:00:00.000 + 0, // show consume msg switch + 0, // if run in sim case 10000, }; -char* g_pRowValue = NULL; +char* g_pRowValue = NULL; TdFilePtr g_fp = NULL; static void printHelp() { @@ -125,10 +118,8 @@ static void printHelp() { exit(EXIT_SUCCESS); } -void parseArgument(int32_t argc, char *argv[]) { - +void parseArgument(int32_t argc, char* argv[]) { g_stConfInfo.startTimestamp = 1640966400000; // 2020-01-01 00:00:00.000 - for (int32_t i = 1; i < argc; i++) { if (strcmp(argv[i], "-h") == 0 || strcmp(argv[i], "--help") == 0) { @@ -156,7 +147,7 @@ void parseArgument(int32_t argc, char *argv[]) { g_stConfInfo.batchNumOfRow = atoi(argv[++i]); } else if (strcmp(argv[i], "-r") == 0) { g_stConfInfo.totalRowsOfPerTbl = atoi(argv[++i]); - } else if (strcmp(argv[i], "-l") == 0) { + } else if (strcmp(argv[i], "-l") == 0) { g_stConfInfo.numOfColumn = atoi(argv[++i]); } else if (strcmp(argv[i], "-q") == 0) { g_stConfInfo.ratio = atof(argv[++i]); @@ -168,7 +159,7 @@ void parseArgument(int32_t argc, char *argv[]) { g_stConfInfo.simCase = atol(argv[++i]); } else { printf("%s unknow para: %s %s", GREEN, argv[++i], NC); - exit(-1); + exit(-1); } } @@ -191,73 +182,71 @@ void parseArgument(int32_t argc, char *argv[]) { pPrint("%s totalRowsOfT2:%d %s", GREEN, g_stConfInfo.totalRowsOfT2, NC); pPrint("%s startTimestamp:%" PRId64" %s", GREEN, g_stConfInfo.startTimestamp, NC); pPrint("%s showMsgFlag:%d %s", GREEN, g_stConfInfo.showMsgFlag, NC); -#endif +#endif } -static int running = 1; +static int running = 1; /*static void msg_process(tmq_message_t* message) { tmqShowMsg(message); }*/ // calc dir size (not include itself 4096Byte) -int64_t getDirectorySize(char *dir) -{ - TdDirPtr pDir; - TdDirEntryPtr pDirEntry; - int64_t totalSize=0; +int64_t getDirectorySize(char* dir) { + TdDirPtr pDir; + TdDirEntryPtr pDirEntry; + int64_t totalSize = 0; - if ((pDir = taosOpenDir(dir)) == NULL) { - fprintf(stderr, "Cannot open dir: %s\n", dir); - return -1; + if ((pDir = taosOpenDir(dir)) == NULL) { + fprintf(stderr, "Cannot open dir: %s\n", dir); + return -1; + } + + // lstat(dir, &statbuf); + // totalSize+=statbuf.st_size; + + while ((pDirEntry = taosReadDir(pDir)) != NULL) { + char subdir[1024]; + char* fileName = taosGetDirEntryName(pDirEntry); + sprintf(subdir, "%s/%s", dir, fileName); + + // printf("===d_name: %s\n", entry->d_name); + if (taosIsDir(subdir)) { + if (strcmp(".", fileName) == 0 || strcmp("..", fileName) == 0) { + continue; + } + + int64_t subDirSize = getDirectorySize(subdir); + totalSize += subDirSize; + } else if (0 == strcmp(strchr(fileName, '.'), ".log")) { // only calc .log file size, and not include .idx file + int64_t file_size = 0; + taosStatFile(subdir, &file_size, NULL); + totalSize += file_size; } + } - //lstat(dir, &statbuf); - //totalSize+=statbuf.st_size; - - while ((pDirEntry = taosReadDir(pDir)) != NULL) { - char subdir[1024]; - char* fileName = taosGetDirEntryName(pDirEntry); - sprintf(subdir, "%s/%s", dir, fileName); - - //printf("===d_name: %s\n", entry->d_name); - if (taosIsDir(subdir)) { - if (strcmp(".", fileName) == 0 || strcmp("..", fileName) == 0) { - continue; - } - - int64_t subDirSize = getDirectorySize(subdir); - totalSize+=subDirSize; - } else if (0 == strcmp(strchr(fileName, '.'), ".log")) { // only calc .log file size, and not include .idx file - int64_t file_size = 0; - taosStatFile(subdir, &file_size, NULL); - totalSize+=file_size; - } - } - - taosCloseDir(pDir); - return totalSize; + taosCloseDir(pDir); + return totalSize; } - -int queryDB(TAOS *taos, char *command) { - TAOS_RES *pRes = taos_query(taos, command); - int code = taos_errno(pRes); - //if ((code != 0) && (code != TSDB_CODE_RPC_AUTH_REQUIRED)) { - if (code != 0) { - pError("failed to reason:%s, sql: %s", tstrerror(code), command); - taos_free_result(pRes); - return -1; - } - taos_free_result(pRes); - return 0 ; +int queryDB(TAOS* taos, char* command) { + TAOS_RES* pRes = taos_query(taos, command); + int code = taos_errno(pRes); + // if ((code != 0) && (code != TSDB_CODE_RPC_AUTH_REQUIRED)) { + if (code != 0) { + pError("failed to reason:%s, sql: %s", tstrerror(code), command); + taos_free_result(pRes); + return -1; + } + taos_free_result(pRes); + return 0; } int32_t init_env() { char sqlStr[1024] = {0}; - + TAOS* pConn = taos_connect(NULL, "root", "taosdata", NULL, 0); if (pConn == NULL) { return -1; } - + sprintf(sqlStr, "create database if not exists %s vgroups %d", g_stConfInfo.dbName, g_stConfInfo.numOfVgroups); TAOS_RES* pRes = taos_query(pConn, sqlStr); if (taos_errno(pRes) != 0) { @@ -282,19 +271,19 @@ int32_t init_env() { int32_t dataLen = 0; int32_t sqlLen = 0; - sqlLen += sprintf(sqlStr+sqlLen, "create stable if not exists %s (ts timestamp, ", g_stConfInfo.stbName); + sqlLen += sprintf(sqlStr + sqlLen, "create stable if not exists %s (ts timestamp, ", g_stConfInfo.stbName); for (int32_t i = 0; i < g_stConfInfo.numOfColumn; i++) { - if (i == g_stConfInfo.numOfColumn - 1) { - sqlLen += sprintf(sqlStr+sqlLen, "c%d int) ", i); - memcpy(g_pRowValue + dataLen, "66778899", strlen("66778899")); - dataLen += strlen("66778899"); - } else { - sqlLen += sprintf(sqlStr+sqlLen, "c%d int, ", i); - memcpy(g_pRowValue + dataLen, "66778899, ", strlen("66778899, ")); - dataLen += strlen("66778899, "); - } + if (i == g_stConfInfo.numOfColumn - 1) { + sqlLen += sprintf(sqlStr + sqlLen, "c%d int) ", i); + memcpy(g_pRowValue + dataLen, "66778899", strlen("66778899")); + dataLen += strlen("66778899"); + } else { + sqlLen += sprintf(sqlStr + sqlLen, "c%d int, ", i); + memcpy(g_pRowValue + dataLen, "66778899, ", strlen("66778899, ")); + dataLen += strlen("66778899, "); + } } - sqlLen += sprintf(sqlStr+sqlLen, "tags (t0 int)"); + sqlLen += sprintf(sqlStr + sqlLen, "tags (t0 int)"); pRes = taos_query(pConn, sqlStr); if (taos_errno(pRes) != 0) { @@ -313,7 +302,7 @@ int32_t init_env() { taos_free_result(pRes); } - //const char* sql = "select * from tu1"; + // const char* sql = "select * from tu1"; sprintf(sqlStr, "create topic test_stb_topic_1 as select ts,c0 from %s", g_stConfInfo.stbName); /*pRes = tmq_create_topic(pConn, "test_stb_topic_1", sqlStr, strlen(sqlStr));*/ pRes = taos_query(pConn, sqlStr); @@ -327,6 +316,7 @@ int32_t init_env() { } tmq_t* build_consumer() { +#if 0 char sqlStr[1024] = {0}; TAOS* pConn = taos_connect(NULL, "root", "taosdata", NULL, 0); @@ -338,10 +328,16 @@ tmq_t* build_consumer() { printf("error in use db, reason:%s\n", taos_errstr(pRes)); } taos_free_result(pRes); +#endif tmq_conf_t* conf = tmq_conf_new(); tmq_conf_set(conf, "group.id", "tg2"); - tmq_t* tmq = tmq_consumer_new(pConn, conf, NULL, 0); + tmq_conf_set(conf, "td.connect.user", "root"); + tmq_conf_set(conf, "td.connect.pass", "taosdata"); + tmq_conf_set(conf, "td.connect.db", g_stConfInfo.dbName); + tmq_t* tmq = tmq_consumer_new1(conf, NULL, 0); + assert(tmq); + tmq_conf_destroy(conf); return tmq; } @@ -396,20 +392,20 @@ void perf_loop(tmq_t* tmq, tmq_list_t* topics, int32_t totalMsgs, int64_t walLog if (tmqmessage) { batchCnt++; /*skipLogNum += tmqGetSkipLogNum(tmqmessage);*/ - if (0 != g_stConfInfo.showMsgFlag) { + if (0 != g_stConfInfo.showMsgFlag) { /*msg_process(tmqmessage);*/ - } + } tmq_message_destroy(tmqmessage); } else { break; } } int64_t endTime = taosGetTimestampUs(); - double consumeTime = (double)(endTime - startTime) / 1000000; + double consumeTime = (double)(endTime - startTime) / 1000000; if (batchCnt != totalMsgs) { - printf("%s inserted msgs: %d and consume msgs: %d mismatch %s", GREEN, totalMsgs, batchCnt, NC); - /*exit(-1);*/ + printf("%s inserted msgs: %d and consume msgs: %d mismatch %s", GREEN, totalMsgs, batchCnt, NC); + /*exit(-1);*/ } if (0 == g_stConfInfo.simCase) { @@ -417,12 +413,14 @@ void perf_loop(tmq_t* tmq, tmq_list_t* topics, int32_t totalMsgs, int64_t walLog } else { printf("{consume success: %d}", totalMsgs); } - taosFprintfFile(g_fp, "|%10d | %10.3f | %8.2f | %10.2f| %10.2f |\n", batchCnt, consumeTime, (double)batchCnt / consumeTime, (double)walLogSize / (1024 * 1024.0) / consumeTime, (double)walLogSize / 1024.0 / batchCnt); + taosFprintfFile(g_fp, "|%10d | %10.3f | %8.2f | %10.2f| %10.2f |\n", batchCnt, consumeTime, + (double)batchCnt / consumeTime, (double)walLogSize / (1024 * 1024.0) / consumeTime, + (double)walLogSize / 1024.0 / batchCnt); err = tmq_consumer_close(tmq); if (err) { fprintf(stderr, "%% Failed to close consumer: %s\n", tmq_err2str(err)); - exit(-1); + exit(-1); } } @@ -430,7 +428,7 @@ void perf_loop(tmq_t* tmq, tmq_list_t* topics, int32_t totalMsgs, int64_t walLog int32_t syncWriteData() { TAOS* pConn = taos_connect(NULL, "root", "taosdata", NULL, 0); if (pConn == NULL) { - return -1; + return -1; } char sqlStr[1024] = {0}; @@ -449,11 +447,11 @@ int32_t syncWriteData() { } int32_t totalMsgs = 0; - + int64_t time_counter = g_stConfInfo.startTimestamp; for (int i = 0; i < g_stConfInfo.totalRowsOfPerTbl;) { for (int tID = 0; tID <= g_stConfInfo.numOfTables - 1; tID++) { - int inserted = i; + int inserted = i; int64_t tmp_time = time_counter; int32_t data_len = 0; @@ -465,22 +463,22 @@ int32_t syncWriteData() { k++; if (inserted >= g_stConfInfo.totalRowsOfPerTbl) { - break; + break; } - if (data_len > MAX_SQL_STR_LEN - MAX_ROW_STR_LEN) { + if (data_len > MAX_SQL_STR_LEN - MAX_ROW_STR_LEN) { break; - } + } } int code = queryDB(pConn, buffer); - if (0 != code){ + if (0 != code) { fprintf(stderr, "insert data error!\n"); - taosMemoryFreeClear(buffer); - return -1; - } + taosMemoryFreeClear(buffer); + return -1; + } - totalMsgs++; + totalMsgs++; if (tID == g_stConfInfo.numOfTables - 1) { i = inserted; @@ -492,12 +490,11 @@ int32_t syncWriteData() { return totalMsgs; } - // sync insertion int32_t syncWriteDataByRatio() { TAOS* pConn = taos_connect(NULL, "root", "taosdata", NULL, 0); if (pConn == NULL) { - return -1; + return -1; } char sqlStr[1024] = {0}; @@ -518,27 +515,27 @@ int32_t syncWriteDataByRatio() { int32_t totalMsgs = 0; int32_t insertedOfT1 = 0; - int32_t insertedOfT2 = 0; + int32_t insertedOfT2 = 0; int64_t tsOfT1 = g_stConfInfo.startTimestamp; int64_t tsOfT2 = g_stConfInfo.startTimestamp; int64_t tmp_time; - + for (;;) { - if ((insertedOfT1 >= g_stConfInfo.totalRowsOfPerTbl) && (insertedOfT2 >= g_stConfInfo.totalRowsOfT2)) { + if ((insertedOfT1 >= g_stConfInfo.totalRowsOfPerTbl) && (insertedOfT2 >= g_stConfInfo.totalRowsOfT2)) { break; - } - + } + for (int tID = 0; tID <= g_stConfInfo.numOfTables - 1; tID++) { if (0 == tID) { - tmp_time = tsOfT1; + tmp_time = tsOfT1; if (insertedOfT1 >= g_stConfInfo.totalRowsOfPerTbl) { - continue; + continue; } - } else if (1 == tID){ - tmp_time = tsOfT2; + } else if (1 == tID) { + tmp_time = tsOfT2; if (insertedOfT2 >= g_stConfInfo.totalRowsOfT2) { - continue; + continue; } } @@ -548,79 +545,86 @@ int32_t syncWriteDataByRatio() { for (k = 0; k < g_stConfInfo.batchNumOfRow;) { data_len += sprintf(buffer + data_len, "(%" PRId64 ", %s) ", tmp_time++, g_pRowValue); k++; - if (0 == tID) { + if (0 == tID) { insertedOfT1++; - if (insertedOfT1 >= g_stConfInfo.totalRowsOfPerTbl) { - break; - } - } else if (1 == tID){ + if (insertedOfT1 >= g_stConfInfo.totalRowsOfPerTbl) { + break; + } + } else if (1 == tID) { insertedOfT2++; - if (insertedOfT2 >= g_stConfInfo.totalRowsOfT2) { - break; - } - } + if (insertedOfT2 >= g_stConfInfo.totalRowsOfT2) { + break; + } + } - if (data_len > MAX_SQL_STR_LEN - MAX_ROW_STR_LEN) { + if (data_len > MAX_SQL_STR_LEN - MAX_ROW_STR_LEN) { break; - } + } } int code = queryDB(pConn, buffer); - if (0 != code){ + if (0 != code) { fprintf(stderr, "insert data error!\n"); - taosMemoryFreeClear(buffer); - return -1; - } - - if (0 == tID) { - tsOfT1 = tmp_time; - } else if (1 == tID){ - tsOfT2 = tmp_time; + taosMemoryFreeClear(buffer); + return -1; } - totalMsgs++; + if (0 == tID) { + tsOfT1 = tmp_time; + } else if (1 == tID) { + tsOfT2 = tmp_time; + } + + totalMsgs++; } } - pPrint("expect insert rows: T1[%d] T2[%d], actual insert rows: T1[%d] T2[%d]\n", g_stConfInfo.totalRowsOfPerTbl, g_stConfInfo.totalRowsOfT2, insertedOfT1, insertedOfT2); + pPrint("expect insert rows: T1[%d] T2[%d], actual insert rows: T1[%d] T2[%d]\n", g_stConfInfo.totalRowsOfPerTbl, + g_stConfInfo.totalRowsOfT2, insertedOfT1, insertedOfT2); taosMemoryFreeClear(buffer); return totalMsgs; } void printParaIntoFile() { // FILE *fp = fopen(g_stConfInfo.resultFileName, "a"); - TdFilePtr pFile = taosOpenFile(g_stConfInfo.resultFileName, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND | TD_FILE_STREAM); + TdFilePtr pFile = + taosOpenFile(g_stConfInfo.resultFileName, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND | TD_FILE_STREAM); if (NULL == pFile) { fprintf(stderr, "Failed to open %s for save result\n", g_stConfInfo.resultFileName); - exit -1; + exit - 1; }; g_fp = pFile; - time_t tTime = taosGetTimestampSec(); + time_t tTime = taosGetTimestampSec(); struct tm tm = *localtime(&tTime); taosFprintfFile(pFile, "###################################################################\n"); - taosFprintfFile(pFile, "# configDir: %s\n", configDir); - taosFprintfFile(pFile, "# dbName: %s\n", g_stConfInfo.dbName); - taosFprintfFile(pFile, "# stbName: %s\n", g_stConfInfo.stbName); - taosFprintfFile(pFile, "# vnodeWalPath: %s\n", g_stConfInfo.vnodeWalPath); - taosFprintfFile(pFile, "# numOfTables: %d\n", g_stConfInfo.numOfTables); - taosFprintfFile(pFile, "# numOfThreads: %d\n", g_stConfInfo.numOfThreads); - taosFprintfFile(pFile, "# numOfVgroups: %d\n", g_stConfInfo.numOfVgroups); - taosFprintfFile(pFile, "# runMode: %d\n", g_stConfInfo.runMode); - taosFprintfFile(pFile, "# ratio: %f\n", g_stConfInfo.ratio); - taosFprintfFile(pFile, "# numOfColumn: %d\n", g_stConfInfo.numOfColumn); - taosFprintfFile(pFile, "# batchNumOfRow: %d\n", g_stConfInfo.batchNumOfRow); - taosFprintfFile(pFile, "# totalRowsOfPerTbl: %d\n", g_stConfInfo.totalRowsOfPerTbl); - taosFprintfFile(pFile, "# totalRowsOfT2: %d\n", g_stConfInfo.totalRowsOfT2); + taosFprintfFile(pFile, "# configDir: %s\n", configDir); + taosFprintfFile(pFile, "# dbName: %s\n", g_stConfInfo.dbName); + taosFprintfFile(pFile, "# stbName: %s\n", g_stConfInfo.stbName); + taosFprintfFile(pFile, "# vnodeWalPath: %s\n", g_stConfInfo.vnodeWalPath); + taosFprintfFile(pFile, "# numOfTables: %d\n", g_stConfInfo.numOfTables); + taosFprintfFile(pFile, "# numOfThreads: %d\n", g_stConfInfo.numOfThreads); + taosFprintfFile(pFile, "# numOfVgroups: %d\n", g_stConfInfo.numOfVgroups); + taosFprintfFile(pFile, "# runMode: %d\n", g_stConfInfo.runMode); + taosFprintfFile(pFile, "# ratio: %f\n", g_stConfInfo.ratio); + taosFprintfFile(pFile, "# numOfColumn: %d\n", g_stConfInfo.numOfColumn); + taosFprintfFile(pFile, "# batchNumOfRow: %d\n", g_stConfInfo.batchNumOfRow); + taosFprintfFile(pFile, "# totalRowsOfPerTbl: %d\n", g_stConfInfo.totalRowsOfPerTbl); + taosFprintfFile(pFile, "# totalRowsOfT2: %d\n", g_stConfInfo.totalRowsOfT2); taosFprintfFile(pFile, "# Test time: %d-%02d-%02d %02d:%02d:%02d\n", tm.tm_year + 1900, tm.tm_mon + 1, - tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec); + tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec); taosFprintfFile(pFile, "###################################################################\n"); - taosFprintfFile(pFile, "|-------------------------------insert info-----------------------------|--------------------------------consume info---------------------------------|\n"); - taosFprintfFile(pFile, "|batch size| insert msgs | insert time(s) | msgs/s | walLogSize(MB) | consume msgs | consume time(s) | msgs/s | MB/s | avg msg size(KB) |\n"); + taosFprintfFile(pFile, + "|-------------------------------insert " + "info-----------------------------|--------------------------------consume " + "info---------------------------------|\n"); + taosFprintfFile(pFile, + "|batch size| insert msgs | insert time(s) | msgs/s | walLogSize(MB) | consume msgs | consume " + "time(s) | msgs/s | MB/s | avg msg size(KB) |\n"); taosFprintfFile(g_fp, "|%10d", g_stConfInfo.batchNumOfRow); } -int main(int32_t argc, char *argv[]) { +int main(int32_t argc, char* argv[]) { parseArgument(argc, argv); printParaIntoFile(); @@ -630,70 +634,70 @@ int main(int32_t argc, char *argv[]) { code = init_env(); if (code != 0) { fprintf(stderr, "%% init_env error!\n"); - return -1; + return -1; } int32_t totalMsgs = 0; if (g_stConfInfo.runMode != TMQ_RUN_ONLY_CONSUME) { - int64_t startTs = taosGetTimestampUs(); if (1 == g_stConfInfo.ratio) { - totalMsgs = syncWriteData(); + totalMsgs = syncWriteData(); } else { - totalMsgs = syncWriteDataByRatio(); - } - - if (totalMsgs <= 0) { - pError("inset data error!\n"); - return -1; - } - int64_t endTs = taosGetTimestampUs(); - int64_t delay = endTs - startTs; - - int32_t totalRows = 0; - if (1 == g_stConfInfo.ratio) { - totalRows = g_stConfInfo.totalRowsOfPerTbl * g_stConfInfo.numOfTables; - } else { - totalRows = g_stConfInfo.totalRowsOfPerTbl * (1 + g_stConfInfo.ratio); - } - - float seconds = delay / 1000000.0; - float rowsSpeed = totalRows / seconds; - float msgsSpeed = totalMsgs / seconds; - - - if ((0 == g_stConfInfo.simCase) && (strlen(g_stConfInfo.vnodeWalPath))) { - walLogSize = getDirectorySize(g_stConfInfo.vnodeWalPath); - if (walLogSize <= 0) { - printf("%s size incorrect!", g_stConfInfo.vnodeWalPath); - exit(-1); - } else { - pPrint(".log file size in vnode2/wal: %.3f MBytes\n", (double)walLogSize/(1024 * 1024.0)); - } - } - - if (0 == g_stConfInfo.simCase) { - pPrint("insert result: %d rows, %d msgs, time:%.3f sec, speed:%.1f rows/second, %.1f msgs/second\n", totalRows, totalMsgs, seconds, rowsSpeed, msgsSpeed); + totalMsgs = syncWriteDataByRatio(); } - taosFprintfFile(g_fp, "|%10d | %10.3f | %8.2f | %10.3f ", totalMsgs, seconds, msgsSpeed, (double)walLogSize/(1024 * 1024.0)); + + if (totalMsgs <= 0) { + pError("inset data error!\n"); + return -1; + } + int64_t endTs = taosGetTimestampUs(); + int64_t delay = endTs - startTs; + + int32_t totalRows = 0; + if (1 == g_stConfInfo.ratio) { + totalRows = g_stConfInfo.totalRowsOfPerTbl * g_stConfInfo.numOfTables; + } else { + totalRows = g_stConfInfo.totalRowsOfPerTbl * (1 + g_stConfInfo.ratio); + } + + float seconds = delay / 1000000.0; + float rowsSpeed = totalRows / seconds; + float msgsSpeed = totalMsgs / seconds; + + if ((0 == g_stConfInfo.simCase) && (strlen(g_stConfInfo.vnodeWalPath))) { + walLogSize = getDirectorySize(g_stConfInfo.vnodeWalPath); + if (walLogSize <= 0) { + printf("%s size incorrect!", g_stConfInfo.vnodeWalPath); + exit(-1); + } else { + pPrint(".log file size in vnode2/wal: %.3f MBytes\n", (double)walLogSize / (1024 * 1024.0)); + } + } + + if (0 == g_stConfInfo.simCase) { + pPrint("insert result: %d rows, %d msgs, time:%.3f sec, speed:%.1f rows/second, %.1f msgs/second\n", totalRows, + totalMsgs, seconds, rowsSpeed, msgsSpeed); + } + taosFprintfFile(g_fp, "|%10d | %10.3f | %8.2f | %10.3f ", totalMsgs, seconds, msgsSpeed, + (double)walLogSize / (1024 * 1024.0)); } if (g_stConfInfo.runMode == TMQ_RUN_ONLY_INSERT) { return 0; } - - tmq_t* tmq = build_consumer(); + + tmq_t* tmq = build_consumer(); tmq_list_t* topic_list = build_topic_list(); - if ((NULL == tmq) || (NULL == topic_list)){ + if ((NULL == tmq) || (NULL == topic_list)) { return -1; } - + perf_loop(tmq, topic_list, totalMsgs, walLogSize); taosMemoryFreeClear(g_pRowValue); - taosFprintfFile(g_fp, "\n"); - taosCloseFile(&g_fp); + taosFprintfFile(g_fp, "\n"); + taosCloseFile(&g_fp); return 0; } diff --git a/tests/test/c/tmqSim.c b/tests/test/c/tmqSim.c index 7862788a71..eaca8f151e 100644 --- a/tests/test/c/tmqSim.c +++ b/tests/test/c/tmqSim.c @@ -35,8 +35,8 @@ #define MAX_ROW_STR_LEN (16 * 1024) typedef struct { - int32_t expectMsgCnt; - int32_t consumeMsgCnt; + int32_t expectMsgCnt; + int32_t consumeMsgCnt; TdThread thread; } SThreadInfo; @@ -45,12 +45,12 @@ typedef struct { char dbName[32]; char topicString[256]; char keyString[1024]; - char topicString1[256]; - char keyString1[1024]; + char topicString1[256]; + char keyString1[1024]; int32_t showMsgFlag; int32_t consumeDelay; // unit s int32_t consumeMsgCnt; - int32_t checkMode; + int32_t checkMode; // save result after parse agrvs int32_t numOfTopic; @@ -59,13 +59,13 @@ typedef struct { int32_t numOfKey; char key[32][64]; char value[32][64]; - - int32_t numOfTopic1; - char topics1[32][64]; - - int32_t numOfKey1; - char key1[32][64]; - char value1[32][64]; + + int32_t numOfTopic1; + char topics1[32][64]; + + int32_t numOfKey1; + char key1[32][64]; + char value1[32][64]; } SConfInfo; static SConfInfo g_stConfInfo; @@ -186,18 +186,18 @@ void parseInputString() { ltrim(g_stConfInfo.topics[g_stConfInfo.numOfTopic]); // printf("%s\n", g_stConfInfo.topics[g_stConfInfo.numOfTopic]); g_stConfInfo.numOfTopic++; - + token = strtok(NULL, delim); - } + } token = strtok(g_stConfInfo.topicString1, delim); - while(token != NULL) { - //printf("%s\n", token ); - strcpy(g_stConfInfo.topics1[g_stConfInfo.numOfTopic1], token); + while (token != NULL) { + // printf("%s\n", token ); + strcpy(g_stConfInfo.topics1[g_stConfInfo.numOfTopic1], token); ltrim(g_stConfInfo.topics1[g_stConfInfo.numOfTopic1]); - //printf("%s\n", g_stConfInfo.topics[g_stConfInfo.numOfTopic]); - g_stConfInfo.numOfTopic1++; - + // printf("%s\n", g_stConfInfo.topics[g_stConfInfo.numOfTopic]); + g_stConfInfo.numOfTopic1++; + token = strtok(NULL, delim); } @@ -214,28 +214,28 @@ void parseInputString() { // g_stConfInfo.value[g_stConfInfo.numOfKey]); g_stConfInfo.numOfKey++; } - + token = strtok(NULL, delim); } token = strtok(g_stConfInfo.keyString1, delim); - while(token != NULL) { - //printf("%s\n", token ); - { - char* pstr = token; - ltrim(pstr); - char *ret = strchr(pstr, ch); - memcpy(g_stConfInfo.key1[g_stConfInfo.numOfKey1], pstr, ret-pstr); - strcpy(g_stConfInfo.value1[g_stConfInfo.numOfKey1], ret+1); - //printf("key: %s, value: %s\n", g_stConfInfo.key[g_stConfInfo.numOfKey], g_stConfInfo.value[g_stConfInfo.numOfKey]); - g_stConfInfo.numOfKey1++; + while (token != NULL) { + // printf("%s\n", token ); + { + char* pstr = token; + ltrim(pstr); + char* ret = strchr(pstr, ch); + memcpy(g_stConfInfo.key1[g_stConfInfo.numOfKey1], pstr, ret - pstr); + strcpy(g_stConfInfo.value1[g_stConfInfo.numOfKey1], ret + 1); + // printf("key: %s, value: %s\n", g_stConfInfo.key[g_stConfInfo.numOfKey], + // g_stConfInfo.value[g_stConfInfo.numOfKey]); + g_stConfInfo.numOfKey1++; } - + token = strtok(NULL, delim); } } - static int running = 1; /*static void msg_process(tmq_message_t* message) { tmqShowMsg(message); }*/ @@ -253,6 +253,7 @@ int queryDB(TAOS* taos, char* command) { } tmq_t* build_consumer() { +#if 0 char sqlStr[1024] = {0}; TAOS* pConn = taos_connect(NULL, "root", "taosdata", NULL, 0); @@ -266,13 +267,19 @@ tmq_t* build_consumer() { exit(-1); } taos_free_result(pRes); +#endif tmq_conf_t* conf = tmq_conf_new(); // tmq_conf_set(conf, "group.id", "tg2"); for (int32_t i = 0; i < g_stConfInfo.numOfKey; i++) { tmq_conf_set(conf, g_stConfInfo.key[i], g_stConfInfo.value[i]); } - tmq_t* tmq = tmq_consumer_new(pConn, conf, NULL, 0); + tmq_conf_set(conf, "td.connect.user", "root"); + tmq_conf_set(conf, "td.connect.pass", "taosdata"); + tmq_conf_set(conf, "td.connect.db", g_stConfInfo.dbName); + tmq_t* tmq = tmq_consumer_new1(conf, NULL, 0); + assert(tmq); + tmq_conf_destroy(conf); return tmq; } @@ -285,10 +292,10 @@ tmq_list_t* build_topic_list() { return topic_list; } - tmq_t* build_consumer_x() { +#if 0 char sqlStr[1024] = {0}; - + TAOS* pConn = taos_connect(NULL, "root", "taosdata", NULL, 0); assert(pConn != NULL); @@ -296,23 +303,29 @@ tmq_t* build_consumer_x() { TAOS_RES* pRes = taos_query(pConn, sqlStr); if (taos_errno(pRes) != 0) { printf("error in use db, reason:%s\n", taos_errstr(pRes)); - taos_free_result(pRes); - exit(-1); + taos_free_result(pRes); + exit(-1); } taos_free_result(pRes); +#endif tmq_conf_t* conf = tmq_conf_new(); - //tmq_conf_set(conf, "group.id", "tg2"); + // tmq_conf_set(conf, "group.id", "tg2"); for (int32_t i = 0; i < g_stConfInfo.numOfKey1; i++) { tmq_conf_set(conf, g_stConfInfo.key1[i], g_stConfInfo.value1[i]); } - tmq_t* tmq = tmq_consumer_new(pConn, conf, NULL, 0); + tmq_conf_set(conf, "td.connect.user", "root"); + tmq_conf_set(conf, "td.connect.pass", "taosdata"); + tmq_conf_set(conf, "td.connect.db", g_stConfInfo.dbName); + tmq_t* tmq = tmq_consumer_new1(conf, NULL, 0); + assert(tmq); + tmq_conf_destroy(conf); return tmq; } tmq_list_t* build_topic_list_x() { tmq_list_t* topic_list = tmq_list_new(); - //tmq_list_append(topic_list, "test_stb_topic_1"); + // tmq_list_append(topic_list, "test_stb_topic_1"); for (int32_t i = 0; i < g_stConfInfo.numOfTopic1; i++) { tmq_list_append(topic_list, g_stConfInfo.topics1[i]); } @@ -367,9 +380,9 @@ int32_t parallel_consume(tmq_t* tmq, int threadLable) { if (tmqMsg) { totalMsgs++; - //printf("threadFlag: %d, totalMsgs: %d\n", threadLable, totalMsgs); + // printf("threadFlag: %d, totalMsgs: %d\n", threadLable, totalMsgs); - #if 0 +#if 0 TAOS_ROW row; while (NULL != (row = tmq_get_row(tmqMsg))) { totalRows++; @@ -396,65 +409,63 @@ int32_t parallel_consume(tmq_t* tmq, int threadLable) { exit(-1); } - //printf("%d", totalMsgs); // output to sim for check result + // printf("%d", totalMsgs); // output to sim for check result return totalMsgs; } - -void *threadFunc(void *param) { +void* threadFunc(void* param) { int32_t totalMsgs = 0; - SThreadInfo *pInfo = (SThreadInfo *)param; + SThreadInfo* pInfo = (SThreadInfo*)param; - tmq_t* tmq = build_consumer_x(); + tmq_t* tmq = build_consumer_x(); tmq_list_t* topic_list = build_topic_list_x(); - if ((NULL == tmq) || (NULL == topic_list)){ + if ((NULL == tmq) || (NULL == topic_list)) { return NULL; } - + tmq_resp_err_t err = tmq_subscribe(tmq, topic_list); if (err) { printf("tmq_subscribe() fail, reason: %s\n", tmq_err2str(err)); exit(-1); } - //if (0 == g_stConfInfo.consumeMsgCnt) { - // loop_consume(tmq); - //} else { - pInfo->consumeMsgCnt = parallel_consume(tmq, 1); + // if (0 == g_stConfInfo.consumeMsgCnt) { + // loop_consume(tmq); + // } else { + pInfo->consumeMsgCnt = parallel_consume(tmq, 1); //} err = tmq_unsubscribe(tmq); if (err) { printf("tmq_unsubscribe() fail, reason: %s\n", tmq_err2str(err)); - pInfo->consumeMsgCnt = -1; + pInfo->consumeMsgCnt = -1; return NULL; } return NULL; } - int main(int32_t argc, char* argv[]) { parseArgument(argc, argv); parseInputString(); - int32_t numOfThreads = 1; + int32_t numOfThreads = 1; TdThreadAttr thattr; taosThreadAttrInit(&thattr); taosThreadAttrSetDetachState(&thattr, PTHREAD_CREATE_JOINABLE); - SThreadInfo *pInfo = (SThreadInfo *)taosMemoryCalloc(numOfThreads, sizeof(SThreadInfo)); + SThreadInfo* pInfo = (SThreadInfo*)taosMemoryCalloc(numOfThreads, sizeof(SThreadInfo)); if (g_stConfInfo.numOfTopic1) { // pthread_create one thread to consume for (int32_t i = 0; i < numOfThreads; ++i) { pInfo[i].expectMsgCnt = 0; pInfo[i].consumeMsgCnt = 0; - taosThreadCreate(&(pInfo[i].thread), &thattr, threadFunc, (void *)(pInfo + i)); + taosThreadCreate(&(pInfo[i].thread), &thattr, threadFunc, (void*)(pInfo + i)); } } - int32_t totalMsgs = 0; + int32_t totalMsgs = 0; tmq_t* tmq = build_consumer(); tmq_list_t* topic_list = build_topic_list(); if ((NULL == tmq) || (NULL == topic_list)) { @@ -479,48 +490,48 @@ int main(int32_t argc, char* argv[]) { exit(-1); } - if (g_stConfInfo.numOfTopic1) { + if (g_stConfInfo.numOfTopic1) { for (int32_t i = 0; i < numOfThreads; i++) { taosThreadJoin(pInfo[i].thread, NULL); } - //printf("consumer: %d, cosumer1: %d\n", totalMsgs, pInfo->consumeMsgCnt); - if (0 == g_stConfInfo.checkMode) { - if ((totalMsgs + pInfo->consumeMsgCnt) == g_stConfInfo.consumeMsgCnt) { - printf("success"); + // printf("consumer: %d, cosumer1: %d\n", totalMsgs, pInfo->consumeMsgCnt); + if (0 == g_stConfInfo.checkMode) { + if ((totalMsgs + pInfo->consumeMsgCnt) == g_stConfInfo.consumeMsgCnt) { + printf("success"); } else { - printf("fail, consumer msg cnt: %d, %d", totalMsgs, pInfo->consumeMsgCnt); - } - } else if (1 == g_stConfInfo.checkMode) { + printf("fail, consumer msg cnt: %d, %d", totalMsgs, pInfo->consumeMsgCnt); + } + } else if (1 == g_stConfInfo.checkMode) { if ((totalMsgs == g_stConfInfo.consumeMsgCnt) && (pInfo->consumeMsgCnt == g_stConfInfo.consumeMsgCnt)) { - printf("success"); + printf("success"); } else { - printf("fail, consumer msg cnt: %d, %d", totalMsgs, pInfo->consumeMsgCnt); - } - } else if (2 == g_stConfInfo.checkMode) { - if ((totalMsgs + pInfo->consumeMsgCnt) == 3 * g_stConfInfo.consumeMsgCnt) { - printf("success"); + printf("fail, consumer msg cnt: %d, %d", totalMsgs, pInfo->consumeMsgCnt); + } + } else if (2 == g_stConfInfo.checkMode) { + if ((totalMsgs + pInfo->consumeMsgCnt) == 3 * g_stConfInfo.consumeMsgCnt) { + printf("success"); } else { - printf("fail, consumer msg cnt: %d, %d", totalMsgs, pInfo->consumeMsgCnt); - } - } else if (3 == g_stConfInfo.checkMode) { + printf("fail, consumer msg cnt: %d, %d", totalMsgs, pInfo->consumeMsgCnt); + } + } else if (3 == g_stConfInfo.checkMode) { if ((totalMsgs == 2 * g_stConfInfo.consumeMsgCnt) && (pInfo->consumeMsgCnt == 2 * g_stConfInfo.consumeMsgCnt)) { - printf("success"); + printf("success"); } else { - printf("fail, consumer msg cnt: %d, %d", totalMsgs, pInfo->consumeMsgCnt); - } - } else if (4 == g_stConfInfo.checkMode) { - if (((totalMsgs == 0) && (pInfo->consumeMsgCnt == 3 * g_stConfInfo.consumeMsgCnt)) - || ((pInfo->consumeMsgCnt == 0) && (totalMsgs == 3 * g_stConfInfo.consumeMsgCnt)) - || ((pInfo->consumeMsgCnt == g_stConfInfo.consumeMsgCnt) && (totalMsgs == 2 * g_stConfInfo.consumeMsgCnt)) - || ((pInfo->consumeMsgCnt == 2 * g_stConfInfo.consumeMsgCnt) && (totalMsgs == g_stConfInfo.consumeMsgCnt))) { - printf("success"); + printf("fail, consumer msg cnt: %d, %d", totalMsgs, pInfo->consumeMsgCnt); + } + } else if (4 == g_stConfInfo.checkMode) { + if (((totalMsgs == 0) && (pInfo->consumeMsgCnt == 3 * g_stConfInfo.consumeMsgCnt)) || + ((pInfo->consumeMsgCnt == 0) && (totalMsgs == 3 * g_stConfInfo.consumeMsgCnt)) || + ((pInfo->consumeMsgCnt == g_stConfInfo.consumeMsgCnt) && (totalMsgs == 2 * g_stConfInfo.consumeMsgCnt)) || + ((pInfo->consumeMsgCnt == 2 * g_stConfInfo.consumeMsgCnt) && (totalMsgs == g_stConfInfo.consumeMsgCnt))) { + printf("success"); } else { - printf("fail, consumer msg cnt: %d, %d", totalMsgs, pInfo->consumeMsgCnt); - } - } else { - printf("fail, check mode unknow. consumer msg cnt: %d, %d", totalMsgs, pInfo->consumeMsgCnt); - } + printf("fail, consumer msg cnt: %d, %d", totalMsgs, pInfo->consumeMsgCnt); + } + } else { + printf("fail, check mode unknow. consumer msg cnt: %d, %d", totalMsgs, pInfo->consumeMsgCnt); + } } return 0;