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

fix(tmq): race condition

This commit is contained in:
Liu Jicong 2022-08-03 19:14:36 +08:00
parent e345023dfd
commit 3ace0b150b
1 changed files with 94 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
source/client/src/tmq.c
View File

@ -509,10 +509,11 @@ static int32_t tmqSendCommitReq(tmq_t* tmq, SMqClientVg* pVg, SMqClientTopic* pT
pMsgSendInfo->msgType = TDMT_VND_MQ_COMMIT_OFFSET;
// send msg
atomic_add_fetch_32(&pParamSet->waitingRspNum, 1);
atomic_add_fetch_32(&pParamSet->totalRspNum, 1);
int64_t transporterId = 0;
asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &pVg->epSet, &transporterId, pMsgSendInfo);
pParamSet->waitingRspNum++;
pParamSet->totalRspNum++;
return 0;
}
@ -2873,7 +2874,8 @@ static int32_t taosDeleteData(TAOS* taos, void* meta, int32_t metaLen) {
// getTbName(req.tableFName);
char sql[256] = {0};
sprintf(sql, "delete from `%s` where `%s` >= %" PRId64" and `%s` <= %" PRId64, req.tableFName, req.tsColName, req.skey, req.tsColName, req.ekey);
sprintf(sql, "delete from `%s` where `%s` >= %" PRId64 " and `%s` <= %" PRId64, req.tableFName, req.tsColName,
req.skey, req.tsColName, req.ekey);
printf("delete sql:%s\n", sql);
TAOS_RES* res = taos_query(taos, sql);