Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
未验证 提交 84b19fbe 编写于 作者: L Liu Jicong 提交者： GitHub
浏览文件
操作

Merge pull request #11908 from taosdata/feature/tq 

enh(tmq): use sem_timewait to reduce busy loop
上级 3c1f52eb 9723145c
隐藏空白更改
内联 并排
Showing with 42 addition and 110 deletion
include/common/tmsg.h
浏览文件 @ 84b19fbe
......@@ -2383,7 +2383,7 @@ typedef struct {
int32_t epoch;
uint64_t reqId;
int64_t consumerId;
int64_t blockingTime;
int64_t waitTime;
int64_t currentOffset;
} SMqPollReqV2;
......@@ -2400,53 +2400,6 @@ typedef struct {
SSchemaWrapper schema;
} SMqSubTopicEp;
typedef struct {
SMqRspHead head;
int64_t reqOffset;
int64_t rspOffset;
int32_t skipLogNum;
int32_t dataLen;
SArray* blockPos; // beginning pos for each SRetrieveTableRsp
void* blockData; // serialized batched SRetrieveTableRsp
} SMqPollRspV2;
static FORCE_INLINE int32_t tEncodeSMqPollRspV2(void** buf, const SMqPollRspV2* pRsp) {
int32_t tlen = 0;
tlen += taosEncodeFixedI64(buf, pRsp->reqOffset);
tlen += taosEncodeFixedI64(buf, pRsp->rspOffset);
tlen += taosEncodeFixedI32(buf, pRsp->skipLogNum);
tlen += taosEncodeFixedI32(buf, pRsp->dataLen);
if (pRsp->dataLen != 0) {
int32_t sz = taosArrayGetSize(pRsp->blockPos);
tlen += taosEncodeFixedI32(buf, sz);
for (int32_t i = 0; i < sz; i++) {
int32_t blockPos = *(int32_t*)taosArrayGet(pRsp->blockPos, i);
tlen += taosEncodeFixedI32(buf, blockPos);
}
tlen += taosEncodeBinary(buf, pRsp->blockData, pRsp->dataLen);
}
return tlen;
}
static FORCE_INLINE void* tDecodeSMqPollRspV2(const void* buf, SMqPollRspV2* pRsp) {
buf = taosDecodeFixedI64(buf, &pRsp->reqOffset);
buf = taosDecodeFixedI64(buf, &pRsp->rspOffset);
buf = taosDecodeFixedI32(buf, &pRsp->skipLogNum);
buf = taosDecodeFixedI32(buf, &pRsp->dataLen);
if (pRsp->dataLen != 0) {
int32_t sz;
buf = taosDecodeFixedI32(buf, &sz);
pRsp->blockPos = taosArrayInit(sz, sizeof(int32_t));
for (int32_t i = 0; i < sz; i++) {
int32_t blockPos;
buf = taosDecodeFixedI32(buf, &blockPos);
taosArrayPush(pRsp->blockPos, &blockPos);
}
buf = taosDecodeBinary(buf, &pRsp->blockData, pRsp->dataLen);
}
return (void*)buf;
}
typedef struct {
SMqRspHead head;
int64_t reqOffset;
......
include/libs/wal/wal.h
浏览文件 @ 84b19fbe
......@@ -72,8 +72,6 @@ extern "C" {
#define WAL_FILE_LEN (WAL_PATH_LEN + 32)
#define WAL_MAGIC 0xFAFBFCFDULL
#define WAL_CUR_FAILED 1
#pragma pack(push, 1)
typedef enum {
TAOS_WAL_NOLOG = 0,
......
source/client/src/tmq.c
浏览文件 @ 84b19fbe
......@@ -68,7 +68,7 @@ struct tmq_conf_t {
char* user;
char* pass;
char* db;
tmq_commit_cb* commit_cb;
tmq_commit_cb* commitCb;
};
struct tmq_t {
......@@ -115,8 +115,8 @@ enum {
enum {
TMQ_CONSUMER_STATUS__INIT = 0,
TMQ_CONSUMER_STATUS__SUBSCRIBED,
TMQ_CONSUMER_STATUS__READY,
TMQ_CONSUMER_STATUS__NO_TOPIC,
};
enum {
......@@ -300,6 +300,7 @@ void tmqAssignDelayedHbTask(void* param, void* tmrId) {
int8_t* pTaskType = taosAllocateQitem(sizeof(int8_t));
*pTaskType = TMQ_DELAYED_TASK__HB;
taosWriteQitem(tmq->delayedTask, pTaskType);
tsem_post(&tmq->rspSem);
}
void tmqAssignDelayedCommitTask(void* param, void* tmrId) {
......@@ -307,6 +308,7 @@ void tmqAssignDelayedCommitTask(void* param, void* tmrId) {
int8_t* pTaskType = taosAllocateQitem(sizeof(int8_t));
*pTaskType = TMQ_DELAYED_TASK__COMMIT;
taosWriteQitem(tmq->delayedTask, pTaskType);
tsem_post(&tmq->rspSem);
}
void tmqAssignDelayedReportTask(void* param, void* tmrId) {
......@@ -314,6 +316,7 @@ void tmqAssignDelayedReportTask(void* param, void* tmrId) {
int8_t* pTaskType = taosAllocateQitem(sizeof(int8_t));
*pTaskType = TMQ_DELAYED_TASK__REPORT;
taosWriteQitem(tmq->delayedTask, pTaskType);
tsem_post(&tmq->rspSem);
}
int32_t tmqHandleAllDelayedTask(tmq_t* tmq) {
......@@ -364,7 +367,6 @@ int32_t tmqSubscribeCb(void* param, const SDataBuf* pMsg, int32_t code) {
SMqSubscribeCbParam* pParam = (SMqSubscribeCbParam*)param;
pParam->rspErr = code;
tmq_t* tmq = pParam->tmq;
atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__SUBSCRIBED);
tsem_post(&pParam->rspSem);
return 0;
}
......@@ -475,7 +477,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
strcpy(pTmq->groupId, conf->groupId);
pTmq->autoCommit = conf->autoCommit;
pTmq->autoCommitInterval = conf->autoCommitInterval;
pTmq->commit_cb = conf->commit_cb;
pTmq->commit_cb = conf->commitCb;
pTmq->resetOffsetCfg = conf->resetOffset;
// assign consumerId
......@@ -686,7 +688,7 @@ FAIL:
void tmq_conf_set_offset_commit_cb(tmq_conf_t* conf, tmq_commit_cb* cb) {
//
conf->commit_cb = cb;
conf->commitCb = cb;
}
TAOS_RES* tmq_create_stream(TAOS* taos, const char* streamName, const char* tbName, const char* sql) {
......@@ -798,7 +800,7 @@ int32_t tmqPollCb(void* param, const SDataBuf* pMsg, int32_t code) {
// do not write into queue since updating epoch reset
tscWarn("msg discard from vg %d since from earlier epoch, rsp epoch %d, current epoch %d", pParam->vgId, msgEpoch,
tmqEpoch);
/*tsem_post(&tmq->rspSem);*/
tsem_post(&tmq->rspSem);
return 0;
}
......@@ -843,14 +845,14 @@ int32_t tmqPollCb(void* param, const SDataBuf* pMsg, int32_t code) {
pRspWrapper->msg.reqOffset, pRspWrapper->msg.rspOffset);
taosWriteQitem(tmq->mqueue, pRspWrapper);
/*tsem_post(&tmq->rspSem);*/
tsem_post(&tmq->rspSem);
return 0;
CREATE_MSG_FAIL:
if (pParam->epoch == tmq->epoch) {
atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
}
/*tsem_post(&tmq->rspSem);*/
tsem_post(&tmq->rspSem);
return -1;
}
......@@ -927,6 +929,12 @@ bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqAskEpRsp* pRsp) {
if (tmq->clientTopics) taosArrayDestroy(tmq->clientTopics);
taosHashCleanup(pHash);
tmq->clientTopics = newTopics;
if (taosArrayGetSize(tmq->clientTopics) == 0)
atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__NO_TOPIC);
else
atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__READY);
atomic_store_32(&tmq->epoch, epoch);
return set;
}
......@@ -955,9 +963,7 @@ int32_t tmqAskEpCb(void* param, const SDataBuf* pMsg, int32_t code) {
tDecodeSMqAskEpRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &rsp);
/*printf("rsp epoch %ld sz %ld\n", rsp.epoch, rsp.topics->size);*/
/*printf("tmq epoch %ld sz %ld\n", tmq->epoch, tmq->clientTopics->size);*/
if (tmqUpdateEp(tmq, head->epoch, &rsp)) {
atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__READY);
}
tmqUpdateEp(tmq, head->epoch, &rsp);
tDeleteSMqAskEpRsp(&rsp);
} else {
SMqAskEpRspWrapper* pWrapper = taosAllocateQitem(sizeof(SMqAskEpRspWrapper));
......@@ -972,7 +978,7 @@ int32_t tmqAskEpCb(void* param, const SDataBuf* pMsg, int32_t code) {
tDecodeSMqAskEpRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &pWrapper->msg);
taosWriteQitem(tmq->mqueue, pWrapper);
/*tsem_post(&tmq->rspSem);*/
tsem_post(&tmq->rspSem);
taosMemoryFree(pParam);
}
......@@ -1076,7 +1082,7 @@ tmq_resp_err_t tmq_seek(tmq_t* tmq, const tmq_topic_vgroup_t* offset) {
return TMQ_RESP_ERR__FAIL;
}
SMqPollReqV2* tmqBuildConsumeReqImpl(tmq_t* tmq, int64_t blockingTime, SMqClientTopic* pTopic, SMqClientVg* pVg) {
SMqPollReqV2* tmqBuildConsumeReqImpl(tmq_t* tmq, int64_t waitTime, SMqClientTopic* pTopic, SMqClientVg* pVg) {
int64_t reqOffset;
if (pVg->currentOffset >= 0) {
reqOffset = pVg->currentOffset;
......@@ -1101,7 +1107,7 @@ SMqPollReqV2* tmqBuildConsumeReqImpl(tmq_t* tmq, int64_t blockingTime, SMqClient
pReq->subKey[tlen] = TMQ_SEPARATOR;
strcpy(pReq->subKey + tlen + 1, pTopic->topicName);
pReq->blockingTime = blockingTime;
pReq->waitTime = waitTime;
pReq->consumerId = tmq->consumerId;
pReq->epoch = tmq->epoch;
pReq->currentOffset = reqOffset;
......@@ -1130,7 +1136,7 @@ SMqRspObj* tmqBuildRspFromWrapper(SMqPollRspWrapper* pWrapper) {
return pRspObj;
}
int32_t tmqPollImpl(tmq_t* tmq, int64_t blockingTime) {
int32_t tmqPollImpl(tmq_t* tmq, int64_t waitTime) {
/*printf("call poll\n");*/
for (int i = 0; i < taosArrayGetSize(tmq->clientTopics); i++) {
SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, i);
......@@ -1151,17 +1157,17 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t blockingTime) {
#endif
}
atomic_store_32(&pVg->vgSkipCnt, 0);
SMqPollReqV2* pReq = tmqBuildConsumeReqImpl(tmq, blockingTime, pTopic, pVg);
SMqPollReqV2* pReq = tmqBuildConsumeReqImpl(tmq, waitTime, pTopic, pVg);
if (pReq == NULL) {
atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
/*tsem_post(&tmq->rspSem);*/
tsem_post(&tmq->rspSem);
return -1;
}
SMqPollCbParam* pParam = taosMemoryMalloc(sizeof(SMqPollCbParam));
if (pParam == NULL) {
taosMemoryFree(pReq);
atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
/*tsem_post(&tmq->rspSem);*/
tsem_post(&tmq->rspSem);
return -1;
}
pParam->tmq = tmq;
......@@ -1176,7 +1182,7 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t blockingTime) {
taosMemoryFree(pReq);
taosMemoryFree(pParam);
atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
/*tsem_post(&tmq->rspSem);*/
tsem_post(&tmq->rspSem);
return -1;
}
......@@ -1222,7 +1228,7 @@ int32_t tmqHandleNoPollRsp(tmq_t* tmq, SMqRspWrapper* rspWrapper, bool* pReset)
return 0;
}
SMqRspObj* tmqHandleAllRsp(tmq_t* tmq, int64_t blockingTime, bool pollIfReset) {
SMqRspObj* tmqHandleAllRsp(tmq_t* tmq, int64_t waitTime, bool pollIfReset) {
while (1) {
SMqRspWrapper* rspWrapper = NULL;
taosGetQitem(tmq->qall, (void**)&rspWrapper);
......@@ -1261,37 +1267,41 @@ SMqRspObj* tmqHandleAllRsp(tmq_t* tmq, int64_t blockingTime, bool pollIfReset) {
taosFreeQitem(rspWrapper);
if (pollIfReset && reset) {
tscDebug("consumer %ld reset and repoll", tmq->consumerId);
tmqPollImpl(tmq, blockingTime);
tmqPollImpl(tmq, waitTime);
}
}
}
}
TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t blocking_time) {
TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t wait_time) {
SMqRspObj* rspObj;
int64_t startTime = taosGetTimestampMs();
rspObj = tmqHandleAllRsp(tmq, blocking_time, false);
rspObj = tmqHandleAllRsp(tmq, wait_time, false);
if (rspObj) {
return (TAOS_RES*)rspObj;
}
if (atomic_load_8(&tmq->status) != TMQ_CONSUMER_STATUS__READY) {
return NULL;
}
while (1) {
tmqHandleAllDelayedTask(tmq);
tmqPollImpl(tmq, blocking_time);
/*tsem_wait(&tmq->rspSem);*/
tmqPollImpl(tmq, wait_time);
rspObj = tmqHandleAllRsp(tmq, blocking_time, false);
rspObj = tmqHandleAllRsp(tmq, wait_time, false);
if (rspObj) {
return (TAOS_RES*)rspObj;
}
if (blocking_time != 0) {
if (wait_time != 0) {
int64_t endTime = taosGetTimestampMs();
if (endTime - startTime > blocking_time) {
int64_t leftTime = endTime - startTime;
if (leftTime > wait_time) {
tscDebug("consumer %ld (epoch %d) timeout, no rsp", tmq->consumerId, tmq->epoch);
return NULL;
}
tsem_timewait(&tmq->rspSem, leftTime * 1000);
}
}
}
......
source/dnode/vnode/src/tq/tq.c
浏览文件 @ 84b19fbe
......@@ -78,7 +78,7 @@ int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_
SMqDataBlkRsp rsp = {0};
rsp.reqOffset = pExec->pushHandle.reqOffset;
rsp.blockData = taosArrayInit(0, sizeof(int32_t));
rsp.blockData = taosArrayInit(0, sizeof(void*));
rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t));
if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {
......@@ -210,35 +210,6 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
tmsgPutToQueue(&pTq->pVnode->msgCb, FETCH_QUEUE, &req);
#if 0
void* pIter = taosHashIterate(pTq->tqPushMgr->pHash, NULL);
while (pIter != NULL) {
STqPusher* pusher = *(STqPusher**)pIter;
if (pusher->type == TQ_PUSHER_TYPE__STREAM) {
STqStreamPusher* streamPusher = (STqStreamPusher*)pusher;
// repack
STqStreamToken* token = taosMemoryMalloc(sizeof(STqStreamToken));
if (token == NULL) {
taosHashCancelIterate(pTq->tqPushMgr->pHash, pIter);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
token->type = TQ_STREAM_TOKEN__DATA;
token->data = msg;
// set input
// exec
}
// send msg to ep
}
// iterate hash
// process all msg
// if waiting
// memcpy and send msg to fetch thread
// TODO: add reference
// if handle waiting, launch query and response to consumer
//
// if no waiting handle, return
#endif
return 0;
}
......@@ -377,7 +348,7 @@ int32_t tqDeserializeConsumer(STQ* pTq, const STqSerializedHead* pHead, STqConsu
int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
SMqPollReqV2* pReq = pMsg->pCont;
int64_t consumerId = pReq->consumerId;
int64_t waitTime = pReq->blockingTime;
int64_t waitTime = pReq->waitTime;
int32_t reqEpoch = pReq->epoch;
int64_t fetchOffset;
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或