feature/qnode

include/common/tmsg.h

@@ -1002,7 +1002,7 @@ typedef struct {
 } SUpdateTagValRsp;
 
 typedef struct SSubQueryMsg {
-  uint64_t schedulerId;
+  uint64_t sId;
   uint64_t queryId;
   uint64_t taskId;
   uint32_t contentLen;
@@ -1010,7 +1010,7 @@ typedef struct SSubQueryMsg {
 } SSubQueryMsg;
 
 typedef struct SResReadyMsg {
-  uint64_t schedulerId;
+  uint64_t sId;
   uint64_t queryId;
   uint64_t taskId;
 } SResReadyMsg;
@@ -1020,13 +1020,13 @@ typedef struct SResReadyRsp {
 } SResReadyRsp;
 
 typedef struct SResFetchMsg {
-  uint64_t schedulerId;
+  uint64_t sId;
   uint64_t queryId;
   uint64_t taskId;
 } SResFetchMsg;
 
 typedef struct SSchTasksStatusMsg {
-  uint64_t schedulerId;
+  uint64_t sId;
 } SSchTasksStatusMsg;
 
 typedef struct STaskStatus {
@@ -1041,7 +1041,7 @@ typedef struct SSchedulerStatusRsp {
 } SSchedulerStatusRsp;
 
 typedef struct STaskCancelMsg {
-  uint64_t schedulerId;
+  uint64_t sId;
   uint64_t queryId;
   uint64_t taskId;
 } STaskCancelMsg;
@@ -1051,7 +1051,7 @@ typedef struct STaskCancelRsp {
 } STaskCancelRsp;
 
 typedef struct STaskDropMsg {
-  uint64_t schedulerId;
+  uint64_t sId;
   uint64_t queryId;
   uint64_t taskId;
 } STaskDropMsg;

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

@@ -22,13 +22,27 @@ int vnodeQueryOpen(SVnode *pVnode) {
 
 int vnodeProcessQueryReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
   vInfo("query message is processed");
-  qWorkerProcessQueryMsg(pVnode, pVnode->pQuery, pMsg);
-  return 0;
+  return qWorkerProcessQueryMsg(pVnode, pVnode->pQuery, pMsg);
 }
 
 int vnodeProcessFetchReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
   vInfo("fetch message is processed");
-  qWorkerProcessFetchMsg(pVnode, pVnode->pQuery, pMsg);
+  switch (pMsg->msgType) {
+    case TDMT_VND_FETCH:
+      return qWorkerProcessFetchMsg(pVnode, pVnode->pQuery, pMsg);
+    case TDMT_VND_RES_READY:
+      return qWorkerProcessReadyMsg(pVnode, pVnode->pQuery, pMsg);
+    case TDMT_VND_TASKS_STATUS:
+      return qWorkerProcessStatusMsg(pVnode, pVnode->pQuery, pMsg);
+    case TDMT_VND_CANCEL_TASK:
+      return qWorkerProcessCancelMsg(pVnode, pVnode->pQuery, pMsg);
+    case TDMT_VND_DROP_TASK:
+      return qWorkerProcessDropMsg(pVnode, pVnode->pQuery, pMsg);
+    default:
+      vError("unknown msg type:%d in fetch queue", pMsg->msgType);
+      return TSDB_CODE_VND_APP_ERROR;
+      break;
+  }
   return 0;
 }
 

source/libs/qworker/inc/qworkerInt.h

@@ -42,25 +42,41 @@ enum {
   QW_WRITE,
 };
 
-typedef struct SQWorkerTaskStatus {  
+enum {
+  QW_EXIST_ACQUIRE = 1,
+  QW_EXIST_RET_ERR,
+};
+
+enum {
+  QW_NOT_EXIST_RET_ERR = 1,
+  QW_NOT_EXIST_ADD,
+};
+
+enum {
+  QW_ADD_RET_ERR = 1,
+  QW_ADD_ACQUIRE,
+};
+
+
+typedef struct SQWTaskStatus {  
   SRWLatch lock;
   int32_t  code;
   int8_t   status;
   int8_t   ready; 
   bool     cancel;
   bool     drop;
-} SQWorkerTaskStatus;
+} SQWTaskStatus;
 
 typedef struct SQWorkerResCache {
   SRWLatch lock;
   void *data;
 } SQWorkerResCache;
 
-typedef struct SQWorkerSchStatus {
+typedef struct SQWSchStatus {
   int32_t   lastAccessTs; // timestamp in second
   SRWLatch  tasksLock;
   SHashObj *tasksHash;   // key:queryId+taskId, value: SQWorkerTaskStatus
-} SQWorkerSchStatus;
+} SQWSchStatus;
 
 // Qnode/Vnode level task management
 typedef struct SQWorkerMgmt {
@@ -71,7 +87,7 @@ typedef struct SQWorkerMgmt {
   SHashObj *resHash;       //key: queryId+taskId, value: SQWorkerResCache
 } SQWorkerMgmt;
 
-#define QW_GOT_RES_DATA(data) (false)
+#define QW_GOT_RES_DATA(data) (true)
 #define QW_LOW_RES_DATA(data) (false)
 
 #define QW_TASK_NOT_EXIST(code) (TSDB_CODE_QRY_SCH_NOT_EXIST == (code) || TSDB_CODE_QRY_TASK_NOT_EXIST == (code))
@@ -89,6 +105,8 @@ typedef struct SQWorkerMgmt {
 #define QW_LOCK(type, _lock) (QW_READ == (type) ? taosRLockLatch(_lock) : taosWLockLatch(_lock))
 #define QW_UNLOCK(type, _lock) (QW_READ == (type) ? taosRUnLockLatch(_lock) : taosWUnLockLatch(_lock))
 
+static int32_t qwAcquireScheduler(int32_t rwType, SQWorkerMgmt *mgmt, uint64_t sId, SQWSchStatus **sch, int32_t nOpt);
+
 
 #ifdef __cplusplus
 }

source/libs/qworker/test/qworkerTests.cpp

@@ -40,6 +40,11 @@ int32_t qwtStringToPlan(const char* str, SSubplan** subplan) {
   return 0;
 }
 
+void qwtRpcSendResponse(const SRpcMsg *pRsp) {
+  return;
+}
+
+
 
 void stubSetStringToPlan() {
   static Stub stub;
@@ -54,6 +59,20 @@ void stubSetStringToPlan() {
   }
 }
 
+void stubSetRpcSendResponse() {
+  static Stub stub;
+  stub.set(rpcSendResponse, qwtRpcSendResponse);
+  {
+    AddrAny any("libplanner.so");
+    std::map<std::string,void*> result;
+    any.get_global_func_addr_dynsym("^rpcSendResponse$", result);
+    for (const auto& f : result) {
+      stub.set(f.second, qwtRpcSendResponse);
+    }
+  }
+}
+
+
 
 }
 
@@ -68,30 +87,35 @@ TEST(testCase, normalCase) {
   SRpcMsg dropRpc = {0};
   SSubQueryMsg *queryMsg = (SSubQueryMsg *)calloc(1, sizeof(SSubQueryMsg) + 100);
   queryMsg->queryId = htobe64(1);
-  queryMsg->schedulerId = htobe64(1);
+  queryMsg->sId = htobe64(1);
   queryMsg->taskId = htobe64(1);
   queryMsg->contentLen = htonl(100);
   queryRpc.pCont = queryMsg;
+  queryRpc.contLen = sizeof(SSubQueryMsg) + 100;
 
   SResReadyMsg readyMsg = {0};
-  readyMsg.schedulerId = htobe64(1);
+  readyMsg.sId = htobe64(1);
   readyMsg.queryId = htobe64(1);
   readyMsg.taskId = htobe64(1);
   readyRpc.pCont = &readyMsg;
+  readyRpc.contLen = sizeof(SResReadyMsg);
 
   SResFetchMsg fetchMsg = {0};
-  fetchMsg.schedulerId = htobe64(1);
+  fetchMsg.sId = htobe64(1);
   fetchMsg.queryId = htobe64(1);
   fetchMsg.taskId = htobe64(1);
   fetchRpc.pCont = &fetchMsg;
+  fetchRpc.contLen = sizeof(SResFetchMsg);
 
   STaskDropMsg dropMsg = {0};  
-  dropMsg.schedulerId = htobe64(1);
+  dropMsg.sId = htobe64(1);
   dropMsg.queryId = htobe64(1);
   dropMsg.taskId = htobe64(1);
   dropRpc.pCont = &dropMsg;
+  dropRpc.contLen = sizeof(STaskDropMsg);
 
   stubSetStringToPlan();
+  stubSetRpcSendResponse();
   
   code = qWorkerInit(NULL, &mgmt);
   ASSERT_EQ(code, 0);
@@ -107,7 +131,8 @@ TEST(testCase, normalCase) {
 
   code = qWorkerProcessDropMsg(mockPointer, mgmt, &dropRpc);
   ASSERT_EQ(code, 0);
-  
+
+  qWorkerDestroy(&mgmt);
 }
 
 

source/libs/scheduler/inc/schedulerInt.h

@@ -38,7 +38,7 @@ enum {
 
 typedef struct SSchedulerMgmt {
   uint64_t  taskId; 
-  uint64_t  schedulerId;
+  uint64_t  sId;
   SSchedulerCfg cfg;
   SHashObj *jobs;  // key: queryId, value: SQueryJob*
 } SSchedulerMgmt;

source/libs/scheduler/src/scheduler.c

@@ -343,7 +343,7 @@ int32_t schAsyncSendMsg(SSchJob *job, SSchTask *task, int32_t msgType) {
 
       SSubQueryMsg *pMsg = msg;
       
-      pMsg->schedulerId = htobe64(schMgmt.schedulerId);
+      pMsg->sId = htobe64(schMgmt.sId);
       pMsg->queryId = htobe64(job->queryId);
       pMsg->taskId = htobe64(task->taskId);
       pMsg->contentLen = htonl(task->msgLen);
@@ -359,7 +359,7 @@ int32_t schAsyncSendMsg(SSchJob *job, SSchTask *task, int32_t msgType) {
       }
 
       SResReadyMsg *pMsg = msg;
-      pMsg->schedulerId = htobe64(schMgmt.schedulerId);      
+      pMsg->sId = htobe64(schMgmt.sId);      
       pMsg->queryId = htobe64(job->queryId);
       pMsg->taskId = htobe64(task->taskId);      
       break;
@@ -376,7 +376,7 @@ int32_t schAsyncSendMsg(SSchJob *job, SSchTask *task, int32_t msgType) {
       }
     
       SResFetchMsg *pMsg = msg;
-      pMsg->schedulerId = htobe64(schMgmt.schedulerId);      
+      pMsg->sId = htobe64(schMgmt.sId);      
       pMsg->queryId = htobe64(job->queryId);
       pMsg->taskId = htobe64(task->taskId);      
       break;
@@ -390,7 +390,7 @@ int32_t schAsyncSendMsg(SSchJob *job, SSchTask *task, int32_t msgType) {
       }
     
       STaskDropMsg *pMsg = msg;
-      pMsg->schedulerId = htobe64(schMgmt.schedulerId);      
+      pMsg->sId = htobe64(schMgmt.sId);      
       pMsg->queryId = htobe64(job->queryId);
       pMsg->taskId = htobe64(task->taskId);      
       break;
@@ -717,7 +717,7 @@ int32_t schedulerInit(SSchedulerCfg *cfg) {
     SCH_ERR_LRET(TSDB_CODE_QRY_OUT_OF_MEMORY, "init %d schduler jobs failed", schMgmt.cfg.maxJobNum);
   }
 
-  schMgmt.schedulerId = 1; //TODO GENERATE A UUID
+  schMgmt.sId = 1; //TODO GENERATE A UUID
   
   return TSDB_CODE_SUCCESS;
 }