Merge pull request #9336 from taosdata/feature/qnode

Feature/qnode

include/common/taosmsg.h

@@ -52,6 +52,10 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_CONNECT, "mq-connect" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_DISCONNECT, "mq-disconnect" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_SET_CUR, "mq-set-cur" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_RES_READY, "res-ready" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_TASKS_STATUS, "tasks-status" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CANCEL_TASK, "cancel-task" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_TASK, "drop-task" )
+
 // message from client to mnode
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CONNECT, "connect" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_ACCT, "create-acct" )	
@@ -1107,29 +1111,29 @@ typedef struct {
   /* data */
 } SUpdateTagValRsp;
 
-typedef struct SSchedulerQueryMsg {
+typedef struct SSubQueryMsg {
   uint64_t  schedulerId;
   uint64_t  queryId;
   uint64_t  taskId;
   uint32_t  contentLen;
   char      msg[];
-} SSchedulerQueryMsg;
+} SSubQueryMsg;
 
-typedef struct SSchedulerReadyMsg {
+typedef struct SResReadyMsg {
   uint64_t  schedulerId;
   uint64_t  queryId;
   uint64_t  taskId;
-} SSchedulerReadyMsg;
+} SResReadyMsg;
 
-typedef struct SSchedulerFetchMsg {
+typedef struct SResFetchMsg {
   uint64_t  schedulerId;
   uint64_t  queryId;
   uint64_t  taskId;
-} SSchedulerFetchMsg;
+} SResFetchMsg;
 
-typedef struct SSchedulerStatusMsg {
+typedef struct SSchTasksStatusMsg {
   uint64_t  schedulerId;
-} SSchedulerStatusMsg;
+} SSchTasksStatusMsg;
 
 typedef struct STaskStatus {
   uint64_t  queryId;
@@ -1143,11 +1147,17 @@ typedef struct SSchedulerStatusRsp {
 } SSchedulerStatusRsp;
 
 
-typedef struct SSchedulerCancelMsg {
+typedef struct STaskCancelMsg {
+  uint64_t  schedulerId;
+  uint64_t  queryId;
+  uint64_t  taskId;
+} STaskCancelMsg;
+
+typedef struct STaskDropMsg {
   uint64_t  schedulerId;
   uint64_t  queryId;
   uint64_t  taskId;
-} SSchedulerCancelMsg;
+} STaskDropMsg;
 
 
 #pragma pack(pop)

include/libs/qcom/query.h

@@ -25,12 +25,15 @@ extern "C" {
 #include "tlog.h"
 
 enum {
+  JOB_TASK_STATUS_NULL = 0,
   JOB_TASK_STATUS_NOT_START = 1,
   JOB_TASK_STATUS_EXECUTING,
+  JOB_TASK_STATUS_PARTIAL_SUCCEED,
   JOB_TASK_STATUS_SUCCEED,
   JOB_TASK_STATUS_FAILED,
   JOB_TASK_STATUS_CANCELLING,
-  JOB_TASK_STATUS_CANCELLED
+  JOB_TASK_STATUS_CANCELLED,
+  JOB_TASK_STATUS_DROPPING,
 };
 
 typedef struct STableComInfo {

include/libs/qworker/qworker.h

@@ -42,15 +42,17 @@ typedef struct {
 
 int32_t qWorkerInit(SQWorkerCfg *cfg, void **qWorkerMgmt);
 
-int32_t qWorkerProcessQueryMsg(void *qWorkerMgmt, SSchedulerQueryMsg *msg, SRpcMsg *rsp);
+int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, SRpcMsg **rsp);
 
-int32_t qWorkerProcessReadyMsg(void *qWorkerMgmt, SSchedulerReadyMsg *msg, SRpcMsg *rsp);
+int32_t qWorkerProcessReadyMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, SRpcMsg *rsp);
 
-int32_t qWorkerProcessStatusMsg(void *qWorkerMgmt, SSchedulerStatusMsg *msg, SRpcMsg *rsp);
+int32_t qWorkerProcessStatusMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, SRpcMsg *rsp);
 
-int32_t qWorkerProcessFetchMsg(void *qWorkerMgmt, SSchedulerFetchMsg *msg, SRpcMsg *rsp);
+int32_t qWorkerProcessFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, SRpcMsg *rsp);
 
-int32_t qWorkerProcessCancelMsg(void *qWorkerMgmt, SSchedulerCancelMsg *msg, SRpcMsg *rsp);
+int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, SRpcMsg *rsp);
+
+int32_t qWorkerProcessDropMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, SRpcMsg *rsp);
 
 void qWorkerDestroy(void **qWorkerMgmt);
 

include/util/taoserror.h

@@ -314,6 +314,11 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_QRY_INVALID_TIME_CONDITION    TAOS_DEF_ERROR_CODE(0, 0x070D)  //"invalid time condition")
 #define TSDB_CODE_QRY_SYS_ERROR                 TAOS_DEF_ERROR_CODE(0, 0x070E)  //"System error")
 #define TSDB_CODE_QRY_INVALID_INPUT             TAOS_DEF_ERROR_CODE(0, 0x070F)  //"invalid input")
+#define TSDB_CODE_QRY_SCH_NOT_EXIST             TAOS_DEF_ERROR_CODE(0, 0x0710)  //"Scheduler not exist")
+#define TSDB_CODE_QRY_TASK_NOT_EXIST            TAOS_DEF_ERROR_CODE(0, 0x0711)  //"Task not exist")
+#define TSDB_CODE_QRY_TASK_ALREADY_EXIST        TAOS_DEF_ERROR_CODE(0, 0x0712)  //"Task already exist")
+#define TSDB_CODE_QRY_RES_CACHE_NOT_EXIST       TAOS_DEF_ERROR_CODE(0, 0x0713)  //"Task result cache not exist")
+#define TSDB_CODE_QRY_TASK_CANCELLED            TAOS_DEF_ERROR_CODE(0, 0x0714)  //"Task cancelled")
 
 
 // grant

include/util/thash.h

@@ -33,6 +33,8 @@ typedef void (*_hash_free_fn_t)(void *);
 
 #define HASH_INDEX(v, c) ((v) & ((c)-1))
 
+#define HASH_NODE_EXIST(code) (code == -2)
+
 /**
  * murmur hash algorithm
  * @key  usually string

source/dnode/mgmt/impl/src/dndTransport.c

@@ -45,6 +45,10 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
   pMgmt->msgFp[TSDB_MSG_TYPE_MQ_CONNECT] = dndProcessVnodeWriteMsg;
   pMgmt->msgFp[TSDB_MSG_TYPE_MQ_DISCONNECT] = dndProcessVnodeWriteMsg;
   pMgmt->msgFp[TSDB_MSG_TYPE_MQ_SET_CUR] = dndProcessVnodeWriteMsg;
+  pMgmt->msgFp[TSDB_MSG_TYPE_RES_READY] = dndProcessVnodeFetchMsg;
+  pMgmt->msgFp[TSDB_MSG_TYPE_TASKS_STATUS] = dndProcessVnodeFetchMsg;
+  pMgmt->msgFp[TSDB_MSG_TYPE_CANCEL_TASK] = dndProcessVnodeFetchMsg;
+  pMgmt->msgFp[TSDB_MSG_TYPE_DROP_TASK] = dndProcessVnodeFetchMsg;
 
   // msg from client to mnode
   pMgmt->msgFp[TSDB_MSG_TYPE_CONNECT] = dndProcessMnodeReadMsg;

source/libs/catalog/test/catalogTests.cpp

@@ -29,6 +29,8 @@
 #include "catalog.h"
 #include "tep.h"
 #include "trpc.h"
+#include "stub.h"
+#include "addr_any.h"
 
 typedef struct SAppInstInfo {
   int64_t           numOfConns;
@@ -86,6 +88,27 @@ void sendCreateDbMsg(void *shandle, SEpSet *pEpSet) {
   ASSERT_EQ(rpcRsp.code, 0);
 }
 
+void __rpcSendRecv(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
+  SUseDbRsp *rspMsg = NULL; //todo
+
+  return;
+}
+
+
+void initTestEnv() {
+  static Stub stub;
+  stub.set(rpcSendRecv, __rpcSendRecv);
+  {
+    AddrAny any("libtransport.so");
+    std::map<std::string,void*> result;
+    any.get_global_func_addr_dynsym("^rpcSendRecv$", result);
+    for (const auto& f : result) {
+      stub.set(f.second, __rpcSendRecv);
+    }
+  }
+}
+
+
 }
 
 TEST(testCase, normalCase) {

source/libs/qworker/inc/qworkerInt.h

@@ -20,6 +20,8 @@
 extern "C" {
 #endif
 
+#include "tlockfree.h"
+
 #define QWORKER_DEFAULT_SCHEDULER_NUMBER 10000
 #define QWORKER_DEFAULT_RES_CACHE_NUMBER 10000
 #define QWORKER_DEFAULT_SCH_TASK_NUMBER 10000
@@ -30,36 +32,63 @@ enum {
   QW_READY_RESPONSED,
 };
 
-typedef struct SQWorkerTaskStatus {
-  int8_t status;
-  int8_t ready; 
+enum {
+  QW_TASK_INFO_STATUS = 1,
+  QW_TASK_INFO_READY,
+};
+
+enum {
+  QW_READ = 1,
+  QW_WRITE,
+};
+
+typedef struct SQWorkerTaskStatus {  
+  SRWLatch lock;
+  int32_t  code;
+  int8_t   status;
+  int8_t   ready; 
+  bool     cancel;
+  bool     drop;
 } SQWorkerTaskStatus;
 
 typedef struct SQWorkerResCache {
+  SRWLatch lock;
   void *data;
 } SQWorkerResCache;
 
-typedef struct SQWorkerSchTaskStatus {
+typedef struct SQWorkerSchStatus {
   int32_t   lastAccessTs; // timestamp in second
-  SHashObj *taskStatus;   // key:queryId+taskId, value: SQWorkerTaskStatus
-} SQWorkerSchTaskStatus;
+  SRWLatch  tasksLock;
+  SHashObj *tasksHash;   // key:queryId+taskId, value: SQWorkerTaskStatus
+} SQWorkerSchStatus;
 
 // Qnode/Vnode level task management
 typedef struct SQWorkerMgmt {
   SQWorkerCfg cfg;
-  SHashObj *scheduleHash;    //key: schedulerId, value: SQWorkerSchTaskStatus
+  SRWLatch  schLock;
+  SRWLatch  resLock;
+  SHashObj *schHash;    //key: schedulerId, value: SQWorkerSchStatus
   SHashObj *resHash;       //key: queryId+taskId, value: SQWorkerResCache
 } SQWorkerMgmt;
 
-#define QW_TASK_DONE(status) (status == JOB_TASK_STATUS_SUCCEED || status == JOB_TASK_STATUS_FAILED || status == status == JOB_TASK_STATUS_CANCELLED)
+#define QW_GOT_RES_DATA(data) (false)
+#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))
+#define QW_TASK_ALREADY_EXIST(code) (TSDB_CODE_QRY_TASK_ALREADY_EXIST == (code))
+#define QW_TASK_READY_RESP(status) (status == JOB_TASK_STATUS_SUCCEED || status == JOB_TASK_STATUS_FAILED || status == JOB_TASK_STATUS_CANCELLED || status == JOB_TASK_STATUS_PARTIAL_SUCCEED)
 #define QW_SET_QTID(id, qid, tid) do { *(uint64_t *)(id) = (qid); *(uint64_t *)((char *)(id) + sizeof(qid)) = (tid); } while (0)
 #define QW_GET_QTID(id, qid, tid) do { (qid) = *(uint64_t *)(id); (tid) = *(uint64_t *)((char *)(id) + sizeof(qid)); } while (0)
 
+
 #define QW_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; return _code; } } while (0)
 #define QW_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)
 #define QW_ERR_LRET(c,...) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { qError(__VA_ARGS__); terrno = _code; return _code; } } while (0)
 #define QW_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)
 
+#define QW_LOCK(type, _lock) (QW_READ == (type) ? taosRLockLatch(_lock) : taosWLockLatch(_lock))
+#define QW_UNLOCK(type, _lock) (QW_READ == (type) ? taosRUnLockLatch(_lock) : taosWUnLockLatch(_lock))
+
 
 #ifdef __cplusplus
 }

source/libs/scheduler/src/scheduler.c

@@ -296,44 +296,45 @@ int32_t schAsyncSendMsg(SQueryJob *job, SQueryTask *task, int32_t msgType) {
       }
 
       int32_t len = strlen(task->msg);
-      msgSize = sizeof(SSchedulerQueryMsg) + len;
+      msgSize = sizeof(SSubQueryMsg) + len + 1;
       msg = calloc(1, msgSize);
       if (NULL == msg) {
         qError("calloc %d failed", msgSize);
         SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
       }
 
-      SSchedulerQueryMsg *pMsg = msg;
+      SSubQueryMsg *pMsg = msg;
       
       pMsg->schedulerId = htobe64(schMgmt.schedulerId);
       pMsg->queryId = htobe64(job->queryId);
       pMsg->taskId = htobe64(task->taskId);
       pMsg->contentLen = htonl(len);
       memcpy(pMsg->msg, task->msg, len);
+      pMsg->msg[len] = 0;
       break;
     }
     case TSDB_MSG_TYPE_RES_READY: {
-      msgSize = sizeof(SSchedulerReadyMsg);
+      msgSize = sizeof(SResReadyMsg);
       msg = calloc(1, msgSize);
       if (NULL == msg) {
         qError("calloc %d failed", msgSize);
         SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
       }
 
-      SSchedulerReadyMsg *pMsg = msg;
+      SResReadyMsg *pMsg = msg;
       pMsg->queryId = htobe64(job->queryId);
       pMsg->taskId = htobe64(task->taskId);      
       break;
     }
     case TSDB_MSG_TYPE_FETCH: {
-      msgSize = sizeof(SSchedulerFetchMsg);
+      msgSize = sizeof(SResFetchMsg);
       msg = calloc(1, msgSize);
       if (NULL == msg) {
         qError("calloc %d failed", msgSize);
         SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
       }
     
-      SSchedulerFetchMsg *pMsg = msg;
+      SResFetchMsg *pMsg = msg;
       pMsg->queryId = htobe64(job->queryId);
       pMsg->taskId = htobe64(task->taskId);      
       break;

source/util/src/terror.c

@@ -323,7 +323,10 @@ TAOS_DEFINE_ERROR(TSDB_CODE_QRY_INCONSISTAN,              "File inconsistance in
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_INVALID_TIME_CONDITION,   "One valid time range condition expected")
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_SYS_ERROR,                "System error")
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_INVALID_INPUT,            "invalid input")
-
+TAOS_DEFINE_ERROR(TSDB_CODE_QRY_SCH_NOT_EXIST,            "Scheduler not exist")
+TAOS_DEFINE_ERROR(TSDB_CODE_QRY_TASK_NOT_EXIST,           "Task not exist")
+TAOS_DEFINE_ERROR(TSDB_CODE_QRY_TASK_ALREADY_EXIST,       "Task already exist")
+TAOS_DEFINE_ERROR(TSDB_CODE_QRY_RES_CACHE_NOT_EXIST,      "Task result cache not exist")
 
 // grant
 TAOS_DEFINE_ERROR(TSDB_CODE_GRANT_EXPIRED,                "License expired")

source/util/src/thash.c

@@ -291,7 +291,7 @@ int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, void *da
     // enable resize
     __rd_unlock(&pHashObj->lock, pHashObj->type);
 
-    return pHashObj->enableUpdate ? 0 : -1;
+    return pHashObj->enableUpdate ? 0 : -2;
   }
 }