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

Merge pull request #19277 from taosdata/fix/qwInvalidMsgOrder 

fix: invalid msg order issue
This commit is contained in:
dapan1121 2022-12-30 15:35:58 +08:00 committed by GitHub
parent bb9785746b 5dc80c0bb1
commit 2a15e52bb9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 4 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
source/libs/qworker/src/qworker.c
View File

@ -681,6 +681,7 @@ int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
void *rsp = NULL;
int32_t dataLen = 0;
bool queryStop = false;
bool qComplete = false;
do {
ctx = NULL;
@ -705,11 +706,12 @@ int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
}
if (rsp) {
bool qComplete = (DS_BUF_EMPTY == sOutput.bufStatus && sOutput.queryEnd);
qComplete = (DS_BUF_EMPTY == sOutput.bufStatus && sOutput.queryEnd);
qwBuildFetchRsp(rsp, &sOutput, dataLen, qComplete);
if (qComplete) {
atomic_store_8((int8_t *)&ctx->queryEnd, true);
atomic_store_8((int8_t *)&ctx->queryContinue, 0);
}
qwMsg->connInfo = ctx->dataConnInfo;
@ -743,8 +745,7 @@ int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
}
QW_LOCK(QW_WRITE, &ctx->lock);
if ((queryStop && (0 == atomic_load_8((int8_t *)&ctx->queryContinue))) || code ||
0 == atomic_load_8((int8_t *)&ctx->queryContinue)) {
if (qComplete || (queryStop && (0 == atomic_load_8((int8_t *)&ctx->queryContinue))) || code) {
// Note: query is not running anymore
QW_SET_PHASE(ctx, 0);
QW_UNLOCK(QW_WRITE, &ctx->lock);