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

enh: refactor batch meta msg

This commit is contained in:
dapan1121 2022-11-17 10:35:25 +08:00
parent fc19382a15
commit 49917d5156
8 changed files with 276 additions and 226 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
include/common/tmsg.h
View File

@ -3168,8 +3168,7 @@ typedef struct {
typedef struct {
SMsgHead header;
int32_t msgNum;
SBatchMsg msg[];
SArray* pMsgs; //SArray<SBatchMsg>
} SBatchReq;
typedef struct {
@ -3178,14 +3177,23 @@ typedef struct {
int32_t msgLen;
int32_t rspCode;
void* msg;
} SBatchRspMsg;
typedef struct {
SArray* pRsps; //SArray<SBatchRspMsg>
} SBatchRsp;
static FORCE_INLINE void tFreeSBatchRsp(void* p) {
int32_t tSerializeSBatchReq(void *buf, int32_t bufLen, SBatchReq *pReq);
int32_t tDeserializeSBatchReq(void *buf, int32_t bufLen, SBatchReq *pReq);
int32_t tSerializeSBatchRsp(void *buf, int32_t bufLen, SBatchRsp *pRsp);
int32_t tDeserializeSBatchRsp(void *buf, int32_t bufLen, SBatchRsp *pRsp);
static FORCE_INLINE void tFreeSBatchRspMsg(void* p) {
if (NULL == p) {
return;
}
SBatchRsp* pRsp = (SBatchRsp*)p;
SBatchRspMsg* pRsp = (SBatchRspMsg*)p;
taosMemoryFree(pRsp->msg);
}

2
include/libs/nodes/querynodes.h
View File

@ -351,7 +351,7 @@ typedef struct SVgDataBlocks {
SVgroupInfo vg;
int32_t numOfTables; // number of tables in current submit block
uint32_t size;
void* pData; // SMsgDesc + SSubmitReq + SSubmitBlk + ...
void* pData; // SSubmitReq + SSubmitBlk + ...
} SVgDataBlocks;
typedef void (*FFreeDataBlockHash)(SHashObj*);

133
source/common/src/tmsg.c
View File

@ -4446,6 +4446,139 @@ void tFreeSExplainRsp(SExplainRsp *pRsp) {
taosMemoryFreeClear(pRsp->subplanInfo);
}
int32_t tSerializeSBatchReq(void *buf, int32_t bufLen, SBatchReq *pReq) {
int32_t headLen = sizeof(SMsgHead);
if (buf != NULL) {
buf = (char *)buf + headLen;
bufLen -= headLen;
}
SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1;
int32_t num = taosArrayGetSize(pReq->pMsgs);
if (tEncodeI32(&encoder, num) < 0) return -1;
for (int32_t i = 0; i < num; ++i) {
SBatchMsg *pMsg = taosArrayGet(pReq->pMsgs, i);
if (tEncodeI32(&encoder, pMsg->msgIdx) < 0) return -1;
if (tEncodeI32(&encoder, pMsg->msgType) < 0) return -1;
if (tEncodeI32(&encoder, pMsg->msgLen) < 0) return -1;
if (tEncodeBinary(&encoder, pMsg->msg, pMsg->msgLen) < 0) return -1;
}
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
tEncoderClear(&encoder);
if (buf != NULL) {
SMsgHead *pHead = (SMsgHead *)((char *)buf - headLen);
pHead->vgId = htonl(pReq->header.vgId);
pHead->contLen = htonl(tlen + headLen);
}
return tlen + headLen;
}
int32_t tDeserializeSBatchReq(void *buf, int32_t bufLen, SBatchReq *pReq) {
int32_t headLen = sizeof(SMsgHead);
SMsgHead *pHead = buf;
pHead->vgId = pReq->header.vgId;
pHead->contLen = pReq->header.contLen;
SDecoder decoder = {0};
tDecoderInit(&decoder, (char *)buf + headLen, bufLen - headLen);
if (tStartDecode(&decoder) < 0) return -1;
int32_t num = 0;
if (tDecodeI32(&decoder, &num) < 0) return -1;
if (num <= 0) {
pReq->pMsgs = NULL;
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
}
pReq->pMsgs = taosArrayInit(num, sizeof(SBatchMsg));
if (NULL == pReq->pMsgs) return -1;
for (int32_t i = 0; i < num; ++i) {
SBatchMsg msg = {0};
if (tDecodeI32(&decoder, &msg.msgIdx) < 0) return -1;
if (tDecodeI32(&decoder, &msg.msgType) < 0) return -1;
if (tDecodeI32(&decoder, &msg.msgLen) < 0) return -1;
if (tDecodeBinary(&decoder, (uint8_t**)&msg.msg, NULL) < 0) return -1;
if (NULL == taosArrayPush(pReq->pMsgs, &msg)) return -1;
}
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
}
int32_t tSerializeSBatchRsp(void *buf, int32_t bufLen, SBatchRsp *pRsp) {
SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1;
int32_t num = taosArrayGetSize(pRsp->pRsps);
if (tEncodeI32(&encoder, num) < 0) return -1;
for (int32_t i = 0; i < num; ++i) {
SBatchRspMsg *pMsg = taosArrayGet(pRsp->pRsps, i);
if (tEncodeI32(&encoder, pMsg->reqType) < 0) return -1;
if (tEncodeI32(&encoder, pMsg->msgIdx) < 0) return -1;
if (tEncodeI32(&encoder, pMsg->msgLen) < 0) return -1;
if (tEncodeI32(&encoder, pMsg->rspCode) < 0) return -1;
if (tEncodeBinary(&encoder, pMsg->msg, pMsg->msgLen) < 0) return -1;
}
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
tEncoderClear(&encoder);
return tlen;
}
int32_t tDeserializeSBatchRsp(void *buf, int32_t bufLen, SBatchRsp *pRsp) {
SDecoder decoder = {0};
tDecoderInit(&decoder, (char *)buf, bufLen);
if (tStartDecode(&decoder) < 0) return -1;
int32_t num = 0;
if (tDecodeI32(&decoder, &num) < 0) return -1;
if (num <= 0) {
pRsp->pRsps = NULL;
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
}
pRsp->pRsps = taosArrayInit(num, sizeof(SBatchRspMsg));
if (NULL == pRsp->pRsps) return -1;
for (int32_t i = 0; i < num; ++i) {
SBatchRspMsg msg = {0};
if (tDecodeI32(&decoder, &msg.reqType) < 0) return -1;
if (tDecodeI32(&decoder, &msg.msgIdx) < 0) return -1;
if (tDecodeI32(&decoder, &msg.msgLen) < 0) return -1;
if (tDecodeI32(&decoder, &msg.rspCode) < 0) return -1;
if (tDecodeBinaryAlloc(&decoder, &msg.msg, NULL) < 0) return -1;
if (NULL == taosArrayPush(pRsp->pRsps, &msg)) return -1;
}
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
}
int32_t tSerializeSSchedulerHbReq(void *buf, int32_t bufLen, SSchedulerHbReq *pReq) {
int32_t headLen = sizeof(SMsgHead);
if (buf != NULL) {

119
source/dnode/mnode/impl/src/mndQuery.c
View File

@ -63,81 +63,60 @@ int32_t mndProcessQueryMsg(SRpcMsg *pMsg) {
return code;
}
static FORCE_INLINE void mnodeFreeSBatchRspMsg(void* p) {
if (NULL == p) {
return;
}
SBatchRspMsg* pRsp = (SBatchRspMsg*)p;
rpcFreeCont(pRsp->msg);
}
int32_t mndProcessBatchMetaMsg(SRpcMsg *pMsg) {
int32_t code = 0;
int32_t offset = 0;
int32_t rspSize = 0;
SBatchReq *batchReq = (SBatchReq *)pMsg->pCont;
int32_t msgNum = ntohl(batchReq->msgNum);
offset += sizeof(SBatchReq);
SBatchReq batchReq = {0};
SBatchMsg req = {0};
SBatchRsp rsp = {0};
SBatchRspMsg rsp = {0};
SBatchRsp batchRsp = {0};
SRpcMsg reqMsg = *pMsg;
SRpcMsg rspMsg = {0};
void *pRsp = NULL;
SMnode *pMnode = pMsg->info.node;
if (tDeserializeSBatchReq(pMsg->pCont, pMsg->contLen, &batchReq)) {
code = TSDB_CODE_OUT_OF_MEMORY;
mError("tDeserializeSBatchReq failed");
goto _exit;
}
int32_t msgNum = taosArrayGetSize(batchReq.pMsgs);
if (msgNum >= MAX_META_MSG_IN_BATCH) {
code = TSDB_CODE_INVALID_MSG;
mError("too many msgs %d in mnode batch meta req", msgNum);
goto _exit;
}
SArray *batchRsp = taosArrayInit(msgNum, sizeof(SBatchRsp));
if (NULL == batchRsp) {
batchRsp.pRsps = taosArrayInit(msgNum, sizeof(SBatchRspMsg));
if (NULL == batchRsp.pRsps) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
for (int32_t i = 0; i < msgNum; ++i) {
if (offset >= pMsg->contLen) {
mError("offset %d is bigger than contLen %d", offset, pMsg->contLen);
terrno = TSDB_CODE_INVALID_MSG_LEN;
taosArrayDestroy(batchRsp);
return -1;
}
SBatchMsg* req = taosArrayGet(batchReq.pMsgs, i);
req.msgIdx = ntohl(*(int32_t *)((char *)pMsg->pCont + offset));
offset += sizeof(req.msgIdx);
if (offset >= pMsg->contLen) {
mError("offset %d is bigger than contLen %d", offset, pMsg->contLen);
terrno = TSDB_CODE_INVALID_MSG_LEN;
taosArrayDestroy(batchRsp);
return -1;
}
req.msgType = ntohl(*(int32_t *)((char *)pMsg->pCont + offset));
offset += sizeof(req.msgType);
if (offset >= pMsg->contLen) {
mError("offset %d is bigger than contLen %d", offset, pMsg->contLen);
terrno = TSDB_CODE_INVALID_MSG_LEN;
taosArrayDestroy(batchRsp);
return -1;
}
req.msgLen = ntohl(*(int32_t *)((char *)pMsg->pCont + offset));
offset += sizeof(req.msgLen);
if (offset >= pMsg->contLen) {
mError("offset %d is bigger than contLen %d", offset, pMsg->contLen);
terrno = TSDB_CODE_INVALID_MSG_LEN;
taosArrayDestroy(batchRsp);
return -1;
}
req.msg = (char *)pMsg->pCont + offset;
offset += req.msgLen;
reqMsg.msgType = req.msgType;
reqMsg.pCont = req.msg;
reqMsg.contLen = req.msgLen;
reqMsg.msgType = req->msgType;
reqMsg.pCont = req->msg;
reqMsg.contLen = req->msgLen;
reqMsg.info.rsp = NULL;
reqMsg.info.rspLen = 0;
MndMsgFp fp = pMnode->msgFp[TMSG_INDEX(req.msgType)];
MndMsgFp fp = pMnode->msgFp[TMSG_INDEX(req->msgType)];
if (fp == NULL) {
mError("msg:%p, failed to get msg handle, app:%p type:%s", pMsg, pMsg->info.ahandle, TMSG_INFO(pMsg->msgType));
terrno = TSDB_CODE_MSG_NOT_PROCESSED;
taosArrayDestroy(batchRsp);
taosArrayDestroy(batchRsp.pRsps);
return -1;
}
@ -146,49 +125,29 @@ int32_t mndProcessBatchMetaMsg(SRpcMsg *pMsg) {
} else {
rsp.rspCode = 0;
}
rsp.msgIdx = req.msgIdx;
rsp.msgIdx = req->msgIdx;
rsp.reqType = reqMsg.msgType;
rsp.msgLen = reqMsg.info.rspLen;
rsp.msg = reqMsg.info.rsp;
taosArrayPush(batchRsp, &rsp);
rspSize += sizeof(rsp) + rsp.msgLen - POINTER_BYTES;
taosArrayPush(batchRsp.pRsps, &rsp);
}
rspSize += sizeof(int32_t);
offset = 0;
rspSize = tSerializeSBatchRsp(NULL, 0, &batchRsp);
if (rspSize < 0) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
pRsp = rpcMallocCont(rspSize);
if (pRsp == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
*(int32_t *)((char *)pRsp + offset) = htonl(msgNum);
offset += sizeof(msgNum);
for (int32_t i = 0; i < msgNum; ++i) {
SBatchRsp *p = taosArrayGet(batchRsp, i);
*(int32_t *)((char *)pRsp + offset) = htonl(p->reqType);
offset += sizeof(p->reqType);
*(int32_t *)((char *)pRsp + offset) = htonl(p->msgIdx);
offset += sizeof(p->msgIdx);
*(int32_t *)((char *)pRsp + offset) = htonl(p->msgLen);
offset += sizeof(p->msgLen);
*(int32_t *)((char *)pRsp + offset) = htonl(p->rspCode);
offset += sizeof(p->rspCode);
if (p->msg != NULL) {
memcpy((char *)pRsp + offset, p->msg, p->msgLen);
offset += p->msgLen;
}
rpcFreeCont(p->msg);
if (tSerializeSBatchRsp(pRsp, rspSize, &batchRsp) < 0) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
taosArrayDestroy(batchRsp);
batchRsp = NULL;
_exit:
pMsg->info.rsp = pRsp;
@ -198,7 +157,7 @@ _exit:
mError("mnd get batch meta failed cause of %s", tstrerror(code));
}
taosArrayDestroyEx(batchRsp, tFreeSBatchRsp);
taosArrayDestroyEx(batchRsp.pRsps, mnodeFreeSBatchRspMsg);
return code;
}

125
source/dnode/vnode/src/vnd/vnodeQuery.c
View File

@ -264,77 +264,54 @@ _exit:
return TSDB_CODE_SUCCESS;
}
static FORCE_INLINE void vnodeFreeSBatchRspMsg(void* p) {
if (NULL == p) {
return;
}
SBatchRspMsg* pRsp = (SBatchRspMsg*)p;
rpcFreeCont(pRsp->msg);
}
int32_t vnodeGetBatchMeta(SVnode *pVnode, SRpcMsg *pMsg) {
int32_t code = 0;
int32_t offset = 0;
int32_t rspSize = 0;
SBatchReq *batchReq = (SBatchReq *)pMsg->pCont;
int32_t msgNum = ntohl(batchReq->msgNum);
offset += sizeof(SBatchReq);
SBatchMsg req = {0};
SBatchRsp rsp = {0};
SBatchReq batchReq = {0};
SBatchMsg *req = NULL;
SBatchRspMsg rsp = {0};
SBatchRsp batchRsp = {0};
SRpcMsg reqMsg = *pMsg;
SRpcMsg rspMsg = {0};
void *pRsp = NULL;
if (tDeserializeSBatchReq(pMsg->pCont, pMsg->contLen, &batchReq)) {
code = TSDB_CODE_OUT_OF_MEMORY;
qError("tDeserializeSBatchReq failed");
goto _exit;
}
int32_t msgNum = taosArrayGetSize(batchReq.pMsgs);
if (msgNum >= MAX_META_MSG_IN_BATCH) {
code = TSDB_CODE_INVALID_MSG;
qError("too many msgs %d in vnode batch meta req", msgNum);
goto _exit;
}
SArray *batchRsp = taosArrayInit(msgNum, sizeof(SBatchRsp));
if (NULL == batchRsp) {
batchRsp.pRsps = taosArrayInit(msgNum, sizeof(SBatchRspMsg));
if (NULL == batchRsp.pRsps) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
for (int32_t i = 0; i < msgNum; ++i) {
if (offset >= pMsg->contLen) {
qError("vnode offset %d is bigger than contLen %d", offset, pMsg->contLen);
terrno = TSDB_CODE_INVALID_MSG_LEN;
taosArrayDestroy(batchRsp);
return -1;
}
req = taosArrayGet(batchReq.pMsgs, i);
req.msgIdx = ntohl(*(int32_t *)((char *)pMsg->pCont + offset));
offset += sizeof(req.msgIdx);
reqMsg.msgType = req->msgType;
reqMsg.pCont = req->msg;
reqMsg.contLen = req->msgLen;
if (offset >= pMsg->contLen) {
qError("vnode offset %d is bigger than contLen %d", offset, pMsg->contLen);
terrno = TSDB_CODE_INVALID_MSG_LEN;
taosArrayDestroy(batchRsp);
return -1;
}
req.msgType = ntohl(*(int32_t *)((char *)pMsg->pCont + offset));
offset += sizeof(req.msgType);
if (offset >= pMsg->contLen) {
qError("vnode offset %d is bigger than contLen %d", offset, pMsg->contLen);
terrno = TSDB_CODE_INVALID_MSG_LEN;
taosArrayDestroy(batchRsp);
return -1;
}
req.msgLen = ntohl(*(int32_t *)((char *)pMsg->pCont + offset));
offset += sizeof(req.msgLen);
if (offset >= pMsg->contLen) {
qError("vnode offset %d is bigger than contLen %d", offset, pMsg->contLen);
terrno = TSDB_CODE_INVALID_MSG_LEN;
taosArrayDestroy(batchRsp);
return -1;
}
req.msg = (char *)pMsg->pCont + offset;
offset += req.msgLen;
reqMsg.msgType = req.msgType;
reqMsg.pCont = req.msg;
reqMsg.contLen = req.msgLen;
switch (req.msgType) {
switch (req->msgType) {
case TDMT_VND_TABLE_META:
vnodeGetTableMeta(pVnode, &reqMsg, false);
break;
@ -342,63 +319,37 @@ int32_t vnodeGetBatchMeta(SVnode *pVnode, SRpcMsg *pMsg) {
vnodeGetTableCfg(pVnode, &reqMsg, false);
break;
default:
qError("invalid req msgType %d", req.msgType);
qError("invalid req msgType %d", req->msgType);
reqMsg.code = TSDB_CODE_INVALID_MSG;
reqMsg.pCont = NULL;
reqMsg.contLen = 0;
break;
}
rsp.msgIdx = req.msgIdx;
rsp.msgIdx = req->msgIdx;
rsp.reqType = reqMsg.msgType;
rsp.msgLen = reqMsg.contLen;
rsp.rspCode = reqMsg.code;
rsp.msg = reqMsg.pCont;
taosArrayPush(batchRsp, &rsp);
rspSize += sizeof(rsp) + rsp.msgLen - POINTER_BYTES;
taosArrayPush(batchRsp.pRsps, &rsp);
}
rspSize += sizeof(int32_t);
offset = 0;
if (rspSize > MAX_META_BATCH_RSP_SIZE) {
qError("rspSize:%d overload", rspSize);
code = TSDB_CODE_INVALID_MSG_LEN;
rspSize = tSerializeSBatchRsp(NULL, 0, &batchRsp);
if (rspSize < 0) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
pRsp = rpcMallocCont(rspSize);
if (pRsp == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
*(int32_t *)((char *)pRsp + offset) = htonl(msgNum);
offset += sizeof(msgNum);
for (int32_t i = 0; i < msgNum; ++i) {
SBatchRsp *p = taosArrayGet(batchRsp, i);
*(int32_t *)((char *)pRsp + offset) = htonl(p->reqType);
offset += sizeof(p->reqType);
*(int32_t *)((char *)pRsp + offset) = htonl(p->msgIdx);
offset += sizeof(p->msgIdx);
*(int32_t *)((char *)pRsp + offset) = htonl(p->msgLen);
offset += sizeof(p->msgLen);
*(int32_t *)((char *)pRsp + offset) = htonl(p->rspCode);
offset += sizeof(p->rspCode);
if (p->msg) {
memcpy((char *)pRsp + offset, p->msg, p->msgLen);
offset += p->msgLen;
}
taosMemoryFreeClear(p->msg);
if (tSerializeSBatchRsp(pRsp, rspSize, &batchRsp)) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
taosArrayDestroy(batchRsp);
batchRsp = NULL;
_exit:
rspMsg.info = pMsg->info;
@ -411,7 +362,7 @@ _exit:
qError("vnd get batch meta failed cause of %s", tstrerror(code));
}
taosArrayDestroyEx(batchRsp, tFreeSBatchRsp);
taosArrayDestroyEx(batchRsp.pRsps, vnodeFreeSBatchRspMsg);
tmsgSendRsp(&rspMsg);

1
source/libs/catalog/inc/catalogInt.h
View File

@ -234,7 +234,6 @@ typedef struct SCatalog {
typedef struct SCtgBatch {
int32_t batchId;
int32_t msgType;
int32_t msgSize;
SArray* pMsgs;
SRequestConnInfo conn;
char dbFName[TSDB_DB_FNAME_LEN];

104
source/libs/catalog/src/ctgRemote.c
View File

@ -26,19 +26,29 @@ int32_t ctgHandleBatchRsp(SCtgJob* pJob, SCtgTaskCallbackParam* cbParam, SDataBu
SCatalog* pCtg = pJob->pCtg;
int32_t taskNum = taosArrayGetSize(cbParam->taskId);
SDataBuf taskMsg = *pMsg;
int32_t offset = 0;
int32_t msgNum = (TSDB_CODE_SUCCESS == rspCode && pMsg->pData && (pMsg->len > 0)) ? ntohl(*(int32_t*)pMsg->pData) : 0;
int32_t msgNum = 0;
SBatchRsp batchRsp = {0};
SBatchRspMsg rsp = {0};
SBatchRspMsg *pRsp = NULL;
if (TSDB_CODE_SUCCESS == rspCode && pMsg->pData && (pMsg->len > 0)) {
if (tDeserializeSBatchRsp(pMsg->pData, pMsg->len, &batchRsp) < 0) {
ctgError("tDeserializeSBatchRsp failed, msgLen:%d", pMsg->len);
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
msgNum = taosArrayGetSize(batchRsp.pRsps);
}
ASSERT(taskNum == msgNum || 0 == msgNum);
ctgDebug("QID:0x%" PRIx64 " ctg got batch %d rsp %s", pJob->queryId, cbParam->batchId,
TMSG_INFO(cbParam->reqType + 1));
offset += sizeof(msgNum);
SBatchRsp rsp = {0};
SHashObj* pBatchs = taosHashInit(taskNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
if (NULL == pBatchs) {
ctgError("taosHashInit %d batch failed", taskNum);
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
}
for (int32_t i = 0; i < taskNum; ++i) {
@ -46,25 +56,18 @@ int32_t ctgHandleBatchRsp(SCtgJob* pJob, SCtgTaskCallbackParam* cbParam, SDataBu
int32_t* msgIdx = taosArrayGet(cbParam->msgIdx, i);
SCtgTask* pTask = taosArrayGet(pJob->pTasks, *taskId);
if (msgNum > 0) {
rsp.reqType = ntohl(*(int32_t*)((char*)pMsg->pData + offset));
offset += sizeof(rsp.reqType);
rsp.msgIdx = ntohl(*(int32_t*)((char*)pMsg->pData + offset));
offset += sizeof(rsp.msgIdx);
rsp.msgLen = ntohl(*(int32_t*)((char*)pMsg->pData + offset));
offset += sizeof(rsp.msgLen);
rsp.rspCode = ntohl(*(int32_t*)((char*)pMsg->pData + offset));
offset += sizeof(rsp.rspCode);
rsp.msg = ((char*)pMsg->pData) + offset;
offset += rsp.msgLen;
pRsp = taosArrayGet(batchRsp.pRsps, i);
taskMsg.msgType = rsp.reqType;
taskMsg.pData = rsp.msg;
taskMsg.len = rsp.msgLen;
taskMsg.msgType = pRsp->reqType;
taskMsg.pData = pRsp->msg;
taskMsg.len = pRsp->msgLen;
ASSERT(rsp.msgIdx == *msgIdx);
ASSERT(pRsp->msgIdx == *msgIdx);
} else {
rsp.msgIdx = *msgIdx;
rsp.reqType = -1;
pRsp = &rsp;
pRsp->msgIdx = *msgIdx;
pRsp->reqType = -1;
pRsp->rspCode = 0;
taskMsg.msgType = -1;
taskMsg.pData = NULL;
taskMsg.len = 0;
@ -72,20 +75,22 @@ int32_t ctgHandleBatchRsp(SCtgJob* pJob, SCtgTaskCallbackParam* cbParam, SDataBu
SCtgTaskReq tReq;
tReq.pTask = pTask;
tReq.msgIdx = rsp.msgIdx;
tReq.msgIdx = pRsp->msgIdx;
SCtgMsgCtx* pMsgCtx = CTG_GET_TASK_MSGCTX(pTask, tReq.msgIdx);
pMsgCtx->pBatchs = pBatchs;
ctgDebug("QID:0x%" PRIx64 " ctg task %d idx %d start to handle rsp %s, pBatchs: %p", pJob->queryId, pTask->taskId,
rsp.msgIdx, TMSG_INFO(taskMsg.msgType + 1), pBatchs);
pRsp->msgIdx, TMSG_INFO(taskMsg.msgType + 1), pBatchs);
(*gCtgAsyncFps[pTask->type].handleRspFp)(&tReq, rsp.reqType, &taskMsg, (rsp.rspCode ? rsp.rspCode : rspCode));
(*gCtgAsyncFps[pTask->type].handleRspFp)(&tReq, pRsp->reqType, &taskMsg, (pRsp->rspCode ? pRsp->rspCode : rspCode));
}
CTG_ERR_JRET(ctgLaunchBatchs(pJob->pCtg, pJob, pBatchs));
_return:
taosArrayDestroyEx(batchRsp.pRsps, tFreeSBatchRspMsg);
ctgFreeBatchs(pBatchs);
CTG_RET(code);
}
@ -481,7 +486,6 @@ int32_t ctgAddBatch(SCatalog* pCtg, int32_t vgId, SRequestConnInfo* pConn, SCtgT
if (NULL == taosArrayPush(newBatch.pMsgIdxs, &req.msgIdx)) {
CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
}
newBatch.msgSize = sizeof(SBatchReq) + sizeof(req) + msgSize - POINTER_BYTES;
if (vgId > 0) {
SName* pName = NULL;
@ -533,8 +537,6 @@ int32_t ctgAddBatch(SCatalog* pCtg, int32_t vgId, SRequestConnInfo* pConn, SCtgT
CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
}
pBatch->msgSize += sizeof(req) + msgSize - POINTER_BYTES;
if (vgId > 0) {
SName* pName = NULL;
if (TDMT_VND_TABLE_CFG == msgType) {
@ -570,38 +572,35 @@ _return:
return code;
}
int32_t ctgBuildBatchReqMsg(SCtgBatch* pBatch, int32_t vgId, void** msg) {
*msg = taosMemoryCalloc(1, pBatch->msgSize);
if (NULL == (*msg)) {
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
int32_t offset = 0;
int32_t ctgBuildBatchReqMsg(SCtgBatch* pBatch, int32_t vgId, void** msg, int32_t *pSize) {
int32_t num = taosArrayGetSize(pBatch->pMsgs);
SBatchReq* pBatchReq = (SBatchReq*)(*msg);
if (num >= CTG_MAX_REQ_IN_BATCH) {
qError("too many msgs %d in one batch request", num);
CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
}
pBatchReq->header.vgId = htonl(vgId);
pBatchReq->msgNum = htonl(num);
offset += sizeof(SBatchReq);
SBatchReq batchReq = {0};
for (int32_t i = 0; i < num; ++i) {
SBatchMsg* pReq = taosArrayGet(pBatch->pMsgs, i);
*(int32_t*)((char*)(*msg) + offset) = htonl(pReq->msgIdx);
offset += sizeof(pReq->msgIdx);
*(int32_t*)((char*)(*msg) + offset) = htonl(pReq->msgType);
offset += sizeof(pReq->msgType);
*(int32_t*)((char*)(*msg) + offset) = htonl(pReq->msgLen);
offset += sizeof(pReq->msgLen);
memcpy((char*)(*msg) + offset, pReq->msg, pReq->msgLen);
offset += pReq->msgLen;
batchReq.header.vgId = vgId;
batchReq.pMsgs = pBatch->pMsgs;
int32_t msgSize = tSerializeSBatchReq(NULL, 0, &batchReq);
if (msgSize < 0) {
qError("tSerializeSBatchReq failed");
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
*msg = taosMemoryCalloc(1, msgSize);
if (NULL == (*msg)) {
qError("calloc batchReq msg failed, size:%d", msgSize);
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
if (tSerializeSBatchReq(*msg, msgSize, &batchReq) < 0) {
qError("tSerializeSBatchReq failed");
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
ASSERT(pBatch->msgSize == offset);
*pSize = msgSize;
qDebug("batch req %d to vg %d msg built with %d meta reqs", pBatch->batchId, vgId, num);
@ -616,12 +615,13 @@ int32_t ctgLaunchBatchs(SCatalog* pCtg, SCtgJob* pJob, SHashObj* pBatchs) {
size_t len = 0;
int32_t* vgId = taosHashGetKey(p, &len);
SCtgBatch* pBatch = (SCtgBatch*)p;
int32_t msgSize = 0;
ctgDebug("QID:0x%" PRIx64 " ctg start to launch batch %d", pJob->queryId, pBatch->batchId);
CTG_ERR_JRET(ctgBuildBatchReqMsg(pBatch, *vgId, &msg));
CTG_ERR_JRET(ctgBuildBatchReqMsg(pBatch, *vgId, &msg, &msgSize));
code = ctgAsyncSendMsg(pCtg, &pBatch->conn, pJob, pBatch->pTaskIds, pBatch->batchId, pBatch->pMsgIdxs,
pBatch->dbFName, *vgId, pBatch->msgType, msg, pBatch->msgSize);
pBatch->dbFName, *vgId, pBatch->msgType, msg, msgSize);
pBatch->pTaskIds = NULL;
CTG_ERR_JRET(code);

2
source/util/src/thash.c
View File

@ -147,7 +147,7 @@ static FORCE_INLINE SHashNode *doSearchInEntryList(SHashObj *pHashObj, SHashEntr
uint32_t hashVal) {
SHashNode *pNode = pe->next;
while (pNode) {
atomic_add_fetch_64(&pHashObj->compTimes, 1);
//atomic_add_fetch_64(&pHashObj->compTimes, 1);
if ((pNode->keyLen == keyLen) && ((*(pHashObj->equalFp))(GET_HASH_NODE_KEY(pNode), key, keyLen) == 0) &&
pNode->removed == 0) {
assert(pNode->hashVal == hashVal);