feature/qnode

include/common/tmsg.h

@@ -875,6 +875,7 @@ typedef struct SSubQueryMsg {
   uint64_t sId;
   uint64_t queryId;
   uint64_t taskId;
+  int8_t   taskType;
   uint32_t contentLen;
   char     msg[];
 } SSubQueryMsg;

include/libs/executor/executor.h

@@ -84,6 +84,13 @@ void* qGetResultRetrieveMsg(qTaskInfo_t qinfo);
  */
 int32_t qKillTask(qTaskInfo_t qinfo);
 
+/**
+ * kill the ongoing query asynchronously
+ * @param qinfo  qhandle
+ * @return
+ */
+int32_t qAsyncKillTask(qTaskInfo_t qinfo);
+
 /**
  * return whether query is completed or not
  * @param qinfo

include/libs/qcom/query.h

@@ -38,6 +38,11 @@ enum {
   JOB_TASK_STATUS_FREEING,
 };
 
+enum {
+  TASK_TYPE_PERSISTENT = 1,
+  TASK_TYPE_TEMP,
+};
+
 typedef struct STableComInfo {
   uint8_t numOfTags;      // the number of tags in schema
   uint8_t precision;      // the number of precision

include/util/taoserror.h

@@ -362,6 +362,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_QRY_DUPLICATTED_OPERATION     TAOS_DEF_ERROR_CODE(0, 0x0718)  //"Duplicatted operation")
 #define TSDB_CODE_QRY_TASK_MSG_ERROR            TAOS_DEF_ERROR_CODE(0, 0x0719)  //"Task message error")
 #define TSDB_CODE_QRY_JOB_FREED                 TAOS_DEF_ERROR_CODE(0, 0x071A)  //"Job freed")
+#define TSDB_CODE_QRY_TASK_STATUS_ERROR         TAOS_DEF_ERROR_CODE(0, 0x071B)  //"Task status error")
 
 // grant
 #define TSDB_CODE_GRANT_EXPIRED                 TAOS_DEF_ERROR_CODE(0, 0x0800)  //"License expired")

source/libs/executor/src/executorMain.c

@@ -317,6 +317,19 @@ int32_t qKillTask(qTaskInfo_t qinfo) {
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t qAsyncKillTask(qTaskInfo_t qinfo) {
+  SQInfo *pQInfo = (SQInfo *)qinfo;
+
+  if (pQInfo == NULL || !isValidQInfo(pQInfo)) {
+    return TSDB_CODE_QRY_INVALID_QHANDLE;
+  }
+
+  qDebug("QInfo:0x%"PRIx64" query async killed", pQInfo->qId);
+  setQueryKilled(pQInfo);
+
+  return TSDB_CODE_SUCCESS;
+}
+
 int32_t qIsTaskCompleted(qTaskInfo_t qinfo) {
   SExecTaskInfo *pTaskInfo = (SExecTaskInfo *)qinfo;
 

source/libs/qworker/inc/qworkerInt.h

@@ -84,6 +84,7 @@ typedef struct SQWMsg {
 
 typedef struct SQWPhaseInput {
   int8_t         status;
+  int8_t         taskType;
   int32_t        code;
   qTaskInfo_t    taskHandle;
   DataSinkHandle sinkHandle;
@@ -91,8 +92,7 @@ typedef struct SQWPhaseInput {
 
 typedef struct SQWPhaseOutput {
   int32_t rspCode;
-  bool    needStop;
-  bool    needRsp;
+  bool    needStop;  
 } SQWPhaseOutput;
 
 
@@ -104,10 +104,15 @@ typedef struct SQWTaskStatus {
 typedef struct SQWTaskCtx {
   SRWLatch        lock;
   int8_t          phase;
+  int8_t          taskType;
 
+  void           *readyConnection;
+  void           *dropConnection;
+  void           *cancelConnection;
+  
   bool            emptyRes;
   int8_t          queryContinue;
-  int8_t          inQueue;
+  int8_t          queryInQueue;
   int32_t         rspCode; 
 
   int8_t          events[QW_EVENT_MAX];
@@ -170,6 +175,10 @@ typedef struct SQWorkerMgmt {
 #define QW_TASK_WLOG(param, ...) qWarn("QW:%p QID:%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
 #define QW_TASK_DLOG(param, ...) qDebug("QW:%p QID:%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
 
+#define QW_TASK_ELOG_E(param) qError("QW:%p QID:%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId)
+#define QW_TASK_WLOG_E(param) qWarn("QW:%p QID:%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId)
+#define QW_TASK_DLOG_E(param) qDebug("QW:%p QID:%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId)
+
 #define QW_SCH_TASK_ELOG(param, ...) qError("QW:%p SID:%"PRIx64",QID:%"PRIx64",TID:%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__)
 #define QW_SCH_TASK_WLOG(param, ...) qWarn("QW:%p SID:%"PRIx64",QID:%"PRIx64",TID:%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__)
 #define QW_SCH_TASK_DLOG(param, ...) qDebug("QW:%p SID:%"PRIx64",QID:%"PRIx64",TID:%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__)

source/libs/qworker/inc/qworkerMsg.h

@@ -23,7 +23,7 @@ extern "C" {
 #include "qworkerInt.h"
 #include "dataSinkMgt.h"
 
-int32_t qwProcessQuery(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWMsg *qwMsg);
+int32_t qwProcessQuery(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWMsg *qwMsg, int8_t taskType);
 int32_t qwProcessCQuery(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWMsg *qwMsg);
 int32_t qwProcessReady(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWMsg *qwMsg);
 int32_t qwProcessFetch(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWMsg *qwMsg);

source/libs/qworker/src/qworkerMsg.c

@@ -290,7 +290,7 @@ int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
 
   QW_SCH_TASK_DLOG("processQuery start, node:%p", node);
 
-  QW_RET(qwProcessQuery(QW_FPARAMS(), &qwMsg));
+  QW_RET(qwProcessQuery(QW_FPARAMS(), &qwMsg, msg->taskType));
 
   QW_SCH_TASK_DLOG("processQuery end, node:%p", node);
 
@@ -374,7 +374,9 @@ int32_t qWorkerProcessStatusMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
     QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
   }  
 
+  SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt;
   msg->sId = htobe64(msg->sId);
+  uint64_t sId = msg->sId;
 
   SSchedulerStatusRsp *sStatus = NULL;
   

source/libs/qworker/test/CMakeLists.txt

@@ -8,7 +8,7 @@ AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
 ADD_EXECUTABLE(qworkerTest ${SOURCE_LIST})
 TARGET_LINK_LIBRARIES(
         qworkerTest
-        PUBLIC os util common transport gtest qcom planner qworker
+        PUBLIC os util common transport gtest qcom planner qworker executor
 )
 
 TARGET_INCLUDE_DIRECTORIES(

source/libs/qworker/test/qworkerTests.cpp

@@ -32,11 +32,75 @@
 #include "qworker.h"
 #include "stub.h"
 #include "addr_any.h"
+#include "executor.h"
+#include "dataSinkMgt.h"
 
 
 namespace {
 
-bool testStop = false;
+bool qwtTestEnableSleep = true;
+bool qwtTestStop = false;
+bool qwtTestDeadLoop = true;
+int32_t qwtTestMTRunSec = 10;
+int32_t qwtTestPrintNum = 100000;
+int32_t qwtTestCaseIdx = 0;
+int32_t qwtTestCaseNum = 4;
+
+void qwtInitLogFile() {
+  const char    *defaultLogFileNamePrefix = "taosdlog";
+  const int32_t  maxLogFileNum = 10;
+
+  tsAsyncLog = 0;
+  qDebugFlag = 159;
+
+  char temp[128] = {0};
+  sprintf(temp, "%s/%s", tsLogDir, defaultLogFileNamePrefix);
+  if (taosInitLog(temp, tsNumOfLogLines, maxLogFileNum) < 0) {
+    printf("failed to open log file in directory:%s\n", tsLogDir);
+  }
+
+}
+
+void qwtBuildQueryReqMsg(SRpcMsg *queryRpc) {
+  SSubQueryMsg *queryMsg = (SSubQueryMsg *)calloc(1, sizeof(SSubQueryMsg) + 100);
+  queryMsg->queryId = htobe64(1);
+  queryMsg->sId = htobe64(1);
+  queryMsg->taskId = htobe64(1);
+  queryMsg->contentLen = htonl(100);
+  queryRpc->pCont = queryMsg;
+  queryRpc->contLen = sizeof(SSubQueryMsg) + 100;
+}
+
+void qwtBuildReadyReqMsg(SResReadyReq *readyMsg, SRpcMsg *readyRpc) {
+  readyMsg->sId = htobe64(1);
+  readyMsg->queryId = htobe64(1);
+  readyMsg->taskId = htobe64(1);
+  readyRpc->pCont = readyMsg;
+  readyRpc->contLen = sizeof(SResReadyReq);
+}
+
+void qwtBuildFetchReqMsg(SResFetchReq *fetchMsg, SRpcMsg *fetchRpc) {
+  fetchMsg->sId = htobe64(1);
+  fetchMsg->queryId = htobe64(1);
+  fetchMsg->taskId = htobe64(1);
+  fetchRpc->pCont = fetchMsg;
+  fetchRpc->contLen = sizeof(SResFetchReq);
+}
+
+void qwtBuildDropReqMsg(STaskDropReq *dropMsg, SRpcMsg *dropRpc) {
+  dropMsg->sId = htobe64(1);
+  dropMsg->queryId = htobe64(1);
+  dropMsg->taskId = htobe64(1);
+  dropRpc->pCont = dropMsg;
+  dropRpc->contLen = sizeof(STaskDropReq);
+}
+
+void qwtBuildStatusReqMsg(SSchTasksStatusReq *statusMsg, SRpcMsg *statusRpc) {
+  statusMsg->sId = htobe64(1);
+  statusRpc->pCont = statusMsg;
+  statusRpc->contLen = sizeof(SSchTasksStatusReq);
+  statusRpc->msgType = TDMT_VND_TASKS_STATUS;
+}
 
 int32_t qwtStringToPlan(const char* str, SSubplan** subplan) {
   return 0;
@@ -48,6 +112,7 @@ int32_t qwtPutReqToQueue(void *node, struct SRpcMsg *pMsg) {
 
 
 void qwtRpcSendResponse(const SRpcMsg *pRsp) {
+/*
   if (TDMT_VND_TASKS_STATUS_RSP == pRsp->msgType) {
     SSchedulerStatusRsp *rsp = (SSchedulerStatusRsp *)pRsp->pCont;
     printf("task num:%d\n", rsp->num);
@@ -56,9 +121,63 @@ void qwtRpcSendResponse(const SRpcMsg *pRsp) {
       printf("qId:%"PRIx64",tId:%"PRIx64",status:%d\n", task->queryId, task->taskId, task->status);
     }
   }
+*/  
   return;
 }
 
+int32_t qwtCreateExecTask(void* tsdb, int32_t vgId, struct SSubplan* pPlan, qTaskInfo_t* pTaskInfo, DataSinkHandle* handle) {
+  int32_t idx = qwtTestCaseIdx % qwtTestCaseNum;
+  
+  if (0 == idx) {
+    *pTaskInfo = qwtTestCaseIdx;
+    *handle = qwtTestCaseIdx+1;
+  } else if (1 == idx) {
+    *pTaskInfo = NULL;
+    *handle = NULL;
+  } else if (2 == idx) {
+    *pTaskInfo = qwtTestCaseIdx;
+    *handle = NULL;
+  } else if (3 == idx) {
+    *pTaskInfo = NULL;
+    *handle = qwtTestCaseIdx;
+  }
+
+  ++qwtTestCaseIdx;
+  
+  return 0;
+}
+
+int32_t qwtExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t *useconds) {
+  return 0;
+}
+
+int32_t qwtKillTask(qTaskInfo_t qinfo) {
+  return 0;
+}
+
+void qwtDestroyTask(qTaskInfo_t qHandle) {
+
+}
+
+
+int32_t qwtPutDataBlock(DataSinkHandle handle, const SInputData* pInput, bool* pContinue) {
+  return 0;
+}
+
+void qwtEndPut(DataSinkHandle handle, uint64_t useconds) {
+}
+
+void qwtGetDataLength(DataSinkHandle handle, int32_t* pLen, bool* pQueryEnd) {
+}
+
+int32_t qwtGetDataBlock(DataSinkHandle handle, SOutputData* pOutput) {
+  return 0;
+}
+
+void qwtDestroyDataSinker(DataSinkHandle handle) {
+
+}
+
 
 
 void stubSetStringToPlan() {
@@ -74,11 +193,118 @@ void stubSetStringToPlan() {
   }
 }
 
+void stubSetExecTask() {
+  static Stub stub;
+  stub.set(qExecTask, qwtExecTask);
+  {
+    AddrAny any("libexecutor.so");
+    std::map<std::string,void*> result;
+    any.get_global_func_addr_dynsym("^qExecTask$", result);
+    for (const auto& f : result) {
+      stub.set(f.second, qwtExecTask);
+    }
+  }
+}
+
+
+
+void stubSetCreateExecTask() {
+  static Stub stub;
+  stub.set(qCreateExecTask, qwtCreateExecTask);
+  {
+    AddrAny any("libexecutor.so");
+    std::map<std::string,void*> result;
+    any.get_global_func_addr_dynsym("^qCreateExecTask$", result);
+    for (const auto& f : result) {
+      stub.set(f.second, qwtCreateExecTask);
+    }
+  }
+}
+
+void stubSetAsyncKillTask() {
+  static Stub stub;
+  stub.set(qAsyncKillTask, qwtKillTask);
+  {
+    AddrAny any("libexecutor.so");
+    std::map<std::string,void*> result;
+    any.get_global_func_addr_dynsym("^qAsyncKillTask$", result);
+    for (const auto& f : result) {
+      stub.set(f.second, qwtKillTask);
+    }
+  }
+}
+
+void stubSetDestroyTask() {
+  static Stub stub;
+  stub.set(qDestroyTask, qwtDestroyTask);
+  {
+    AddrAny any("libexecutor.so");
+    std::map<std::string,void*> result;
+    any.get_global_func_addr_dynsym("^qDestroyTask$", result);
+    for (const auto& f : result) {
+      stub.set(f.second, qwtDestroyTask);
+    }
+  }
+}
+
+
+void stubSetDestroyDataSinker() {
+  static Stub stub;
+  stub.set(dsDestroyDataSinker, qwtDestroyDataSinker);
+  {
+    AddrAny any("libexecutor.so");
+    std::map<std::string,void*> result;
+    any.get_global_func_addr_dynsym("^dsDestroyDataSinker$", result);
+    for (const auto& f : result) {
+      stub.set(f.second, qwtDestroyDataSinker);
+    }
+  }
+}
+
+void stubSetGetDataLength() {
+  static Stub stub;
+  stub.set(dsGetDataLength, qwtGetDataLength);
+  {
+    AddrAny any("libexecutor.so");
+    std::map<std::string,void*> result;
+    any.get_global_func_addr_dynsym("^dsGetDataLength$", result);
+    for (const auto& f : result) {
+      stub.set(f.second, qwtGetDataLength);
+    }
+  }
+}
+
+void stubSetEndPut() {
+  static Stub stub;
+  stub.set(dsEndPut, qwtEndPut);
+  {
+    AddrAny any("libexecutor.so");
+    std::map<std::string,void*> result;
+    any.get_global_func_addr_dynsym("^dsEndPut$", result);
+    for (const auto& f : result) {
+      stub.set(f.second, qwtEndPut);
+    }
+  }
+}
+
+void stubSetPutDataBlock() {
+  static Stub stub;
+  stub.set(dsPutDataBlock, qwtPutDataBlock);
+  {
+    AddrAny any("libexecutor.so");
+    std::map<std::string,void*> result;
+    any.get_global_func_addr_dynsym("^dsPutDataBlock$", result);
+    for (const auto& f : result) {
+      stub.set(f.second, qwtPutDataBlock);
+    }
+  }
+}
+
 void stubSetRpcSendResponse() {
   static Stub stub;
   stub.set(rpcSendResponse, qwtRpcSendResponse);
   {
-    AddrAny any("libplanner.so");
+    AddrAny any("libtransport.so");
     std::map<std::string,void*> result;
     any.get_global_func_addr_dynsym("^rpcSendResponse$", result);
     for (const auto& f : result) {
@@ -87,24 +313,35 @@ void stubSetRpcSendResponse() {
   }
 }
 
+void stubSetGetDataBlock() {
+  static Stub stub;
+  stub.set(dsGetDataBlock, qwtGetDataBlock);
+  {
+    AddrAny any("libtransport.so");
+    std::map<std::string,void*> result;
+    any.get_global_func_addr_dynsym("^dsGetDataBlock$", result);
+    for (const auto& f : result) {
+      stub.set(f.second, qwtGetDataBlock);
+    }
+  }
+}
+
+
 void *queryThread(void *param) {
   SRpcMsg queryRpc = {0};
   int32_t code = 0;
   uint32_t n = 0;
   void *mockPointer = (void *)0x1;    
   void *mgmt = param;
-  SSubQueryMsg *queryMsg = (SSubQueryMsg *)calloc(1, sizeof(SSubQueryMsg) + 100);
-  queryMsg->queryId = htobe64(1);
-  queryMsg->sId = htobe64(1);
-  queryMsg->taskId = htobe64(1);
-  queryMsg->contentLen = htonl(100);
-  queryRpc.pCont = queryMsg;
-  queryRpc.contLen = sizeof(SSubQueryMsg) + 100;
 
-  while (!testStop) {
-    qWorkerProcessQueryMsg(mockPointer, mgmt, &queryRpc);
-    usleep(rand()%5);
-    if (++n % 50000 == 0) {
+  while (!qwtTestStop) {
+    qwtBuildQueryReqMsg(&queryRpc);
+    qWorkerProcessQueryMsg(mockPointer, mgmt, &queryRpc);    
+    free(queryRpc.pCont);
+    if (qwtTestEnableSleep) {
+      usleep(rand()%5);
+    }
+    if (++n % qwtTestPrintNum == 0) {
       printf("query:%d\n", n);
     }
   }
@@ -119,16 +356,14 @@ void *readyThread(void *param) {
   void *mockPointer = (void *)0x1;    
   void *mgmt = param;
   SResReadyReq readyMsg = {0};
-  readyMsg.sId = htobe64(1);
-  readyMsg.queryId = htobe64(1);
-  readyMsg.taskId = htobe64(1);
-  readyRpc.pCont = &readyMsg;
-  readyRpc.contLen = sizeof(SResReadyReq);
 
-  while (!testStop) {
+  while (!qwtTestStop) {
+    qwtBuildReadyReqMsg(&readyMsg, &readyRpc);
     code = qWorkerProcessReadyMsg(mockPointer, mgmt, &readyRpc);
-    usleep(rand()%5);
-    if (++n % 50000 == 0) {
+    if (qwtTestEnableSleep) {
+      usleep(rand()%5);
+    }
+    if (++n % qwtTestPrintNum == 0) {
       printf("ready:%d\n", n);
     }    
   }
@@ -143,16 +378,14 @@ void *fetchThread(void *param) {
   void *mockPointer = (void *)0x1;    
   void *mgmt = param;
   SResFetchReq fetchMsg = {0};
-  fetchMsg.sId = htobe64(1);
-  fetchMsg.queryId = htobe64(1);
-  fetchMsg.taskId = htobe64(1);
-  fetchRpc.pCont = &fetchMsg;
-  fetchRpc.contLen = sizeof(SResFetchReq);
 
-  while (!testStop) {
+  while (!qwtTestStop) {
+    qwtBuildFetchReqMsg(&fetchMsg, &fetchRpc);
     code = qWorkerProcessFetchMsg(mockPointer, mgmt, &fetchRpc);
-    usleep(rand()%5);
-    if (++n % 50000 == 0) {
+    if (qwtTestEnableSleep) {
+      usleep(rand()%5);
+    }
+    if (++n % qwtTestPrintNum == 0) {
       printf("fetch:%d\n", n);
     }    
   }
@@ -167,16 +400,14 @@ void *dropThread(void *param) {
   void *mockPointer = (void *)0x1;    
   void *mgmt = param;
   STaskDropReq dropMsg = {0};  
-  dropMsg.sId = htobe64(1);
-  dropMsg.queryId = htobe64(1);
-  dropMsg.taskId = htobe64(1);
-  dropRpc.pCont = &dropMsg;
-  dropRpc.contLen = sizeof(STaskDropReq);
 
-  while (!testStop) {
+  while (!qwtTestStop) {
+    qwtBuildDropReqMsg(&dropMsg, &dropRpc);
     code = qWorkerProcessDropMsg(mockPointer, mgmt, &dropRpc);
-    usleep(rand()%5);
-    if (++n % 50000 == 0) {
+    if (qwtTestEnableSleep) {
+      usleep(rand()%5);
+    }
+    if (++n % qwtTestPrintNum == 0) {
       printf("drop:%d\n", n);
     }    
   }
@@ -191,16 +422,14 @@ void *statusThread(void *param) {
   void *mockPointer = (void *)0x1;    
   void *mgmt = param;
   SSchTasksStatusReq statusMsg = {0};
-  statusMsg.sId = htobe64(1);
-  statusRpc.pCont = &statusMsg;
-  statusRpc.contLen = sizeof(SSchTasksStatusReq);
-  statusRpc.msgType = TDMT_VND_TASKS_STATUS;
 
-  while (!testStop) {
-    statusMsg.sId = htobe64(1);
+  while (!qwtTestStop) {
+    qwtBuildStatusReqMsg(&statusMsg, &statusRpc);
     code = qWorkerProcessStatusMsg(mockPointer, mgmt, &statusRpc);
-    usleep(rand()%5);
-    if (++n % 50000 == 0) {
+    if (qwtTestEnableSleep) {
+      usleep(rand()%5);
+    }
+    if (++n % qwtTestPrintNum == 0) {
       printf("status:%d\n", n);
     }    
   }
@@ -209,6 +438,27 @@ void *statusThread(void *param) {
 }
 
 
+void *controlThread(void *param) {
+  SRpcMsg queryRpc = {0};
+  int32_t code = 0;
+  uint32_t n = 0;
+  void *mockPointer = (void *)0x1;    
+  void *mgmt = param;
+
+  while (!qwtTestStop) {
+    qwtBuildQueryReqMsg(&queryRpc);
+    qWorkerProcessQueryMsg(mockPointer, mgmt, &queryRpc);    
+    free(queryRpc.pCont);
+    if (qwtTestEnableSleep) {
+      usleep(rand()%5);
+    }
+    if (++n % qwtTestPrintNum == 0) {
+      printf("query:%d\n", n);
+    }
+  }
+
+  return NULL;
+}
 
 
 
@@ -224,6 +474,8 @@ TEST(seqTest, normalCase) {
   SRpcMsg fetchRpc = {0};
   SRpcMsg dropRpc = {0};
   SRpcMsg statusRpc = {0};
+
+  qwtInitLogFile();
   
   SSubQueryMsg *queryMsg = (SSubQueryMsg *)calloc(1, sizeof(SSubQueryMsg) + 100);
   queryMsg->queryId = htobe64(1);
@@ -262,6 +514,15 @@ TEST(seqTest, normalCase) {
   
   stubSetStringToPlan();
   stubSetRpcSendResponse();
+  stubSetExecTask();
+  stubSetCreateExecTask();
+  stubSetAsyncKillTask();
+  stubSetDestroyTask();
+  stubSetDestroyDataSinker();
+  stubSetGetDataLength();
+  stubSetEndPut();
+  stubSetPutDataBlock();
+  stubSetGetDataBlock();
   
   code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue);
   ASSERT_EQ(code, 0);
@@ -308,6 +569,8 @@ TEST(seqTest, cancelFirst) {
   SRpcMsg queryRpc = {0};
   SRpcMsg dropRpc = {0};
   SRpcMsg statusRpc = {0};
+
+  qwtInitLogFile();
   
   SSubQueryMsg *queryMsg = (SSubQueryMsg *)calloc(1, sizeof(SSubQueryMsg) + 100);
   queryMsg->queryId = htobe64(1);
@@ -348,7 +611,7 @@ TEST(seqTest, cancelFirst) {
   ASSERT_EQ(code, 0);
 
   code = qWorkerProcessQueryMsg(mockPointer, mgmt, &queryRpc);
-  ASSERT_EQ(code, 0);
+  ASSERT_EQ(code, TSDB_CODE_QRY_TASK_DROPPED);
 
   statusMsg.sId = htobe64(1);
   code = qWorkerProcessStatusMsg(mockPointer, mgmt, &statusRpc);
@@ -366,44 +629,16 @@ TEST(seqTest, randCase) {
   SRpcMsg fetchRpc = {0};
   SRpcMsg dropRpc = {0};
   SRpcMsg statusRpc = {0};
-  
-  SSubQueryMsg *queryMsg = (SSubQueryMsg *)calloc(1, sizeof(SSubQueryMsg) + 100);
-  queryMsg->queryId = htobe64(1);
-  queryMsg->sId = htobe64(1);
-  queryMsg->taskId = htobe64(1);
-  queryMsg->contentLen = htonl(100);
-  queryRpc.pCont = queryMsg;
-  queryRpc.contLen = sizeof(SSubQueryMsg) + 100;
-
   SResReadyReq readyMsg = {0};
-  readyMsg.sId = htobe64(1);
-  readyMsg.queryId = htobe64(1);
-  readyMsg.taskId = htobe64(1);
-  readyRpc.pCont = &readyMsg;
-  readyRpc.contLen = sizeof(SResReadyReq);
-
   SResFetchReq fetchMsg = {0};
-  fetchMsg.sId = htobe64(1);
-  fetchMsg.queryId = htobe64(1);
-  fetchMsg.taskId = htobe64(1);
-  fetchRpc.pCont = &fetchMsg;
-  fetchRpc.contLen = sizeof(SResFetchReq);
-
   STaskDropReq dropMsg = {0};  
-  dropMsg.sId = htobe64(1);
-  dropMsg.queryId = htobe64(1);
-  dropMsg.taskId = htobe64(1);
-  dropRpc.pCont = &dropMsg;
-  dropRpc.contLen = sizeof(STaskDropReq);
-
   SSchTasksStatusReq statusMsg = {0};
-  statusMsg.sId = htobe64(1);
-  statusRpc.pCont = &statusMsg;
-  statusRpc.contLen = sizeof(SSchTasksStatusReq);
-  statusRpc.msgType = TDMT_VND_TASKS_STATUS;
+
+  qwtInitLogFile();
   
   stubSetStringToPlan();
   stubSetRpcSendResponse();
+  stubSetCreateExecTask();
 
   srand(time(NULL));
   
@@ -416,20 +651,25 @@ TEST(seqTest, randCase) {
     int32_t r = rand() % maxr;
     
     if (r >= 0 && r < maxr/5) {
-      printf("Query,%d\n", t++);
+      printf("Query,%d\n", t++);      
+      qwtBuildQueryReqMsg(&queryRpc);
       code = qWorkerProcessQueryMsg(mockPointer, mgmt, &queryRpc);
+      free(queryRpc.pCont);
     } else if (r >= maxr/5 && r < maxr * 2/5) {
       printf("Ready,%d\n", t++);
+      qwtBuildReadyReqMsg(&readyMsg, &readyRpc);
       code = qWorkerProcessReadyMsg(mockPointer, mgmt, &readyRpc);
     } else if (r >= maxr * 2/5 && r < maxr* 3/5) {
       printf("Fetch,%d\n", t++);
+      qwtBuildFetchReqMsg(&fetchMsg, &fetchRpc);
       code = qWorkerProcessFetchMsg(mockPointer, mgmt, &fetchRpc);
     } else if (r >= maxr * 3/5 && r < maxr * 4/5) {
       printf("Drop,%d\n", t++);
+      qwtBuildDropReqMsg(&dropMsg, &dropRpc);
       code = qWorkerProcessDropMsg(mockPointer, mgmt, &dropRpc);
     } else if (r >= maxr * 4/5 && r < maxr-1) {
       printf("Status,%d\n", t++);
-      statusMsg.sId = htobe64(1);
+      qwtBuildStatusReqMsg(&statusMsg, &statusRpc);
       code = qWorkerProcessStatusMsg(mockPointer, mgmt, &statusRpc);
       ASSERT_EQ(code, 0);
     } else {
@@ -445,6 +685,8 @@ TEST(seqTest, multithreadRand) {
   void *mgmt = NULL;
   int32_t code = 0;
   void *mockPointer = (void *)0x1;
+
+  qwtInitLogFile();
   
   stubSetStringToPlan();
   stubSetRpcSendResponse();
@@ -464,16 +706,70 @@ TEST(seqTest, multithreadRand) {
   pthread_create(&(t4), &thattr, dropThread, NULL);
   pthread_create(&(t5), &thattr, statusThread, NULL);
 
-  int32_t t = 0;
-  int32_t maxr = 10001;
-  sleep(300);
-  testStop = true;
-  sleep(1);
+  while (true) {
+    if (qwtTestDeadLoop) {
+      sleep(1);
+    } else {
+      sleep(qwtTestMTRunSec);
+      break;
+    }
+  }
+  
+  qwtTestStop = true;
+  sleep(3);
+  
+  qWorkerDestroy(&mgmt);
+}
+
+TEST(rcTest, multithread) {
+  void *mgmt = NULL;
+  int32_t code = 0;
+  void *mockPointer = (void *)0x1;
+
+  qwtInitLogFile();
+  
+  stubSetStringToPlan();
+  stubSetRpcSendResponse();
+  stubSetExecTask();
+  stubSetCreateExecTask();
+  stubSetAsyncKillTask();
+  stubSetDestroyTask();
+  stubSetDestroyDataSinker();
+  stubSetGetDataLength();
+  stubSetEndPut();
+  stubSetPutDataBlock();
+  stubSetGetDataBlock();
+
+  srand(time(NULL));
+  
+  code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue);
+  ASSERT_EQ(code, 0);
+
+  pthread_attr_t thattr;
+  pthread_attr_init(&thattr);
+
+  pthread_t t1,t2,t3,t4,t5;
+  pthread_create(&(t1), &thattr, controlThread, mgmt);
+  pthread_create(&(t2), &thattr, queryQueueThread, NULL);
+  pthread_create(&(t3), &thattr, fetchQueueThread, NULL);
+
+  while (true) {
+    if (qwtTestDeadLoop) {
+      sleep(1);
+    } else {
+      sleep(qwtTestMTRunSec);
+      break;
+    }
+  }
+  
+  qwtTestStop = true;
+  sleep(3);
   
   qWorkerDestroy(&mgmt);
 }
 
 
+
 int main(int argc, char** argv) {
   testing::InitGoogleTest(&argc, argv);
   return RUN_ALL_TESTS();

source/libs/scheduler/src/scheduler.c

@@ -1102,6 +1102,7 @@ int32_t schBuildAndSendMsg(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr,
       pMsg->sId = htobe64(schMgmt.sId);
       pMsg->queryId = htobe64(pJob->queryId);
       pMsg->taskId = htobe64(pTask->taskId);
+      pMsg->taskType = TASK_TYPE_TEMP;      
       pMsg->contentLen = htonl(pTask->msgLen);
       memcpy(pMsg->msg, pTask->msg, pTask->msgLen);
       break;
@@ -1487,6 +1488,7 @@ int32_t schedulerConvertDagToTaskList(SQueryDag* pDag, SArray **pTasks) {
     pMsg->sId = htobe64(schMgmt.sId);
     pMsg->queryId = htobe64(plan->id.queryId);
     pMsg->taskId = htobe64(schGenUUID());
+    pMsg->taskType = TASK_TYPE_PERSISTENT;
     pMsg->contentLen = htonl(msgLen);
     memcpy(pMsg->msg, msg, msgLen);
 

source/util/src/terror.c

@@ -361,6 +361,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_QRY_TASK_DROPPING,            "Task dropping")
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_DUPLICATTED_OPERATION,    "Duplicatted operation")
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_TASK_MSG_ERROR,           "Task message error")
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_JOB_FREED,                "Job already freed")
+TAOS_DEFINE_ERROR(TSDB_CODE_QRY_TASK_STATUS_ERROR,        "Task status error")