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

fix:[TS-4592] pollFalg error in multi thread

This commit is contained in:
wangmm0220 2024-10-10 11:10:34 +08:00
parent 2ed4aeb0da
commit 1d852d4f89
1 changed files with 5 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
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,7 @@ 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);
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,7 +1057,7 @@ 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);
@ -1640,6 +1640,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 +2442,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);