feature/qnode

include/common/tmsg.h

@@ -1090,6 +1090,7 @@ typedef struct SResFetchMsg {
 } SResFetchMsg;
 
 typedef struct SSchTasksStatusMsg {
+  SMsgHead header;
   uint64_t sId;
 } SSchTasksStatusMsg;
 
@@ -1105,6 +1106,7 @@ typedef struct SSchedulerStatusRsp {
 } SSchedulerStatusRsp;
 
 typedef struct STaskCancelMsg {
+  SMsgHead header;
   uint64_t sId;
   uint64_t queryId;
   uint64_t taskId;
@@ -1115,6 +1117,7 @@ typedef struct STaskCancelRsp {
 } STaskCancelRsp;
 
 typedef struct STaskDropMsg {
+  SMsgHead header;
   uint64_t sId;
   uint64_t queryId;
   uint64_t taskId;

source/libs/scheduler/inc/schedulerInt.h

@@ -60,17 +60,19 @@ typedef struct SSchLevel {
 
 
 typedef struct SSchTask {
-  uint64_t             taskId;     // task id
+  uint64_t             taskId;         // task id
-  SSchLevel           *level;      // level
+  SSchLevel           *level;          // level
-  SSubplan            *plan;       // subplan
+  SSubplan            *plan;           // subplan
-  char                *msg;        // operator tree
+  char                *msg;            // operator tree
-  int32_t              msgLen;     // msg length
+  int32_t              msgLen;         // msg length
-  int8_t               status;     // task status
+  int8_t               status;         // task status
-  SEpAddr              execAddr;   // task actual executed node address
+  SQueryNodeAddr       execAddr;       // task actual executed node address
-  SQueryProfileSummary summary;    // task execution summary
+  int8_t               condidateIdx;   // current try condidation index
-  int32_t              childReady; // child task ready number
+  SArray              *condidateAddrs; // condidate node addresses, element is SQueryNodeAddr
-  SArray              *children;   // the datasource tasks,from which to fetch the result, element is SQueryTask*
+  SQueryProfileSummary summary;        // task execution summary
-  SArray              *parents;    // the data destination tasks, get data from current task, element is SQueryTask*
+  int32_t              childReady;     // child task ready number
+  SArray              *children;       // the datasource tasks,from which to fetch the result, element is SQueryTask*
+  SArray              *parents;        // the data destination tasks, get data from current task, element is SQueryTask*
 } SSchTask;
 
 typedef struct SSchJobAttr {

source/libs/scheduler/src/scheduler.c

@@ -215,28 +215,49 @@ _return:
   SCH_RET(code);
 }
 
-int32_t schSetTaskExecEpSet(SSchJob *job, SEpSet *epSet) {  
+int32_t schSetTaskCondidateAddrs(SSchJob *job, SSchTask *task) {
-  if (epSet->numOfEps >= SCH_MAX_CONDIDATE_EP_NUM) {
+  if (task->condidateAddrs) {
     return TSDB_CODE_SUCCESS;
   }
 
-  int32_t nodeNum = taosArrayGetSize(job->nodeList);
+  task->condidateIdx = 0;
-  
+  task->condidateAddrs = taosArrayInit(SCH_MAX_CONDIDATE_EP_NUM, sizeof(SQueryNodeAddr));
-  for (int32_t i = 0; i < nodeNum && epSet->numOfEps < tListLen(epSet->port); ++i) {
+  if (NULL == task->condidateAddrs) {
-    SEpAddr *addr = taosArrayGet(job->nodeList, i);
+    qError("taosArrayInit failed");
-    
+    SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
-    strncpy(epSet->fqdn[epSet->numOfEps], addr->fqdn, sizeof(addr->fqdn));
-    epSet->port[epSet->numOfEps] = addr->port;
-    
-    ++epSet->numOfEps;
   }
 
-  for (int32_t i = 0; i < job->dataSrcEps.numOfEps && epSet->numOfEps < tListLen(epSet->port); ++i) {
+  if (task->plan->execNode.numOfEps > 0) {
+    if (NULL == taosArrayPush(task->condidateAddrs, &task->plan->execNode)) {
+      qError("taosArrayPush failed");
+      SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
+    }
+
+    return TSDB_CODE_SUCCESS;
+  }
+
+  int32_t addNum = 0;
+  int32_t nodeNum = taosArrayGetSize(job->nodeList);
+  
+  for (int32_t i = 0; i < nodeNum && addNum < SCH_MAX_CONDIDATE_EP_NUM; ++i) {
+    SQueryNodeAddr *naddr = taosArrayGet(job->nodeList, i);
+    
+    if (NULL == taosArrayPush(task->condidateAddrs, &task->plan->execNode)) {
+      qError("taosArrayPush failed");
+      SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
+    }
+    
+    ++addNum;
+  }
+
+/*
+  for (int32_t i = 0; i < job->dataSrcEps.numOfEps && addNum < SCH_MAX_CONDIDATE_EP_NUM; ++i) {
     strncpy(epSet->fqdn[epSet->numOfEps], job->dataSrcEps.fqdn[i], sizeof(job->dataSrcEps.fqdn[i]));
     epSet->port[epSet->numOfEps] = job->dataSrcEps.port[i];
     
     ++epSet->numOfEps;
   }
+*/
 
   return TSDB_CODE_SUCCESS;
 }
@@ -394,12 +415,14 @@ int32_t schProcessOnTaskSuccess(SSchJob *job, SSchTask *task) {
     return TSDB_CODE_SUCCESS;
   }
 
+/*
   if (SCH_IS_DATA_SRC_TASK(task) && job->dataSrcEps.numOfEps < SCH_MAX_CONDIDATE_EP_NUM) {
     strncpy(job->dataSrcEps.fqdn[job->dataSrcEps.numOfEps], task->execAddr.fqdn, sizeof(task->execAddr.fqdn));
     job->dataSrcEps.port[job->dataSrcEps.numOfEps] = task->execAddr.port;
 
     ++job->dataSrcEps.numOfEps;
   }
+*/
 
   for (int32_t i = 0; i < parentNum; ++i) {
     SSchTask *par = *(SSchTask **)taosArrayGet(task->parents, i);
@@ -633,6 +656,16 @@ _return:
   SCH_RET(code);
 }
 
+void schConvertAddrToEpSet(SQueryNodeAddr *addr, SEpSet *epSet) {
+  epSet->inUse = addr->inUse;
+  epSet->numOfEps = addr->numOfEps;
+  
+  for (int8_t i = 0; i < epSet->numOfEps; ++i) {
+    strncpy(epSet->fqdn[i], addr->epAddr[i].fqdn, sizeof(addr->epAddr[i].fqdn));
+    epSet->port[i] = addr->epAddr[i].port;
+  }
+}
+
 
 int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, int32_t msgType) {
   uint32_t msgSize = 0;
@@ -665,6 +698,7 @@ int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, int32_t msgType) {
 
       SSubQueryMsg *pMsg = msg;
 
+      pMsg->header.vgId = htonl(task->plan->execNode.nodeId);
       pMsg->sId = htobe64(schMgmt.sId);
       pMsg->queryId = htobe64(job->queryId);
       pMsg->taskId = htobe64(task->taskId);
@@ -681,6 +715,8 @@ int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, int32_t msgType) {
       }
 
       SResReadyMsg *pMsg = msg;
+      
+      pMsg->header.vgId = htonl(task->plan->execNode.nodeId);      
       pMsg->sId = htobe64(schMgmt.sId);      
       pMsg->queryId = htobe64(job->queryId);
       pMsg->taskId = htobe64(task->taskId);      
@@ -698,6 +734,8 @@ int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, int32_t msgType) {
       }
     
       SResFetchMsg *pMsg = msg;
+      
+      pMsg->header.vgId = htonl(task->plan->execNode.nodeId);            
       pMsg->sId = htobe64(schMgmt.sId);      
       pMsg->queryId = htobe64(job->queryId);
       pMsg->taskId = htobe64(task->taskId);      
@@ -712,6 +750,8 @@ int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, int32_t msgType) {
       }
     
       STaskDropMsg *pMsg = msg;
+      
+      pMsg->header.vgId = htonl(task->plan->execNode.nodeId);            
       pMsg->sId = htobe64(schMgmt.sId);      
       pMsg->queryId = htobe64(job->queryId);
       pMsg->taskId = htobe64(task->taskId);      
@@ -723,7 +763,12 @@ int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, int32_t msgType) {
       break;
   }
 
-  SCH_ERR_JRET(schAsyncSendMsg(job->transport, &task->plan->execEpSet, job->queryId, task->taskId, msgType, msg, msgSize));
+  SEpSet epSet;
+  SQueryNodeAddr *addr = taosArrayGet(task->condidateAddrs, task->condidateIdx);
+  
+  schConvertAddrToEpSet(addr, &epSet);
+
+  SCH_ERR_JRET(schAsyncSendMsg(job->transport, &epSet, job->queryId, task->taskId, msgType, msg, msgSize));
 
   return TSDB_CODE_SUCCESS;
 
@@ -737,12 +782,10 @@ _return:
 int32_t schLaunchTask(SSchJob *job, SSchTask *task) {
   SSubplan *plan = task->plan;
   SCH_ERR_RET(qSubPlanToString(plan, &task->msg, &task->msgLen));
-  if (plan->execEpSet.numOfEps <= 0) {
+  SCH_ERR_RET(schSetTaskCondidateAddrs(job, task));
-    SCH_ERR_RET(schSetTaskExecEpSet(job, &plan->execEpSet));
-  }
 
-  if (plan->execEpSet.numOfEps <= 0) {
+  if (NULL == task->condidateAddrs || taosArrayGetSize(task->condidateAddrs) <= 0) {
-    SCH_TASK_ERR_LOG("invalid execEpSet num:%d", plan->execEpSet.numOfEps);
+    SCH_TASK_ERR_LOG("no valid condidate node for task:%"PRIx64, task->taskId);
     SCH_ERR_RET(TSDB_CODE_SCH_INTERNAL_ERROR);
   }