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

enh:[TS-5441] change poll flag to consumer for multi consumers

This commit is contained in:
wangmm0220 2024-10-09 15:30:28 +08:00
parent 67fe609129
commit 0bcaf6f4e5
3 changed files with 11 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
source/client/src/clientTmq.c
View File

@ -134,6 +134,7 @@ struct tmq_t {
// poll info // poll info
int64_t pollCnt; int64_t pollCnt;
int64_t totalRows; int64_t totalRows;
int8_t pollFlag;
// timer // timer
tmr_h hbLiveTimer; tmr_h hbLiveTimer;
@ -287,7 +288,6 @@ typedef struct {
static TdThreadOnce tmqInit = PTHREAD_ONCE_INIT; // initialize only once static TdThreadOnce tmqInit = PTHREAD_ONCE_INIT; // initialize only once
volatile int32_t tmqInitRes = 0; // initialize rsp code volatile int32_t tmqInitRes = 0; // initialize rsp code
static SMqMgmt tmqMgmt = {0}; static SMqMgmt tmqMgmt = {0};
static int8_t pollFlag = 0;
tmq_conf_t* tmq_conf_new() { tmq_conf_t* tmq_conf_new() {
tmq_conf_t* conf = taosMemoryCalloc(1, sizeof(tmq_conf_t)); tmq_conf_t* conf = taosMemoryCalloc(1, sizeof(tmq_conf_t));
@ -977,7 +977,8 @@ void tmqSendHbReq(void* param, void* tmrId) {
SMqHbReq req = {0}; SMqHbReq req = {0};
req.consumerId = tmq->consumerId; req.consumerId = tmq->consumerId;
req.epoch = tmq->epoch; req.epoch = tmq->epoch;
req.pollFlag = atomic_load_8(&pollFlag); req.pollFlag = atomic_load_8(&tmq->pollFlag);
tqDebugC("consumer:0x%" PRIx64 " send hb, pollFlag:%d", tmq->consumerId, req.pollFlag);
req.topics = taosArrayInit(taosArrayGetSize(tmq->clientTopics), sizeof(TopicOffsetRows)); req.topics = taosArrayInit(taosArrayGetSize(tmq->clientTopics), sizeof(TopicOffsetRows));
if (req.topics == NULL) { if (req.topics == NULL) {
goto END; goto END;
@ -1057,13 +1058,13 @@ void tmqSendHbReq(void* param, void* tmrId) {
if (code != 0) { if (code != 0) {
tqErrorC("tmqSendHbReq asyncSendMsgToServer failed"); tqErrorC("tmqSendHbReq asyncSendMsgToServer failed");
} }
(void)atomic_val_compare_exchange_8(&pollFlag, 1, 0); (void)atomic_val_compare_exchange_8(&tmq->pollFlag, 1, 0);
END: END:
tDestroySMqHbReq(&req); tDestroySMqHbReq(&req);
if (tmrId != NULL) { if (tmrId != NULL) {
bool ret = taosTmrReset(tmqSendHbReq, tmq->heartBeatIntervalMs, param, tmqMgmt.timer, &tmq->hbLiveTimer); bool ret = taosTmrReset(tmqSendHbReq, tmq->heartBeatIntervalMs, param, tmqMgmt.timer, &tmq->hbLiveTimer);
tqDebugC("reset timer fo tmq hb:%d", ret); tqDebugC("consumer:0x%" PRIx64 " reset timer for tmq hb:%d, pollFlag:%d", tmq->consumerId, ret, tmq->pollFlag);
} }
int32_t ret = taosReleaseRef(tmqMgmt.rsetId, refId); int32_t ret = taosReleaseRef(tmqMgmt.rsetId, refId);
if (ret != 0){ if (ret != 0){
@ -1422,7 +1423,7 @@ void tmqHandleAllDelayedTask(tmq_t* pTmq) {
tqDebugC("consumer:0x%" PRIx64 " retrieve ep from mnode in 1s", pTmq->consumerId); tqDebugC("consumer:0x%" PRIx64 " retrieve ep from mnode in 1s", pTmq->consumerId);
bool ret = taosTmrReset(tmqAssignAskEpTask, DEFAULT_ASKEP_INTERVAL, (void*)(pTmq->refId), tmqMgmt.timer, bool ret = taosTmrReset(tmqAssignAskEpTask, DEFAULT_ASKEP_INTERVAL, (void*)(pTmq->refId), tmqMgmt.timer,
&pTmq->epTimer); &pTmq->epTimer);
tqDebugC("reset timer fo tmq ask ep:%d", ret); tqDebugC("reset timer for tmq ask ep:%d", ret);
} else if (*pTaskType == TMQ_DELAYED_TASK__COMMIT) { } else if (*pTaskType == TMQ_DELAYED_TASK__COMMIT) {
tmq_commit_cb* pCallbackFn = (pTmq->commitCb != NULL) ? pTmq->commitCb : defaultCommitCbFn; tmq_commit_cb* pCallbackFn = (pTmq->commitCb != NULL) ? pTmq->commitCb : defaultCommitCbFn;
asyncCommitAllOffsets(pTmq, pCallbackFn, pTmq->commitCbUserParam); asyncCommitAllOffsets(pTmq, pCallbackFn, pTmq->commitCbUserParam);
@ -1430,7 +1431,7 @@ void tmqHandleAllDelayedTask(tmq_t* pTmq) {
pTmq->autoCommitInterval / 1000.0); pTmq->autoCommitInterval / 1000.0);
bool ret = taosTmrReset(tmqAssignDelayedCommitTask, pTmq->autoCommitInterval, (void*)(pTmq->refId), tmqMgmt.timer, bool ret = taosTmrReset(tmqAssignDelayedCommitTask, pTmq->autoCommitInterval, (void*)(pTmq->refId), tmqMgmt.timer,
&pTmq->commitTimer); &pTmq->commitTimer);
tqDebugC("reset timer fo commit:%d", ret); tqDebugC("reset timer for commit:%d", ret);
} else { } else {
tqErrorC("consumer:0x%" PRIx64 " invalid task type:%d", pTmq->consumerId, *pTaskType); tqErrorC("consumer:0x%" PRIx64 " invalid task type:%d", pTmq->consumerId, *pTaskType);
} }
@ -1640,6 +1641,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
pTmq->status = TMQ_CONSUMER_STATUS__INIT; pTmq->status = TMQ_CONSUMER_STATUS__INIT;
pTmq->pollCnt = 0; pTmq->pollCnt = 0;
pTmq->epoch = 0; pTmq->epoch = 0;
pTmq->pollFlag = 0;
// set conf // set conf
tstrncpy(pTmq->clientId, conf->clientId, TSDB_CLIENT_ID_LEN); tstrncpy(pTmq->clientId, conf->clientId, TSDB_CLIENT_ID_LEN);
@ -2441,7 +2443,7 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t timeout) {
return NULL; return NULL;
} }
(void)atomic_val_compare_exchange_8(&pollFlag, 0, 1); (void)atomic_val_compare_exchange_8(&tmq->pollFlag, 0, 1);
while (1) { while (1) {
tmqHandleAllDelayedTask(tmq); tmqHandleAllDelayedTask(tmq);

1
source/dnode/mnode/impl/src/mndConsumer.c
View File

@ -239,6 +239,7 @@ static int32_t mndProcessMqHbReq(SRpcMsg *pMsg) {
MND_TMQ_RETURN_CHECK(mndAcquireConsumer(pMnode, consumerId, &pConsumer)); MND_TMQ_RETURN_CHECK(mndAcquireConsumer(pMnode, consumerId, &pConsumer));
MND_TMQ_RETURN_CHECK(checkPrivilege(pMnode, pConsumer, &rsp, pMsg->info.conn.user)); MND_TMQ_RETURN_CHECK(checkPrivilege(pMnode, pConsumer, &rsp, pMsg->info.conn.user));
atomic_store_32(&pConsumer->hbStatus, 0); atomic_store_32(&pConsumer->hbStatus, 0);
mDebug("consumer:0x%" PRIx64 " receive hb pollFlag:%d %d", consumerId, req.pollFlag, pConsumer->pollStatus);
if (req.pollFlag == 1){ if (req.pollFlag == 1){
atomic_store_32(&pConsumer->pollStatus, 0); atomic_store_32(&pConsumer->pollStatus, 0);
} }

2
source/dnode/vnode/src/tq/tqRead.c
View File

@ -953,7 +953,7 @@ END:
} }
int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas, SSubmitTbData** pSubmitTbDataRet) { int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas, SSubmitTbData** pSubmitTbDataRet) {
tqDebug("tq reader retrieve data block %p, %d", pReader->msg.msgStr, pReader->nextBlk); tqTrace("tq reader retrieve data block %p, %d", pReader->msg.msgStr, pReader->nextBlk);
SSDataBlock* block = NULL; SSDataBlock* block = NULL;
SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk); SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk);