Merge pull request #9914 from taosdata/feature/qnode

Feature/qnode

include/common/tmsgdef.h

@@ -172,7 +172,6 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_SHOW_TABLES, "vnode-show-tables", SVShowTablesReq, SVShowTablesRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_SHOW_TABLES_FETCH, "vnode-show-tables-fetch", SVShowTablesFetchReq, SVShowTablesFetchRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_QUERY_CONTINUE, "vnode-query-continue", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_SCHEDULE_DATA_SINK, "vnode-schedule-data-sink", NULL, NULL)
 
   TD_DEF_MSG_TYPE(TDMT_VND_SUBSCRIBE, "vnode-subscribe", SMVSubscribeReq, SMVSubscribeRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_CONSUME, "vnode-consume", SMqCVConsumeReq, SMqCVConsumeRsp)

include/libs/executor/dataSinkMgt.h

@@ -48,7 +48,6 @@ typedef struct SOutputData {
   int8_t  compressed;
   char*   pData;
   bool    queryEnd;
-  int32_t scheduleJobNo;
   int32_t bufStatus;
   int64_t useconds;
   int8_t  precision;

source/dnode/vnode/src/vnd/vnodeQuery.c

@@ -29,8 +29,6 @@ int vnodeProcessQueryReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
       return qWorkerProcessQueryMsg(pVnode->pTsdb, pVnode->pQuery, pMsg);
     case TDMT_VND_QUERY_CONTINUE:
       return qWorkerProcessCQueryMsg(pVnode->pTsdb, pVnode->pQuery, pMsg);
-    case TDMT_VND_SCHEDULE_DATA_SINK:
-      return qWorkerProcessDataSinkMsg(pVnode->pTsdb, pVnode->pQuery, pMsg);
     default:
       vError("unknown msg type:%d in query queue", pMsg->msgType);
       return TSDB_CODE_VND_APP_ERROR;

source/libs/executor/src/dataDispatcher.c

@@ -196,7 +196,6 @@ static int32_t getDataBlock(SDataSinkHandle* pHandle, SOutputData* pOutput) {
   pOutput->bufStatus = updateStatus(pDispatcher);
   pthread_mutex_lock(&pDispatcher->mutex);
   pOutput->queryEnd = pDispatcher->queryEnd;
-  pOutput->scheduleJobNo = 0;
   pOutput->useconds = pDispatcher->useconds;
   pOutput->precision = pDispatcher->schema.precision;
   pthread_mutex_unlock(&pDispatcher->mutex);

source/libs/qworker/inc/qworkerInt.h

@@ -31,8 +31,6 @@ enum {
   QW_PHASE_POST_QUERY,
   QW_PHASE_PRE_CQUERY,
   QW_PHASE_POST_CQUERY,
-  QW_PHASE_PRE_SINK,
-  QW_PHASE_POST_SINK,
   QW_PHASE_PRE_FETCH,
   QW_PHASE_POST_FETCH,
 };
@@ -105,10 +103,12 @@ typedef struct SQWTaskStatus {
 
 typedef struct SQWTaskCtx {
   SRWLatch        lock;
-  int32_t         phase;
+  int8_t          phase;
 
-  int32_t         sinkId;
-  int32_t         readyCode; 
+  bool            emptyRes;
+  int8_t          queryContinue;
+  int8_t          inQueue;
+  int32_t         rspCode; 
 
   int8_t          events[QW_EVENT_MAX];
   
@@ -144,7 +144,11 @@ typedef struct SQWorkerMgmt {
 #define QW_SET_EVENT_RECEIVED(ctx, event) atomic_store_8(&(ctx)->events[event], QW_EVENT_RECEIVED)
 #define QW_SET_EVENT_PROCESSED(ctx, event) atomic_store_8(&(ctx)->events[event], QW_EVENT_PROCESSED)
 
-#define QW_IN_EXECUTOR(ctx) ((ctx)->phase == QW_PHASE_PRE_QUERY || (ctx)->phase == QW_PHASE_PRE_CQUERY || (ctx)->phase == QW_PHASE_PRE_FETCH || (ctx)->phase == QW_PHASE_PRE_SINK)
+#define QW_GET_PHASE(ctx) atomic_load_8(&(ctx)->phase)
+
+#define QW_SET_RSP_CODE(ctx, code) atomic_val_compare_exchange_32(&(ctx)->rspCode, 0, code)
+
+#define QW_IN_EXECUTOR(ctx) (QW_GET_PHASE(ctx) == QW_PHASE_PRE_QUERY || QW_GET_PHASE(ctx) == QW_PHASE_PRE_CQUERY || QW_GET_PHASE(ctx) == QW_PHASE_PRE_FETCH)
 
 #define QW_TASK_NOT_EXIST(code) (TSDB_CODE_QRY_SCH_NOT_EXIST == (code) || TSDB_CODE_QRY_TASK_NOT_EXIST == (code))
 #define QW_TASK_ALREADY_EXIST(code) (TSDB_CODE_QRY_TASK_ALREADY_EXIST == (code))

source/libs/qworker/src/qworker.c

@@ -254,6 +254,19 @@ int32_t qwAcquireTaskCtx(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t qwGetTaskCtx(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWTaskCtx **ctx) {
+  char id[sizeof(qId) + sizeof(tId)] = {0};
+  QW_SET_QTID(id, qId, tId);
+  
+  *ctx = taosHashGet(mgmt->ctxHash, id, sizeof(id));
+  if (NULL == (*ctx)) {
+    QW_TASK_ELOG("ctx not in ctxHash, id:%s", id);
+    QW_ERR_RET(TSDB_CODE_QRY_RES_CACHE_NOT_EXIST);
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
 int32_t qwAddTaskCtxImpl(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, int32_t rwType, int32_t status, SQWTaskCtx **ctx) {
   char id[sizeof(qId) + sizeof(tId)] = {0};
   QW_SET_QTID(id, qId, tId);
@@ -270,6 +283,8 @@ int32_t qwAddTaskCtxImpl(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_
     if (HASH_NODE_EXIST(code)) {
       if (rwType && ctx) {
         QW_RET(qwAcquireTaskCtx(QW_FPARAMS(), rwType, ctx));
+      } else if (ctx) {
+        QW_RET(qwGetTaskCtx(QW_FPARAMS(), ctx));
       } else {
         QW_TASK_ELOG("task ctx already exist, id:%s", id);
         QW_ERR_RET(TSDB_CODE_QRY_TASK_ALREADY_EXIST);
@@ -283,6 +298,8 @@ int32_t qwAddTaskCtxImpl(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_
 
   if (rwType && ctx) {
     QW_RET(qwAcquireTaskCtx(QW_FPARAMS(), rwType, ctx));
+  } else if (ctx) {
+    QW_RET(qwGetTaskCtx(QW_FPARAMS(), ctx));
   }
 
   return TSDB_CODE_SUCCESS;
@@ -292,39 +309,48 @@ int32_t qwAddTaskCtx(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tI
   QW_RET(qwAddTaskCtxImpl(QW_FPARAMS(), 0, 0, NULL));
 }
 
-int32_t qwGetTaskCtx(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWTaskCtx **ctx) {
-  char id[sizeof(qId) + sizeof(tId)] = {0};
-  QW_SET_QTID(id, qId, tId);
-  
-  *ctx = taosHashGet(mgmt->ctxHash, id, sizeof(id));
-  if (NULL == (*ctx)) {
-    QW_TASK_ELOG("ctx not in ctxHash, id:%s", id);
-    QW_ERR_RET(TSDB_CODE_QRY_RES_CACHE_NOT_EXIST);
-  }
-
-  return TSDB_CODE_SUCCESS;
-
-}
 
 
 int32_t qwAddAcquireTaskCtx(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, int32_t rwType, SQWTaskCtx **ctx) {
   return qwAddTaskCtxImpl(QW_FPARAMS(), rwType, 0, ctx);
 }
 
+int32_t qwAddGetTaskCtx(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWTaskCtx **ctx) {
+  return qwAddTaskCtxImpl(QW_FPARAMS(), 0, 0, ctx);
+}
+
+
 void qwReleaseTaskCtx(int32_t rwType, SQWorkerMgmt *mgmt) {
   QW_UNLOCK(rwType, &mgmt->ctxLock);
 }
 
+void qwFreeTaskHandle(QW_FPARAMS_DEF, SQWTaskCtx *ctx) {  
+  // RC WARNING
+  qTaskInfo_t taskHandle = atomic_load_ptr(&ctx->taskHandle);
+  if (taskHandle && atomic_val_compare_exchange_ptr(&ctx->taskHandle, taskHandle, NULL)) {
+    qDestroyTask(taskHandle);
+  }
+}
 
-void qwFreeTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWTaskCtx *ctx) {
-  if (ctx->taskHandle) {
-    qDestroyTask(ctx->taskHandle);
-    ctx->taskHandle = NULL;
+int32_t qwKillTaskHandle(QW_FPARAMS_DEF, SQWTaskCtx *ctx) {
+  int32_t code = 0;
+  // RC WARNING
+  qTaskInfo_t taskHandle = atomic_load_ptr(&ctx->taskHandle);
+  if (taskHandle && atomic_val_compare_exchange_ptr(&ctx->taskHandle, taskHandle, NULL)) {
+    code = qKillTask(taskHandle);
+    atomic_store_ptr(&ctx->taskHandle, taskHandle);
   }
 
-  // TODO
-  if (ctx->sinkHandle) {
+  QW_RET(code);
+}
+
+
+void qwFreeTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx) {
+  qwFreeTaskHandle(QW_FPARAMS(), ctx);
   
+  if (ctx->sinkHandle) {
+    dsDestroyDataSinker(ctx->sinkHandle);
+    ctx->sinkHandle = NULL;
   }
 }
 
@@ -418,7 +444,7 @@ int32_t qwDropTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId,
   SQWTaskCtx *ctx = NULL;
   bool locked = false;
 
-  QW_ERR_JRET(qwAddAcquireTaskCtx(QW_FPARAMS(), QW_READ, &ctx));
+  QW_ERR_JRET(qwAddGetTaskCtx(QW_FPARAMS(), &ctx));
   
   QW_LOCK(QW_WRITE, &ctx->lock);
 
@@ -430,9 +456,7 @@ int32_t qwDropTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId,
   }
 
   if (QW_IN_EXECUTOR(ctx)) {
-    if (ctx->taskHandle) {
-      QW_ERR_JRET(qKillTask(ctx->taskHandle));
-    }
+    QW_ERR_JRET(qwKillTaskHandle(QW_FPARAMS(), ctx));
     
     QW_ERR_JRET(qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_DROPPING));
   } else if (ctx->phase > 0) {
@@ -449,12 +473,12 @@ int32_t qwDropTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId,
 
 _return:
 
-  if (locked) {
-    QW_UNLOCK(QW_WRITE, &ctx->lock);
+  if (code) {
+    QW_SET_RSP_CODE(ctx, code);
   }
 
-  if (ctx) {
-    qwReleaseTaskCtx(QW_READ, mgmt);
+  if (locked) {
+    QW_UNLOCK(QW_WRITE, &ctx->lock);
   }
 
   QW_RET(code);
@@ -465,8 +489,11 @@ int32_t qwExecTask(QW_FPARAMS_DEF, qTaskInfo_t taskHandle, DataSinkHandle sinkHa
   bool  qcontinue = true;
   SSDataBlock* pRes = NULL;
   uint64_t useconds = 0;
+  int32_t i = 0;
+ 
+  while (true) {
+    QW_TASK_DLOG("start to execTask in executor, loopIdx:%d", i++);
     
-  while (qcontinue) {
     code = qExecTask(taskHandle, &pRes, &useconds);
     if (code) {
       QW_TASK_ELOG("qExecTask failed, code:%x", code);
@@ -474,7 +501,7 @@ int32_t qwExecTask(QW_FPARAMS_DEF, qTaskInfo_t taskHandle, DataSinkHandle sinkHa
     }
 
     if (NULL == pRes) {
-      QW_TASK_DLOG("query done, useconds:%"PRIu64, useconds);
+      QW_TASK_DLOG("task query done, useconds:%"PRIu64, useconds);
       dsEndPut(sinkHandle, useconds);
       break;
     }
@@ -485,6 +512,12 @@ int32_t qwExecTask(QW_FPARAMS_DEF, qTaskInfo_t taskHandle, DataSinkHandle sinkHa
       QW_TASK_ELOG("dsPutDataBlock failed, code:%x", code);
       QW_ERR_JRET(code);
     }
+
+    QW_TASK_DLOG("data put into sink, rows:%d, continueExecTask:%d", pRes->info.rows, qcontinue);
+    
+    if (!qcontinue) {
+      break;
+    }
   }
 
 _return:
@@ -499,6 +532,20 @@ int32_t qwGetResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, void
   bool queryEnd = false;
   int32_t code = 0;
 
+  if (ctx->emptyRes) {
+    QW_TASK_DLOG("query empty result, query end, phase:%d", ctx->phase);
+    
+    QW_ERR_RET(qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_SUCCEED));
+    
+    QW_ERR_RET(qwMallocFetchRsp(len, &rsp));      
+    
+    *rspMsg = rsp;
+    *dataLen = 0;
+    pOutput->queryEnd = true;
+    
+    return TSDB_CODE_SUCCESS;
+  }
+
   dsGetDataLength(ctx->sinkHandle, &len, &queryEnd);
 
   if (len < 0) {
@@ -519,9 +566,7 @@ int32_t qwGetResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, void
       QW_ERR_RET(qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_SUCCEED));
 
       QW_ERR_RET(qwMallocFetchRsp(len, &rsp));      
-
       *rspMsg = rsp;
-
       *dataLen = 0;
       
       return TSDB_CODE_SUCCESS;
@@ -540,15 +585,12 @@ int32_t qwGetResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, void
   QW_TASK_DLOG("task got data in sink, dataLength:%d", len);
   
   QW_ERR_RET(qwMallocFetchRsp(len, &rsp));
-
   *rspMsg = rsp;
   
   pOutput->pData = rsp->data;
-  
   code = dsGetDataBlock(ctx->sinkHandle, pOutput);
   if (code) {
     QW_TASK_ELOG("dsGetDataBlock failed, code:%x", code);
-    qwFreeFetchRsp(rsp);
     QW_ERR_RET(code);
   }
 
@@ -576,7 +618,7 @@ int32_t qwHandleTaskEvent(QW_FPARAMS_DEF, int32_t phase, SQWPhaseInput *input, S
 
   switch (phase) {
     case QW_PHASE_PRE_QUERY: {
-      QW_ERR_JRET(qwAddAcquireTaskCtx(QW_FPARAMS(), QW_READ, &ctx));
+      QW_ERR_JRET(qwAddGetTaskCtx(QW_FPARAMS(), &ctx));
             
       ctx->phase = phase;
 
@@ -596,7 +638,6 @@ int32_t qwHandleTaskEvent(QW_FPARAMS_DEF, int32_t phase, SQWPhaseInput *input, S
         output->needStop = true;
         
         QW_ERR_JRET(qwAddTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_CANCELLED));
-        qwFreeTask(QW_FPARAMS(), ctx);
         
         QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_CANCEL);
         
@@ -609,7 +650,7 @@ int32_t qwHandleTaskEvent(QW_FPARAMS_DEF, int32_t phase, SQWPhaseInput *input, S
       break;
     }
     case QW_PHASE_POST_QUERY: {
-      QW_ERR_JRET(qwAddAcquireTaskCtx(QW_FPARAMS(), QW_READ, &ctx));
+      QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx));
       
       QW_LOCK(QW_WRITE, &ctx->lock);
       
@@ -618,7 +659,13 @@ int32_t qwHandleTaskEvent(QW_FPARAMS_DEF, int32_t phase, SQWPhaseInput *input, S
       ctx->taskHandle = input->taskHandle;
       ctx->sinkHandle = input->sinkHandle;
 
-      ctx->readyCode = input->code;
+      if (NULL == ctx->taskHandle && NULL == ctx->sinkHandle) {
+        ctx->emptyRes = true;
+      }
+      
+      if (input->code) {
+        QW_SET_RSP_CODE(ctx, input->code);
+      }
 
       assert(!QW_IS_EVENT_PROCESSED(ctx, QW_EVENT_CANCEL));
 
@@ -654,8 +701,78 @@ int32_t qwHandleTaskEvent(QW_FPARAMS_DEF, int32_t phase, SQWPhaseInput *input, S
       }
       break;
     }
+    case QW_PHASE_PRE_CQUERY: {
+      QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx));
+      
+      QW_LOCK(QW_WRITE, &ctx->lock);
+      
+      locked = true;
+
+      ctx->phase = phase;
+
+      if (QW_IS_EVENT_PROCESSED(ctx, QW_EVENT_CANCEL)) {
+        QW_TASK_WLOG("task already cancelled, phase:%d", phase);
+        output->needStop = true;
+        output->rspCode = TSDB_CODE_QRY_TASK_CANCELLED;        
+        QW_ERR_JRET(TSDB_CODE_QRY_TASK_CANCELLED);
+      }
+
+      if (QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_DROP)) {
+        QW_TASK_WLOG("task is dropping, phase:%d", phase);
+        output->needStop = true;
+        output->rspCode = TSDB_CODE_QRY_TASK_DROPPING;
+      } else if (QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_CANCEL)) {
+        QW_TASK_WLOG("task is cancelling, phase:%d", phase);
+        output->needStop = true;
+        output->rspCode = TSDB_CODE_QRY_TASK_CANCELLING;
+      }
+
+      if (ctx->rspCode) {
+        QW_TASK_ELOG("task already failed, code:%x, phase:%d", ctx->rspCode, phase);
+        output->needStop = true;
+        output->rspCode = ctx->rspCode;        
+        QW_ERR_JRET(output->rspCode);
+      }
+      break;
+    }    
+    case QW_PHASE_POST_CQUERY: {
+      QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx));
+      
+      QW_LOCK(QW_WRITE, &ctx->lock);
+      
+      locked = true;        
+
+      if (input->code) {
+        QW_SET_RSP_CODE(ctx, input->code);
+      }
+      
+      if (QW_IS_EVENT_PROCESSED(ctx, QW_EVENT_CANCEL)) {
+        QW_TASK_WLOG("task already cancelled, phase:%d", phase);
+        output->needStop = true;
+        output->rspCode = TSDB_CODE_QRY_TASK_CANCELLED;        
+        QW_ERR_JRET(TSDB_CODE_QRY_TASK_CANCELLED);
+      }
+
+      if (QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_DROP)) {
+        QW_TASK_WLOG("task is dropping, phase:%d", phase);
+        output->needStop = true;
+        output->rspCode = TSDB_CODE_QRY_TASK_DROPPING;
+      } else if (QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_CANCEL)) {
+        QW_TASK_WLOG("task is cancelling, phase:%d", phase);
+        output->needStop = true;
+        output->rspCode = TSDB_CODE_QRY_TASK_CANCELLING;
+      }
+
+      if (ctx->rspCode) {
+        QW_TASK_ELOG("task failed, code:%x, phase:%d", ctx->rspCode, phase);
+        output->needStop = true;
+        output->rspCode = ctx->rspCode;        
+        QW_ERR_JRET(output->rspCode);
+      }      
+      break;
+    }    
     case QW_PHASE_PRE_FETCH: {
-      QW_ERR_JRET(qwAddAcquireTaskCtx(QW_FPARAMS(), QW_READ, &ctx));
+      QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx));
       
       QW_LOCK(QW_WRITE, &ctx->lock);
       
@@ -693,15 +810,26 @@ int32_t qwHandleTaskEvent(QW_FPARAMS_DEF, int32_t phase, SQWPhaseInput *input, S
         output->rspCode = TSDB_CODE_QRY_TASK_MSG_ERROR;        
         QW_ERR_JRET(TSDB_CODE_QRY_TASK_MSG_ERROR);
       }
+
+      if (ctx->rspCode) {
+        QW_TASK_ELOG("task already failed, code:%x, phase:%d", ctx->rspCode, phase);
+        output->needStop = true;
+        output->rspCode = ctx->rspCode;        
+        QW_ERR_JRET(output->rspCode);
+      }
       break;
     }    
     case QW_PHASE_POST_FETCH: {
-      QW_ERR_JRET(qwAddAcquireTaskCtx(QW_FPARAMS(), QW_READ, &ctx));
+      QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx));
       
       QW_LOCK(QW_WRITE, &ctx->lock);
       
       locked = true;        
 
+      if (input->code) {
+        QW_SET_RSP_CODE(ctx, input->code);
+      }
+      
       if (QW_IS_EVENT_PROCESSED(ctx, QW_EVENT_CANCEL)) {
         QW_TASK_WLOG("task already cancelled, phase:%d", phase);
         output->needStop = true;
@@ -718,12 +846,17 @@ int32_t qwHandleTaskEvent(QW_FPARAMS_DEF, int32_t phase, SQWPhaseInput *input, S
         output->needStop = true;
         output->rspCode = TSDB_CODE_QRY_TASK_CANCELLING;
       }
+
+      if (ctx->rspCode) {
+        QW_TASK_ELOG("task failed, code:%x, phase:%d", ctx->rspCode, phase);
+        output->needStop = true;
+        output->rspCode = ctx->rspCode;        
+        QW_ERR_JRET(output->rspCode);
+      }      
       break;
     }    
-
   }
 
-
 _return:
 
   if (locked) {
@@ -732,10 +865,6 @@ _return:
     QW_UNLOCK(QW_WRITE, &ctx->lock);
   }
 
-  if (ctx) {
-    qwReleaseTaskCtx(QW_READ, mgmt);
-  }
-
   QW_RET(code);
 }
 
@@ -774,12 +903,14 @@ int32_t qwProcessQuery(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t
     QW_ERR_JRET(code);
   }
   
-  QW_ERR_JRET(qwBuildAndSendQueryRsp(qwMsg->connection, TSDB_CODE_SUCCESS));
+  QW_ERR_JRET(qwBuildAndSendQueryRsp(qwMsg->connection, code));
+  QW_TASK_DLOG("query msg rsped, code:%d", code);
 
   queryRsped = true;
 
+  if (pTaskInfo && sinkHandle) {
     QW_ERR_JRET(qwExecTask(QW_FPARAMS(), pTaskInfo, sinkHandle));
-
+  }
 _return:
 
   if (code) {
@@ -788,6 +919,10 @@ _return:
   
   if (!queryRsped) {
     code = qwBuildAndSendQueryRsp(qwMsg->connection, rspCode);
+    if (TSDB_CODE_SUCCESS == code) {
+      QW_TASK_DLOG("query msg rsped, code:%d", rspCode);
+    }
+    
     if (TSDB_CODE_SUCCESS == rspCode && code) {
       rspCode = code;
     }
@@ -811,6 +946,7 @@ _return:
 
   if (queryRsped && output.needRsp) {
     qwBuildAndSendReadyRsp(qwMsg->connection, output.rspCode);    
+    QW_TASK_DLOG("ready msg rsped, code:%x", output.rspCode);
   }
   
   QW_RET(rspCode);
@@ -819,24 +955,34 @@ _return:
 int32_t qwProcessReady(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWMsg *qwMsg) {
   int32_t code = 0;
   SQWTaskCtx *ctx = NULL;
+  int8_t phase = 0;
   
-  QW_ERR_JRET(qwAddAcquireTaskCtx(QW_FPARAMS(), QW_READ, &ctx));
+  QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx));
 
   QW_LOCK(QW_WRITE, &ctx->lock);
 
-  if (ctx->phase == QW_PHASE_PRE_QUERY) {
+  phase = QW_GET_PHASE(ctx);
+  
+  if (phase == QW_PHASE_PRE_QUERY) {
     QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_READY);
-  } else if (ctx->phase == QW_PHASE_POST_QUERY) {
+    QW_TASK_DLOG("ready msg not rsped, phase:%d", phase);
+  } else if (phase == QW_PHASE_POST_QUERY) {
     QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_READY);
-    QW_ERR_JRET(qwBuildAndSendReadyRsp(qwMsg->connection, ctx->readyCode));
+    QW_ERR_JRET(qwBuildAndSendReadyRsp(qwMsg->connection, ctx->rspCode));
+    QW_TASK_DLOG("ready msg rsped, code:%x", ctx->rspCode);
+  } else {
+    QW_TASK_ELOG("invalid phase when got ready msg, phase:%d", phase);
+    assert(0);
   }
 
 _return:
 
+  if (code) {
+    QW_SET_RSP_CODE(ctx, code);
+  }
+
   if (ctx) {
     QW_UNLOCK(QW_WRITE, &ctx->lock);
-
-    qwReleaseTaskCtx(QW_READ, mgmt);
   }
 
   QW_RET(code);
@@ -844,17 +990,17 @@ _return:
 
 
 int32_t qwProcessCQuery(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWMsg *qwMsg) {
+  SQWTaskCtx *ctx = NULL;
   int32_t code = 0;
   bool queryRsped = false;
   bool needStop = false;
   struct SSubplan *plan = NULL;
-  int32_t rspCode = 0;
   SQWPhaseInput input = {0};
   SQWPhaseOutput output = {0};
-  SQWTaskCtx *ctx = NULL;
   void *rsp = NULL;
   int32_t dataLen = 0;
   
+  do {
     QW_ERR_JRET(qwHandleTaskEvent(QW_FPARAMS(), QW_PHASE_PRE_CQUERY, &input, &output));
 
     needStop = output.needStop;
@@ -867,61 +1013,54 @@ int32_t qwProcessCQuery(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t
 
     QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx));
 
+    atomic_store_8(&ctx->inQueue, 0);
+
     qTaskInfo_t     taskHandle = ctx->taskHandle;
     DataSinkHandle  sinkHandle = ctx->sinkHandle;
 
     QW_ERR_JRET(qwExecTask(QW_FPARAMS(), taskHandle, sinkHandle));
 
-  QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_CQUERY);
-
     if (QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_FETCH)) {
       SOutputData sOutput = {0};
       QW_ERR_JRET(qwGetResFromSink(QW_FPARAMS(), ctx, &dataLen, &rsp, &sOutput));
       
-    // Note: schedule data sink firstly and will schedule query after it's done
-    if (sOutput.scheduleJobNo) {
-      if (sOutput.scheduleJobNo > ctx->sinkId) {
-        QW_TASK_DLOG("sink need schedule, scheduleJobNo:%d", sOutput.scheduleJobNo);    
-    
-        ctx->sinkId = sOutput.scheduleJobNo;
-        QW_ERR_JRET(qwBuildAndSendSchSinkMsg(QW_FPARAMS(), qwMsg->connection));
-      }
-    } else if ((!sOutput.queryEnd) && (DS_BUF_LOW == sOutput.bufStatus || DS_BUF_EMPTY == sOutput.bufStatus)) {    
+      if ((!sOutput.queryEnd) && (DS_BUF_LOW == sOutput.bufStatus || DS_BUF_EMPTY == sOutput.bufStatus)) {    
         QW_TASK_DLOG("task not end, need to continue, bufStatus:%d", sOutput.bufStatus);
         
-      if (!QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_CQUERY)) {
-        QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_CQUERY);
-    
-        QW_ERR_JRET(qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_EXECUTING));      
-        
-        QW_ERR_RET(qwBuildAndSendCQueryMsg(QW_FPARAMS(), qwMsg->connection));
-      }
+        // RC WARNING
+        atomic_store_8(&ctx->queryContinue, 1);
       }
       
       if (rsp) {
         qwBuildFetchRsp(rsp, &sOutput, dataLen); 
-    }
         
-  }
+        QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_FETCH);            
         
-_return:
+        qwBuildAndSendFetchRsp(qwMsg->connection, rsp, dataLen, code);                
+        QW_TASK_DLOG("fetch msg rsped, code:%x, dataLen:%d", code, dataLen);
+      } else {
+        atomic_store_8(&ctx->queryContinue, 1);
+      }
+    }
 
-  qwHandleTaskEvent(QW_FPARAMS(), QW_PHASE_POST_CQUERY, &input, &output);
+  _return:
 
-  if (QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_FETCH)) {    
-    if (code) {
+    if (code && QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_FETCH)) {
       QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_FETCH);    
       qwFreeFetchRsp(rsp);
       rsp = NULL;
       qwBuildAndSendFetchRsp(qwMsg->connection, rsp, 0, code);
-    } else if (rsp) {
-      QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_FETCH);
-    
-      qwBuildAndSendFetchRsp(qwMsg->connection, rsp, dataLen, code);
-    }
+      QW_TASK_DLOG("fetch msg rsped, code:%x, dataLen:%d", code, 0);      
     }
 
-  QW_RET(rspCode);
+    input.code = code;
+    qwHandleTaskEvent(QW_FPARAMS(), QW_PHASE_POST_CQUERY, &input, &output);    
+
+    needStop = output.needStop;
+    code = output.rspCode;    
+  } while ((!needStop) && (0 == code) && atomic_val_compare_exchange_8(&ctx->queryContinue, 1, 0));
+
+  QW_RET(code);
 }
 
 
@@ -958,46 +1097,49 @@ int32_t qwProcessFetch(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t
 
   if (NULL == rsp) {
     QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_FETCH);
+  } else {
+    qwBuildFetchRsp(rsp, &sOutput, dataLen);
   }
 
-  // Note: schedule data sink firstly and will schedule query after it's done
-  if (sOutput.scheduleJobNo) {
-    if (sOutput.scheduleJobNo > ctx->sinkId) {
-      QW_TASK_DLOG("sink need schedule, scheduleJobNo:%d", sOutput.scheduleJobNo);    
-
-      ctx->sinkId = sOutput.scheduleJobNo;
-      
-      QW_ERR_JRET(qwBuildAndSendSchSinkMsg(QW_FPARAMS(), qwMsg->connection));
-    }
-  } else if ((!sOutput.queryEnd) && (/* DS_BUF_LOW == sOutput.bufStatus || */ DS_BUF_EMPTY == sOutput.bufStatus)) {    
+  if ((!sOutput.queryEnd) && (DS_BUF_LOW == sOutput.bufStatus || DS_BUF_EMPTY == sOutput.bufStatus)) {    
     QW_TASK_DLOG("task not end, need to continue, bufStatus:%d", sOutput.bufStatus);
 
-    if (!QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_CQUERY)) {
-      QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_CQUERY);
+    QW_LOCK(QW_WRITE, &ctx->lock);
+    locked = true;
 
+    // RC WARNING
+    if (QW_IN_EXECUTOR(ctx)) {
+      atomic_store_8(&ctx->queryContinue, 1);
+    } else if (0 == atomic_load_8(&ctx->inQueue)) {
       QW_ERR_JRET(qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_EXECUTING));      
 
-      QW_ERR_RET(qwBuildAndSendCQueryMsg(QW_FPARAMS(), qwMsg->connection));
+      atomic_store_8(&ctx->inQueue, 1);
+      
+      QW_ERR_JRET(qwBuildAndSendCQueryMsg(QW_FPARAMS(), qwMsg->connection));
     }
   }
   
-  if (rsp) {
-    qwBuildFetchRsp(rsp, &sOutput, dataLen);
+_return:
+
+  if (locked) {
+    QW_UNLOCK(QW_WRITE, &ctx->lock);
   }
 
-_return:
+  input.code = code;
 
   qwHandleTaskEvent(QW_FPARAMS(), QW_PHASE_POST_FETCH, &input, &output);
 
   if (code) {
     qwFreeFetchRsp(rsp);
     rsp = NULL;
-    qwBuildAndSendFetchRsp(qwMsg->connection, rsp, 0, code);
+    dataLen = 0;
+    qwBuildAndSendFetchRsp(qwMsg->connection, rsp, dataLen, code);
+    QW_TASK_DLOG("fetch msg rsped, code:%x, dataLen:%d", code, dataLen);
   } else if (rsp) {
     qwBuildAndSendFetchRsp(qwMsg->connection, rsp, dataLen, code);
+    QW_TASK_DLOG("fetch msg rsped, code:%x, dataLen:%d", code, dataLen);
   }
 
-  
   QW_RET(code);
 }
 

source/libs/qworker/src/qworkerMsg.c

@@ -229,42 +229,6 @@ int32_t qwBuildAndSendShowFetchRsp(SRpcMsg *pMsg, SVShowTablesFetchReq* pFetchRe
   return TSDB_CODE_SUCCESS;
 }
 
-
-int32_t qwBuildAndSendSchSinkMsg(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, void *connection) {
-  SRpcMsg *pMsg = (SRpcMsg *)connection;
-  SSinkDataReq * req = (SSinkDataReq *)rpcMallocCont(sizeof(SSinkDataReq));
-  if (NULL == req) {
-    qError("rpcMallocCont %d failed", (int32_t)sizeof(SSinkDataReq));
-    QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
-  }
-
-  req->header.vgId = mgmt->nodeId;
-  req->sId = sId;
-  req->queryId = qId;
-  req->taskId = tId;
-
-  SRpcMsg pNewMsg = {
-    .handle = pMsg->handle,
-    .ahandle = pMsg->ahandle, 
-    .msgType = TDMT_VND_SCHEDULE_DATA_SINK,
-    .pCont   = req,
-    .contLen = sizeof(SSinkDataReq),
-    .code    = 0,
-  };
-
-  int32_t code = (*mgmt->putToQueueFp)(mgmt->nodeObj, &pNewMsg);
-  if (TSDB_CODE_SUCCESS != code) {
-    qError("put data sink schedule msg to queue failed, code:%x", code);
-    rpcFreeCont(req);
-    QW_ERR_RET(code);
-  }
-
-  qDebug("put data sink schedule msg to query queue");
-
-  return TSDB_CODE_SUCCESS;
-}
-
-
 int32_t qwBuildAndSendCQueryMsg(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, void *connection) {
   SRpcMsg *pMsg = (SRpcMsg *)connection;
   SQueryContinueReq * req = (SQueryContinueReq *)rpcMallocCont(sizeof(SQueryContinueReq));
@@ -366,25 +330,6 @@ int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   return TSDB_CODE_SUCCESS;    
 }
 
-
-
-int32_t qWorkerProcessDataSinkMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg){
-  if (NULL == node || NULL == qWorkerMgmt || NULL == pMsg) {
-    return TSDB_CODE_QRY_INVALID_INPUT;
-  }
-
-  SSinkDataReq *msg = pMsg->pCont;
-  if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
-    qError("invalid sink data msg");
-    QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
-  }
-
-  //dsScheduleProcess();
-  //TODO
-
-  return TSDB_CODE_SUCCESS;
-}
-
 int32_t qWorkerProcessReadyMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg){
   if (NULL == node || NULL == qWorkerMgmt || NULL == pMsg) {
     return TSDB_CODE_QRY_INVALID_INPUT;