mndQuery.c 5.8 KB
Newer Older
D
dapan1121 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "mndQuery.h"
#include "executor.h"
S
Shengliang Guan 已提交
18
#include "mndMnode.h"
D
dapan1121 已提交
19 20
#include "qworker.h"

21
int32_t mndPreProcessQueryMsg(SRpcMsg *pMsg) {
D
dapan1121 已提交
22
  if (TDMT_SCH_QUERY != pMsg->msgType && TDMT_SCH_MERGE_QUERY != pMsg->msgType) return 0;
23
  SMnode *pMnode = pMsg->info.node;
D
dapan1121 已提交
24 25 26
  return qWorkerPreprocessQueryMsg(pMnode->pQuery, pMsg);
}

27
void mndPostProcessQueryMsg(SRpcMsg *pMsg) {
D
dapan1121 已提交
28
  if (TDMT_SCH_QUERY != pMsg->msgType && TDMT_SCH_MERGE_QUERY != pMsg->msgType) return;
D
dapan1121 已提交
29 30 31 32
  SMnode *pMnode = pMsg->info.node;
  qWorkerAbortPreprocessQueryMsg(pMnode->pQuery, pMsg);
}

S
Shengliang Guan 已提交
33
int32_t mndProcessQueryMsg(SRpcMsg *pMsg) {
S
Shengliang Guan 已提交
34
  int32_t     code = -1;
S
Shengliang Guan 已提交
35
  SMnode     *pMnode = pMsg->info.node;
36
  SReadHandle handle = {.mnd = pMnode, .pMsgCb = &pMnode->msgCb};
D
dapan1121 已提交
37

S
Shengliang Guan 已提交
38 39
  mTrace("msg:%p, in query queue is processing", pMsg);
  switch (pMsg->msgType) {
D
dapan1121 已提交
40
    case TDMT_SCH_QUERY:
D
dapan1121 已提交
41
    case TDMT_SCH_MERGE_QUERY:
D
dapan1121 已提交
42
      code = qWorkerProcessQueryMsg(&handle, pMnode->pQuery, pMsg, 0);
S
Shengliang Guan 已提交
43
      break;
D
dapan1121 已提交
44
    case TDMT_SCH_QUERY_CONTINUE:
D
dapan1121 已提交
45
      code = qWorkerProcessCQueryMsg(&handle, pMnode->pQuery, pMsg, 0);
S
Shengliang Guan 已提交
46
      break;
D
dapan1121 已提交
47
    case TDMT_SCH_FETCH:
D
dapan1121 已提交
48
    case TDMT_SCH_MERGE_FETCH:
D
dapan1121 已提交
49
      code = qWorkerProcessFetchMsg(pMnode, pMnode->pQuery, pMsg, 0);
S
Shengliang Guan 已提交
50
      break;
D
dapan1121 已提交
51
    case TDMT_SCH_DROP_TASK:
D
dapan1121 已提交
52
      code = qWorkerProcessDropMsg(pMnode, pMnode->pQuery, pMsg, 0);
S
Shengliang Guan 已提交
53
      break;
D
dapan1121 已提交
54
    case TDMT_SCH_QUERY_HEARTBEAT:
D
dapan1121 已提交
55
      code = qWorkerProcessHbMsg(pMnode, pMnode->pQuery, pMsg, 0);
S
Shengliang Guan 已提交
56
      break;
D
dapan1121 已提交
57
    default:
S
Shengliang Guan 已提交
58
      terrno = TSDB_CODE_VND_APP_ERROR;
S
Shengliang Guan 已提交
59
      mError("unknown msg type:%d in query queue", pMsg->msgType);
D
dapan1121 已提交
60
  }
S
Shengliang Guan 已提交
61 62 63

  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
  return code;
D
dapan1121 已提交
64 65
}

66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86
int32_t mndProcessBatchMetaMsg(SRpcMsg *pMsg) {
  int32_t code = 0;
  int32_t offset = 0;
  int32_t rspSize = 0;
  SBatchReq *batchReq = (SBatchReq*)pMsg->pCont;
  int32_t msgNum = ntohl(batchReq->msgNum);
  offset += sizeof(SBatchReq);
  SBatchMsg req = {0};
  SBatchRsp rsp = {0};
  SRpcMsg reqMsg = *pMsg;
  SRpcMsg rspMsg = {0};
  void* pRsp = NULL;
  SMnode *pMnode = pMsg->info.node;

  SArray* batchRsp = taosArrayInit(msgNum, sizeof(SBatchRsp));
  if (NULL == batchRsp) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }
  
  for (int32_t i = 0; i < msgNum; ++i) {
D
dapan1121 已提交
87 88 89
    req.msgIdx = ntohl(*(int32_t*)((char*)pMsg->pCont + offset));
    offset += sizeof(req.msgIdx);

90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111
    req.msgType = ntohl(*(int32_t*)((char*)pMsg->pCont + offset));
    offset += sizeof(req.msgType);

    req.msgLen = ntohl(*(int32_t*)((char*)pMsg->pCont + offset));
    offset += sizeof(req.msgLen);

    req.msg = (char*)pMsg->pCont + offset;
    offset += req.msgLen;

    reqMsg.msgType = req.msgType;
    reqMsg.pCont = req.msg;
    reqMsg.contLen = req.msgLen;
    reqMsg.info.rsp = NULL;
    reqMsg.info.rspLen = 0;

    MndMsgFp fp = pMnode->msgFp[TMSG_INDEX(req.msgType)];
    if (fp == NULL) {
      mError("msg:%p, failed to get msg handle, app:%p type:%s", pMsg, pMsg->info.ahandle, TMSG_INFO(pMsg->msgType));
      terrno = TSDB_CODE_MSG_NOT_PROCESSED;
      return -1;
    }

D
dapan1121 已提交
112 113 114 115 116
    if ((*fp)(&reqMsg)) {
      rsp.rspCode = terrno;
    } else {
      rsp.rspCode = 0;
    }
D
dapan1121 已提交
117
    rsp.msgIdx = req.msgIdx;
118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142
    rsp.reqType = reqMsg.msgType;
    rsp.msgLen = reqMsg.info.rspLen;
    rsp.msg = reqMsg.info.rsp;
    
    taosArrayPush(batchRsp, &rsp);

    rspSize += sizeof(rsp) + rsp.msgLen - POINTER_BYTES;
  }

  rspSize += sizeof(int32_t);
  offset = 0;
  
  pRsp = rpcMallocCont(rspSize);
  if (pRsp == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

  *(int32_t*)((char*)pRsp + offset) = htonl(msgNum);
  offset += sizeof(msgNum);
  for (int32_t i = 0; i < msgNum; ++i) {
    SBatchRsp *p = taosArrayGet(batchRsp, i);
    
    *(int32_t*)((char*)pRsp + offset) = htonl(p->reqType);
    offset += sizeof(p->reqType);
D
dapan1121 已提交
143 144
    *(int32_t*)((char*)pRsp + offset) = htonl(p->msgIdx);
    offset += sizeof(p->msgIdx);
145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171
    *(int32_t*)((char*)pRsp + offset) = htonl(p->msgLen);
    offset += sizeof(p->msgLen);
    *(int32_t*)((char*)pRsp + offset) = htonl(p->rspCode);
    offset += sizeof(p->rspCode);
    memcpy((char*)pRsp + offset, p->msg, p->msgLen);
    offset += p->msgLen;

    rpcFreeCont(p->msg);
  }

  taosArrayDestroy(batchRsp);
  batchRsp = NULL;

_exit:

  pMsg->info.rsp = pRsp;
  pMsg->info.rspLen = rspSize;

  if (code) {
    mError("mnd get batch meta failed cause of %s", tstrerror(code));
  }

  taosArrayDestroyEx(batchRsp, tFreeSBatchRsp);

  return code;
}

D
dapan1121 已提交
172
int32_t mndInitQuery(SMnode *pMnode) {
D
dapan1121 已提交
173
  if (qWorkerInit(NODE_TYPE_MNODE, MNODE_HANDLE, (void **)&pMnode->pQuery, &pMnode->msgCb) != 0) {
S
Shengliang Guan 已提交
174 175
    mError("failed to init qworker in mnode since %s", terrstr());
    return -1;
D
dapan1121 已提交
176 177
  }

D
dapan1121 已提交
178
  mndSetMsgHandle(pMnode, TDMT_SCH_QUERY, mndProcessQueryMsg);
D
dapan1121 已提交
179
  mndSetMsgHandle(pMnode, TDMT_SCH_MERGE_QUERY, mndProcessQueryMsg);
D
dapan1121 已提交
180 181
  mndSetMsgHandle(pMnode, TDMT_SCH_QUERY_CONTINUE, mndProcessQueryMsg);
  mndSetMsgHandle(pMnode, TDMT_SCH_FETCH, mndProcessQueryMsg);
D
dapan1121 已提交
182
  mndSetMsgHandle(pMnode, TDMT_SCH_MERGE_FETCH, mndProcessQueryMsg);
D
dapan1121 已提交
183 184
  mndSetMsgHandle(pMnode, TDMT_SCH_DROP_TASK, mndProcessQueryMsg);
  mndSetMsgHandle(pMnode, TDMT_SCH_QUERY_HEARTBEAT, mndProcessQueryMsg);
185
  mndSetMsgHandle(pMnode, TDMT_MND_BATCH_META, mndProcessBatchMetaMsg);
D
dapan1121 已提交
186 187 188 189 190

  return 0;
}

void mndCleanupQuery(SMnode *pMnode) { qWorkerDestroy((void **)&pMnode->pQuery); }