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

enh: refact query message

This commit is contained in:
dapan1121 2022-11-17 18:04:57 +08:00
parent fb861a3935
commit 1b0c98433e
5 changed files with 326 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
include/common/tmsg.h
View File

@ -1615,15 +1615,21 @@ typedef struct SSubQueryMsg {
uint64_t taskId; uint64_t taskId;
int64_t refId; int64_t refId;
int32_t execId; int32_t execId;
int32_t msgMask;
int8_t taskType; int8_t taskType;
int8_t explain; int8_t explain;
int8_t needFetch; int8_t needFetch;
uint32_t sqlLen; // the query sql, uint32_t sqlLen;
uint32_t phyLen; char *sql;
int32_t msgMask; uint32_t msgLen;
char msg[]; char *msg;
} SSubQueryMsg; } SSubQueryMsg;
int32_t tSerializeSSubQueryMsg(void *buf, int32_t bufLen, SSubQueryMsg *pReq);
int32_t tDeserializeSSubQueryMsg(void *buf, int32_t bufLen, SSubQueryMsg *pReq);
void tFreeSSubQueryMsg(SSubQueryMsg *pReq);
typedef struct { typedef struct {
SMsgHead header; SMsgHead header;
uint64_t sId; uint64_t sId;
@ -1732,6 +1738,10 @@ typedef struct {
int32_t execId; int32_t execId;
} STaskDropReq; } STaskDropReq;
int32_t tSerializeSTaskDropReq(void *buf, int32_t bufLen, STaskDropReq *pReq);
int32_t tDeserializeSTaskDropReq(void *buf, int32_t bufLen, STaskDropReq *pReq);
typedef struct { typedef struct {
int32_t code; int32_t code;
} STaskDropRsp; } STaskDropRsp;

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

@ -4643,6 +4643,141 @@ int32_t tDeserializeSMqHbReq(void *buf, int32_t bufLen, SMqHbReq *pReq) {
return 0; return 0;
} }
int32_t tSerializeSSubQueryMsg(void *buf, int32_t bufLen, SSubQueryMsg *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;
if (tEncodeU64(&encoder, pReq->sId) < 0) return -1;
if (tEncodeU64(&encoder, pReq->queryId) < 0) return -1;
if (tEncodeU64(&encoder, pReq->taskId) < 0) return -1;
if (tEncodeI64(&encoder, pReq->refId) < 0) return -1;
if (tEncodeI32(&encoder, pReq->execId) < 0) return -1;
if (tEncodeI32(&encoder, pReq->msgMask) < 0) return -1;
if (tEncodeI8(&encoder, pReq->taskType) < 0) return -1;
if (tEncodeI8(&encoder, pReq->explain) < 0) return -1;
if (tEncodeI8(&encoder, pReq->needFetch) < 0) return -1;
if (tEncodeU32(&encoder, pReq->sqlLen) < 0) return -1;
if (tEncodeCStrWithLen(&encoder, pReq->sql, pReq->sqlLen) < 0) return -1;
if (tEncodeU32(&encoder, pReq->msgLen) < 0) return -1;
if (tEncodeCStrWithLen(&encoder, pReq->msg, pReq->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 tDeserializeSSubQueryMsg(void *buf, int32_t bufLen, SSubQueryMsg *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;
if (tDecodeU64(&decoder, &pReq->sId) < 0) return -1;
if (tDecodeU64(&decoder, &pReq->queryId) < 0) return -1;
if (tDecodeU64(&decoder, &pReq->taskId) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->refId) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->execId) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->msgMask) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->taskType) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->explain) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->needFetch) < 0) return -1;
if (tDecodeU32(&decoder, &pReq->sqlLen) < 0) return -1;
if (tDecodeCStrAlloc(&decoder, &pReq->sql) < 0) return -1;
if (tDecodeU32(&decoder, &pReq->msgLen) < 0) return -1;
if (tDecodeCStrAlloc(&decoder, &pReq->msg) < 0) return -1;
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
}
void tFreeSSubQueryMsg(SSubQueryMsg *pReq) {
if (NULL == pReq) {
return;
}
taosMemoryFreeClear(pReq->sql);
taosMemoryFreeClear(pReq->msg);
}
int32_t tSerializeSTaskDropReq(void *buf, int32_t bufLen, STaskDropReq *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;
if (tEncodeU64(&encoder, pReq->sId) < 0) return -1;
if (tEncodeU64(&encoder, pReq->queryId) < 0) return -1;
if (tEncodeU64(&encoder, pReq->taskId) < 0) return -1;
if (tEncodeI64(&encoder, pReq->refId) < 0) return -1;
if (tEncodeI32(&encoder, pReq->execId) < 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 tDeserializeSTaskDropReq(void *buf, int32_t bufLen, STaskDropReq *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;
if (tDecodeU64(&decoder, &pReq->sId) < 0) return -1;
if (tDecodeU64(&decoder, &pReq->queryId) < 0) return -1;
if (tDecodeU64(&decoder, &pReq->taskId) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->refId) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->execId) < 0) return -1;
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
}
int32_t tSerializeSSchedulerHbReq(void *buf, int32_t bufLen, SSchedulerHbReq *pReq) { int32_t tSerializeSSchedulerHbReq(void *buf, int32_t bufLen, SSchedulerHbReq *pReq) {
int32_t headLen = sizeof(SMsgHead); int32_t headLen = sizeof(SMsgHead);

193
source/libs/qworker/src/qwMsg.c
View File

@ -182,23 +182,37 @@ int32_t qwBuildAndSendDropRsp(SRpcHandleInfo *pConn, int32_t code) {
#endif #endif
int32_t qwBuildAndSendDropMsg(QW_FPARAMS_DEF, SRpcHandleInfo *pConn) { int32_t qwBuildAndSendDropMsg(QW_FPARAMS_DEF, SRpcHandleInfo *pConn) {
STaskDropReq *req = (STaskDropReq *)rpcMallocCont(sizeof(STaskDropReq)); STaskDropReq qMsg;
if (NULL == req) { qMsg.header.vgId = mgmt->nodeId;
QW_SCH_TASK_ELOG("rpcMallocCont %d failed", (int32_t)sizeof(STaskDropReq)); qMsg.header.contLen = 0;
qMsg.sId = sId;
qMsg.queryId = qId;
qMsg.taskId = tId;
qMsg.refId = rId;
qMsg.execId = eId;
int32_t msgSize = tSerializeSTaskDropReq(NULL, 0, &qMsg);
if (msgSize < 0) {
QW_SCH_TASK_ELOG("tSerializeSTaskDropReq get size, msgSize:%d", msgSize);
QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
void *msg = rpcMallocCont(msgSize);
if (NULL == msg) {
QW_SCH_TASK_ELOG("rpcMallocCont %d failed", msgSize);
QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
if (tSerializeSTaskDropReq(msg, msgSize, &qMsg) < 0) {
QW_SCH_TASK_ELOG("tSerializeSTaskDropReq failed, msgSize:%d", msgSize);
rpcFreeCont(msg);
QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
} }
req->header.vgId = mgmt->nodeId;
req->sId = sId;
req->queryId = qId;
req->taskId = tId;
req->refId = rId;
req->execId = eId;
SRpcMsg pNewMsg = { SRpcMsg pNewMsg = {
.msgType = TDMT_SCH_DROP_TASK, .msgType = TDMT_SCH_DROP_TASK,
.pCont = req, .pCont = msg,
.contLen = sizeof(STaskDropReq), .contLen = msgSize,
.code = 0, .code = 0,
.info = *pConn, .info = *pConn,
}; };
@ -247,22 +261,37 @@ int32_t qwBuildAndSendCQueryMsg(QW_FPARAMS_DEF, SRpcHandleInfo *pConn) {
} }
int32_t qwRegisterQueryBrokenLinkArg(QW_FPARAMS_DEF, SRpcHandleInfo *pConn) { int32_t qwRegisterQueryBrokenLinkArg(QW_FPARAMS_DEF, SRpcHandleInfo *pConn) {
STaskDropReq *req = (STaskDropReq *)rpcMallocCont(sizeof(STaskDropReq)); STaskDropReq qMsg;
if (NULL == req) { qMsg.header.vgId = mgmt->nodeId;
QW_SCH_TASK_ELOG("rpcMallocCont %d failed", (int32_t)sizeof(STaskDropReq)); qMsg.header.contLen = 0;
qMsg.sId = sId;
qMsg.queryId = qId;
qMsg.taskId = tId;
qMsg.refId = rId;
qMsg.execId = eId;
int32_t msgSize = tSerializeSTaskDropReq(NULL, 0, &qMsg);
if (msgSize < 0) {
QW_SCH_TASK_ELOG("tSerializeSTaskDropReq get size, msgSize:%d", msgSize);
QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
void *msg = rpcMallocCont(msgSize);
if (NULL == msg) {
QW_SCH_TASK_ELOG("rpcMallocCont %d failed", msgSize);
QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
if (tSerializeSTaskDropReq(msg, msgSize, &qMsg) < 0) {
QW_SCH_TASK_ELOG("tSerializeSTaskDropReq failed, msgSize:%d", msgSize);
rpcFreeCont(msg);
QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
} }
req->header.vgId = htonl(mgmt->nodeId);
req->sId = htobe64(sId);
req->queryId = htobe64(qId);
req->taskId = htobe64(tId);
req->refId = htobe64(rId);
SRpcMsg brokenMsg = { SRpcMsg brokenMsg = {
.msgType = TDMT_SCH_DROP_TASK, .msgType = TDMT_SCH_DROP_TASK,
.pCont = req, .pCont = msg,
.contLen = sizeof(STaskDropReq), .contLen = msgSize,
.code = TSDB_CODE_RPC_BROKEN_LINK, .code = TSDB_CODE_RPC_BROKEN_LINK,
.info = *pConn, .info = *pConn,
}; };
@ -312,40 +341,31 @@ int32_t qWorkerPreprocessQueryMsg(void *qWorkerMgmt, SRpcMsg *pMsg, bool chkGran
} }
int32_t code = 0; int32_t code = 0;
SSubQueryMsg *msg = pMsg->pCont;
SQWorker *mgmt = (SQWorker *)qWorkerMgmt; SQWorker *mgmt = (SQWorker *)qWorkerMgmt;
SSubQueryMsg msg = {0};
if (NULL == msg || pMsg->contLen <= sizeof(*msg)) { if (tDeserializeSSubQueryMsg(pMsg->pCont, pMsg->contLen, &msg) < 0) {
QW_ELOG("invalid query msg, msg:%p, msgLen:%d", msg, pMsg->contLen); QW_ELOG("tDeserializeSSubQueryMsg failed, contLen:%d", pMsg->contLen);
QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT); QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
} }
msg->sId = be64toh(msg->sId); if (chkGrant && (!TEST_SHOW_REWRITE_MASK(msg.msgMask)) && (grantCheck(TSDB_GRANT_TIME) != TSDB_CODE_SUCCESS)) {
msg->queryId = be64toh(msg->queryId); QW_ELOG("query failed cause of grant expired, msgMask:%d", msg.msgMask);
msg->taskId = be64toh(msg->taskId); tFreeSSubQueryMsg(&msg);
msg->refId = be64toh(msg->refId);
msg->execId = ntohl(msg->execId);
msg->phyLen = ntohl(msg->phyLen);
msg->sqlLen = ntohl(msg->sqlLen);
msg->msgMask = ntohl(msg->msgMask);
if (chkGrant && (!TEST_SHOW_REWRITE_MASK(msg->msgMask)) && (grantCheck(TSDB_GRANT_TIME) != TSDB_CODE_SUCCESS)) {
QW_ELOG("query failed cause of grant expired, msgMask:%d", msg->msgMask);
QW_ERR_RET(TSDB_CODE_GRANT_EXPIRED); QW_ERR_RET(TSDB_CODE_GRANT_EXPIRED);
} }
uint64_t sId = msg->sId; uint64_t sId = msg.sId;
uint64_t qId = msg->queryId; uint64_t qId = msg.queryId;
uint64_t tId = msg->taskId; uint64_t tId = msg.taskId;
int64_t rId = msg->refId; int64_t rId = msg.refId;
int32_t eId = msg->execId; int32_t eId = msg.execId;
SQWMsg qwMsg = { SQWMsg qwMsg = {
.msgType = pMsg->msgType, .msg = msg->msg + msg->sqlLen, .msgLen = msg->phyLen, .connInfo = pMsg->info}; .msgType = pMsg->msgType, .msg = msg.msg, .msgLen = msg.msgLen, .connInfo = pMsg->info};
QW_SCH_TASK_DLOG("prerocessQuery start, handle:%p", pMsg->info.handle); QW_SCH_TASK_DLOG("prerocessQuery start, handle:%p, SQL:%s", pMsg->info.handle, msg.sql);
QW_ERR_RET(qwPreprocessQuery(QW_FPARAMS(), &qwMsg)); code = qwPreprocessQuery(QW_FPARAMS(), &qwMsg);
QW_SCH_TASK_DLOG("prerocessQuery end, handle:%p", pMsg->info.handle); QW_SCH_TASK_DLOG("prerocessQuery end, handle:%p, code:%x", pMsg->info.handle, code);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -355,19 +375,25 @@ int32_t qWorkerAbortPreprocessQueryMsg(void *qWorkerMgmt, SRpcMsg *pMsg) {
QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT); QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
} }
SSubQueryMsg *msg = pMsg->pCont;
SQWorker *mgmt = (SQWorker *)qWorkerMgmt; SQWorker *mgmt = (SQWorker *)qWorkerMgmt;
SSubQueryMsg msg = {0};
if (tDeserializeSSubQueryMsg(pMsg->pCont, pMsg->contLen, &msg) < 0) {
QW_ELOG("tDeserializeSSubQueryMsg failed, contLen:%d", pMsg->contLen);
QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
}
uint64_t sId = msg->sId; uint64_t sId = msg.sId;
uint64_t qId = msg->queryId; uint64_t qId = msg.queryId;
uint64_t tId = msg->taskId; uint64_t tId = msg.taskId;
int64_t rId = msg->refId; int64_t rId = msg.refId;
int32_t eId = msg->execId; int32_t eId = msg.execId;
QW_SCH_TASK_DLOG("Abort prerocessQuery start, handle:%p", pMsg->info.handle); QW_SCH_TASK_DLOG("Abort prerocessQuery start, handle:%p", pMsg->info.handle);
qwAbortPrerocessQuery(QW_FPARAMS()); qwAbortPrerocessQuery(QW_FPARAMS());
QW_SCH_TASK_DLOG("Abort prerocessQuery end, handle:%p", pMsg->info.handle); QW_SCH_TASK_DLOG("Abort prerocessQuery end, handle:%p", pMsg->info.handle);
tFreeSSubQueryMsg(&msg);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -377,42 +403,41 @@ int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, int
} }
int32_t code = 0; int32_t code = 0;
SSubQueryMsg *msg = pMsg->pCont;
SQWorker *mgmt = (SQWorker *)qWorkerMgmt; SQWorker *mgmt = (SQWorker *)qWorkerMgmt;
qwUpdateTimeInQueue(mgmt, ts, QUERY_QUEUE); qwUpdateTimeInQueue(mgmt, ts, QUERY_QUEUE);
QW_STAT_INC(mgmt->stat.msgStat.queryProcessed, 1); QW_STAT_INC(mgmt->stat.msgStat.queryProcessed, 1);
if (NULL == msg || pMsg->contLen <= sizeof(*msg)) { SSubQueryMsg msg = {0};
QW_ELOG("invalid query msg, msg:%p, msgLen:%d", msg, pMsg->contLen); if (tDeserializeSSubQueryMsg(pMsg->pCont, pMsg->contLen, &msg) < 0) {
QW_ELOG("tDeserializeSSubQueryMsg failed, contLen:%d", pMsg->contLen);
QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT); QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
} }
uint64_t sId = msg->sId; uint64_t sId = msg.sId;
uint64_t qId = msg->queryId; uint64_t qId = msg.queryId;
uint64_t tId = msg->taskId; uint64_t tId = msg.taskId;
int64_t rId = msg->refId; int64_t rId = msg.refId;
int32_t eId = msg->execId; int32_t eId = msg.execId;
SQWMsg qwMsg = {.node = node, SQWMsg qwMsg = {.node = node,
.msg = msg->msg + msg->sqlLen, .msg = msg.msg,
.msgLen = msg->phyLen, .msgLen = msg.msgLen,
.connInfo = pMsg->info, .connInfo = pMsg->info,
.msgType = pMsg->msgType}; .msgType = pMsg->msgType};
qwMsg.msgInfo.explain = msg->explain; qwMsg.msgInfo.explain = msg.explain;
qwMsg.msgInfo.taskType = msg->taskType; qwMsg.msgInfo.taskType = msg.taskType;
qwMsg.msgInfo.needFetch = msg->needFetch; qwMsg.msgInfo.needFetch = msg.needFetch;
char *sql = strndup(msg->msg, msg->sqlLen);
QW_SCH_TASK_DLOG("processQuery start, node:%p, type:%s, handle:%p, SQL:%s", node, TMSG_INFO(pMsg->msgType), QW_SCH_TASK_DLOG("processQuery start, node:%p, type:%s, handle:%p, SQL:%s", node, TMSG_INFO(pMsg->msgType),
pMsg->info.handle, sql); pMsg->info.handle, msg.sql);
QW_ERR_JRET(qwProcessQuery(QW_FPARAMS(), &qwMsg, sql)); code = qwProcessQuery(QW_FPARAMS(), &qwMsg, msg.sql);
msg.sql = NULL;
QW_SCH_TASK_DLOG("processQuery end, node:%p, code:%x", node, code);
_return: tFreeSSubQueryMsg(&msg);
QW_SCH_TASK_DLOG("processQuery end, node:%p, code:%d", node, code); return TSDB_CODE_SUCCESS;
return code;
} }
int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, int64_t ts) { int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, int64_t ts) {
@ -548,28 +573,22 @@ int32_t qWorkerProcessDropMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, int6
} }
int32_t code = 0; int32_t code = 0;
STaskDropReq *msg = pMsg->pCont;
SQWorker *mgmt = (SQWorker *)qWorkerMgmt; SQWorker *mgmt = (SQWorker *)qWorkerMgmt;
qwUpdateTimeInQueue(mgmt, ts, FETCH_QUEUE); qwUpdateTimeInQueue(mgmt, ts, FETCH_QUEUE);
QW_STAT_INC(mgmt->stat.msgStat.dropProcessed, 1); QW_STAT_INC(mgmt->stat.msgStat.dropProcessed, 1);
if (NULL == msg || pMsg->contLen < sizeof(*msg)) { STaskDropReq msg = {0};
QW_ELOG("invalid task drop msg, msg:%p, msgLen:%d", msg, pMsg->contLen); if (tDeserializeSTaskDropReq(pMsg->pCont, pMsg->contLen, &msg) < 0) {
QW_ELOG("tDeserializeSTaskDropReq failed, contLen:%d", pMsg->contLen);
QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT); QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
} }
msg->sId = be64toh(msg->sId); uint64_t sId = msg.sId;
msg->queryId = be64toh(msg->queryId); uint64_t qId = msg.queryId;
msg->taskId = be64toh(msg->taskId); uint64_t tId = msg.taskId;
msg->refId = be64toh(msg->refId); int64_t rId = msg.refId;
msg->execId = ntohl(msg->execId); int32_t eId = msg.execId;
uint64_t sId = msg->sId;
uint64_t qId = msg->queryId;
uint64_t tId = msg->taskId;
int64_t rId = msg->refId;
int32_t eId = msg->execId;
SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0, .code = pMsg->code, .connInfo = pMsg->info}; SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0, .code = pMsg->code, .connInfo = pMsg->info};

29
source/libs/qworker/test/qworkerTests.cpp
View File

@ -114,7 +114,7 @@ void qwtBuildQueryReqMsg(SRpcMsg *queryRpc) {
qwtqueryMsg.queryId = htobe64(atomic_add_fetch_64(&qwtTestQueryId, 1)); qwtqueryMsg.queryId = htobe64(atomic_add_fetch_64(&qwtTestQueryId, 1));
qwtqueryMsg.sId = htobe64(1); qwtqueryMsg.sId = htobe64(1);
qwtqueryMsg.taskId = htobe64(1); qwtqueryMsg.taskId = htobe64(1);
qwtqueryMsg.phyLen = htonl(100); qwtqueryMsg.msgLen = htonl(100);
qwtqueryMsg.sqlLen = 0; qwtqueryMsg.sqlLen = 0;
queryRpc->msgType = TDMT_SCH_QUERY; queryRpc->msgType = TDMT_SCH_QUERY;
queryRpc->pCont = &qwtqueryMsg; queryRpc->pCont = &qwtqueryMsg;
@ -131,12 +131,29 @@ void qwtBuildFetchReqMsg(SResFetchReq *fetchMsg, SRpcMsg *fetchRpc) {
} }
void qwtBuildDropReqMsg(STaskDropReq *dropMsg, SRpcMsg *dropRpc) { void qwtBuildDropReqMsg(STaskDropReq *dropMsg, SRpcMsg *dropRpc) {
dropMsg->sId = htobe64(1); dropMsg->sId = 1;
dropMsg->queryId = htobe64(atomic_load_64(&qwtTestQueryId)); dropMsg->queryId = atomic_load_64(&qwtTestQueryId);
dropMsg->taskId = htobe64(1); dropMsg->taskId = 1;
int32_t msgSize = tSerializeSTaskDropReq(NULL, 0, dropMsg);
if (msgSize < 0) {
return;
}
char *msg = taosMemoryCalloc(1, msgSize);
if (NULL == msg) {
return;
}
if (tSerializeSTaskDropReq(msg, msgSize, dropMsg) < 0) {
taosMemoryFree(msg);
return;
}
dropRpc->msgType = TDMT_SCH_DROP_TASK; dropRpc->msgType = TDMT_SCH_DROP_TASK;
dropRpc->pCont = dropMsg; dropRpc->pCont = msg;
dropRpc->contLen = sizeof(STaskDropReq); dropRpc->contLen = msgSize;
} }
int32_t qwtStringToPlan(const char *str, SSubplan **subplan) { int32_t qwtStringToPlan(const char *str, SSubplan **subplan) {

76
source/libs/scheduler/src/schRemote.c
View File

@ -1042,30 +1042,40 @@ int32_t schBuildAndSendMsg(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr,
case TDMT_SCH_MERGE_QUERY: { case TDMT_SCH_MERGE_QUERY: {
SCH_ERR_RET(schMakeQueryRpcCtx(pJob, pTask, &rpcCtx)); SCH_ERR_RET(schMakeQueryRpcCtx(pJob, pTask, &rpcCtx));
uint32_t len = strlen(pJob->sql); SSubQueryMsg qMsg;
msgSize = sizeof(SSubQueryMsg) + pTask->msgLen + len; qMsg.header.vgId = addr->nodeId;
qMsg.header.contLen = 0;
qMsg.sId = schMgmt.sId;
qMsg.queryId = pJob->queryId;
qMsg.taskId = pTask->taskId;
qMsg.refId = pJob->refId;
qMsg.execId = pTask->execId;
qMsg.msgMask = (pTask->plan->showRewrite) ? QUERY_MSG_MASK_SHOW_REWRITE() : 0;
qMsg.taskType = TASK_TYPE_TEMP;
qMsg.explain = SCH_IS_EXPLAIN_JOB(pJob);
qMsg.needFetch = SCH_TASK_NEED_FETCH(pTask);
qMsg.sqlLen = strlen(pJob->sql);
qMsg.sql = pJob->sql;
qMsg.msgLen = pTask->msgLen;
qMsg.msg = pTask->msg;
msgSize = tSerializeSSubQueryMsg(NULL, 0, &qMsg);
if (msgSize < 0) {
SCH_TASK_ELOG("tSerializeSSubQueryMsg get size, msgSize:%d", msgSize);
SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
msg = taosMemoryCalloc(1, msgSize); msg = taosMemoryCalloc(1, msgSize);
if (NULL == msg) { if (NULL == msg) {
SCH_TASK_ELOG("calloc %d failed", msgSize); SCH_TASK_ELOG("calloc %d failed", msgSize);
SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
} }
SSubQueryMsg *pMsg = msg; if (tSerializeSSubQueryMsg(msg, msgSize, &qMsg) < 0) {
pMsg->header.vgId = htonl(addr->nodeId); SCH_TASK_ELOG("tSerializeSSubQueryMsg failed, msgSize:%d", msgSize);
pMsg->sId = htobe64(schMgmt.sId); taosMemoryFree(msg);
pMsg->queryId = htobe64(pJob->queryId); SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
pMsg->taskId = htobe64(pTask->taskId); }
pMsg->refId = htobe64(pJob->refId);
pMsg->execId = htonl(pTask->execId);
pMsg->taskType = TASK_TYPE_TEMP;
pMsg->explain = SCH_IS_EXPLAIN_JOB(pJob);
pMsg->needFetch = SCH_TASK_NEED_FETCH(pTask);
pMsg->phyLen = htonl(pTask->msgLen);
pMsg->sqlLen = htonl(len);
pMsg->msgMask = htonl((pTask->plan->showRewrite) ? QUERY_MSG_MASK_SHOW_REWRITE() : 0);
memcpy(pMsg->msg, pJob->sql, len);
memcpy(pMsg->msg + len, pTask->msg, pTask->msgLen);
persistHandle = true; persistHandle = true;
SCH_SET_TASK_HANDLE(pTask, rpcAllocHandle()); SCH_SET_TASK_HANDLE(pTask, rpcAllocHandle());
@ -1092,22 +1102,32 @@ int32_t schBuildAndSendMsg(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr,
break; break;
} }
case TDMT_SCH_DROP_TASK: { case TDMT_SCH_DROP_TASK: {
msgSize = sizeof(STaskDropReq); STaskDropReq qMsg;
qMsg.header.vgId = addr->nodeId;
qMsg.header.contLen = 0;
qMsg.sId = schMgmt.sId;
qMsg.queryId = pJob->queryId;
qMsg.taskId = pTask->taskId;
qMsg.refId = pJob->refId;
qMsg.execId = pTask->execId;
msgSize = tSerializeSTaskDropReq(NULL, 0, &qMsg);
if (msgSize < 0) {
SCH_TASK_ELOG("tSerializeSTaskDropReq get size, msgSize:%d", msgSize);
SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
msg = taosMemoryCalloc(1, msgSize); msg = taosMemoryCalloc(1, msgSize);
if (NULL == msg) { if (NULL == msg) {
SCH_TASK_ELOG("calloc %d failed", msgSize); SCH_TASK_ELOG("calloc %d failed", msgSize);
SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
} }
STaskDropReq *pMsg = msg; if (tSerializeSTaskDropReq(msg, msgSize, &qMsg) < 0) {
SCH_TASK_ELOG("tSerializeSTaskDropReq failed, msgSize:%d", msgSize);
pMsg->header.vgId = htonl(addr->nodeId); taosMemoryFree(msg);
SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
pMsg->sId = htobe64(schMgmt.sId); }
pMsg->queryId = htobe64(pJob->queryId);
pMsg->taskId = htobe64(pTask->taskId);
pMsg->refId = htobe64(pJob->refId);
pMsg->execId = htonl(pTask->execId);
break; break;
} }
case TDMT_SCH_QUERY_HEARTBEAT: { case TDMT_SCH_QUERY_HEARTBEAT: {