mndShow.c 10.9 KB
Newer Older
H
refact  
Hongze Cheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * 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/>.
 */

S
Shengliang Guan 已提交
16
#define _DEFAULT_SOURCE
S
Shengliang Guan 已提交
17
#include "mndShow.h"
S
Shengliang Guan 已提交
18

S
Shengliang Guan 已提交
19 20
#define SHOW_STEP_SIZE 100

S
Shengliang Guan 已提交
21
static SShowObj *mndCreateShowObj(SMnode *pMnode, SShowReq *pReq);
S
Shengliang Guan 已提交
22
static void      mndFreeShowObj(SShowObj *pShow);
23
static SShowObj *mndAcquireShowObj(SMnode *pMnode, int64_t showId);
S
Shengliang Guan 已提交
24
static void      mndReleaseShowObj(SShowObj *pShow, bool forceRemove);
S
Shengliang Guan 已提交
25 26
static int32_t   mndProcessShowReq(SMnodeMsg *pReq);
static int32_t   mndProcessRetrieveReq(SMnodeMsg *pReq);
S
Shengliang Guan 已提交
27
static bool      mndCheckRetrieveFinished(SShowObj *pShow);
S
Shengliang Guan 已提交
28 29 30 31

int32_t mndInitShow(SMnode *pMnode) {
  SShowMgmt *pMgmt = &pMnode->showMgmt;

S
Shengliang Guan 已提交
32
  pMgmt->cache = taosCacheInit(TSDB_DATA_TYPE_INT, 5, true, (__cache_free_fn_t)mndFreeShowObj, "show");
S
Shengliang Guan 已提交
33 34 35 36 37 38
  if (pMgmt->cache == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    mError("failed to alloc show cache since %s", terrstr());
    return -1;
  }

S
Shengliang Guan 已提交
39 40
  mndSetMsgHandle(pMnode, TDMT_MND_SHOW, mndProcessShowReq);
  mndSetMsgHandle(pMnode, TDMT_MND_SHOW_RETRIEVE, mndProcessRetrieveReq);
S
Shengliang Guan 已提交
41 42 43
  return 0;
}

S
Shengliang Guan 已提交
44 45 46 47 48 49 50 51
void mndCleanupShow(SMnode *pMnode) {
  SShowMgmt *pMgmt = &pMnode->showMgmt;
  if (pMgmt->cache != NULL) {
    taosCacheCleanup(pMgmt->cache);
    pMgmt->cache = NULL;
  }
}

S
Shengliang Guan 已提交
52
static SShowObj *mndCreateShowObj(SMnode *pMnode, SShowReq *pReq) {
S
Shengliang Guan 已提交
53 54
  SShowMgmt *pMgmt = &pMnode->showMgmt;

55 56
  int64_t showId = atomic_add_fetch_64(&pMgmt->showId, 1);
  if (showId == 0) atomic_add_fetch_64(&pMgmt->showId, 1);
S
Shengliang Guan 已提交
57

S
Shengliang Guan 已提交
58
  int32_t  size = sizeof(SShowObj) + pReq->payloadLen;
S
Shengliang Guan 已提交
59 60 61
  SShowObj showObj = {0};
  showObj.id = showId;
  showObj.pMnode = pMnode;
S
Shengliang Guan 已提交
62 63 64 65
  showObj.type = pReq->type;
  showObj.payloadLen = pReq->payloadLen;
  memcpy(showObj.db, pReq->db, TSDB_DB_FNAME_LEN);
  memcpy(showObj.payload, pReq->payload, pReq->payloadLen);
S
Shengliang Guan 已提交
66

S
Shengliang Guan 已提交
67
  int32_t   keepTime = pMnode->cfg.shellActivityTimer * 6 * 1000;
68
  SShowObj *pShow = taosCachePut(pMgmt->cache, &showId, sizeof(int64_t), &showObj, size, keepTime);
S
Shengliang Guan 已提交
69
  if (pShow == NULL) {
S
Shengliang Guan 已提交
70
    terrno = TSDB_CODE_OUT_OF_MEMORY;
S
Shengliang Guan 已提交
71
    mError("show:0x%" PRIx64 ", failed to put into cache since %s", showId, terrstr());
S
Shengliang Guan 已提交
72 73
    return NULL;
  }
S
Shengliang Guan 已提交
74

S
Shengliang Guan 已提交
75
  mTrace("show:0x%" PRIx64 ", is created, data:%p", showId, pShow);
S
Shengliang Guan 已提交
76
  return pShow;
S
Shengliang Guan 已提交
77 78
}

S
Shengliang Guan 已提交
79
static void mndFreeShowObj(SShowObj *pShow) {
S
Shengliang Guan 已提交
80 81 82 83 84 85 86 87 88 89
  SMnode    *pMnode = pShow->pMnode;
  SShowMgmt *pMgmt = &pMnode->showMgmt;

  ShowFreeIterFp freeFp = pMgmt->freeIterFps[pShow->type];
  if (freeFp != NULL) {
    if (pShow->pIter != NULL) {
      (*freeFp)(pMnode, pShow->pIter);
    }
  }

S
Shengliang Guan 已提交
90
  mTrace("show:0x%" PRIx64 ", is destroyed, data:%p", pShow->id, pShow);
S
Shengliang Guan 已提交
91 92
}

93
static SShowObj *mndAcquireShowObj(SMnode *pMnode, int64_t showId) {
S
Shengliang Guan 已提交
94 95
  SShowMgmt *pMgmt = &pMnode->showMgmt;

96
  SShowObj *pShow = taosCacheAcquireByKey(pMgmt->cache, &showId, sizeof(showId));
S
Shengliang Guan 已提交
97
  if (pShow == NULL) {
S
Shengliang Guan 已提交
98
    mError("show:0x%" PRIx64 ", already destroyed", showId);
S
Shengliang Guan 已提交
99 100 101
    return NULL;
  }

S
Shengliang Guan 已提交
102
  mTrace("show:0x%" PRIx64 ", acquired from cache, data:%p", pShow->id, pShow);
S
Shengliang Guan 已提交
103 104 105 106 107
  return pShow;
}

static void mndReleaseShowObj(SShowObj *pShow, bool forceRemove) {
  if (pShow == NULL) return;
S
Shengliang Guan 已提交
108
  mTrace("show:0x%" PRIx64 ", released from cache, data:%p force:%d", pShow->id, pShow, forceRemove);
S
Shengliang Guan 已提交
109

S
Shengliang Guan 已提交
110 111
  // A bug in tcache.c
  forceRemove = 0;
S
Shengliang Guan 已提交
112 113 114 115

  SMnode    *pMnode = pShow->pMnode;
  SShowMgmt *pMgmt = &pMnode->showMgmt;
  taosCacheRelease(pMgmt->cache, (void **)(&pShow), forceRemove);
S
Shengliang Guan 已提交
116
}
S
Shengliang Guan 已提交
117

S
Shengliang Guan 已提交
118 119
static int32_t mndProcessShowReq(SMnodeMsg *pReq) {
  SMnode    *pMnode = pReq->pMnode;
S
Shengliang Guan 已提交
120
  SShowMgmt *pMgmt = &pMnode->showMgmt;
S
Shengliang Guan 已提交
121 122 123
  SShowReq  *pShowReq = pReq->rpcMsg.pCont;
  int8_t     type = pShowReq->type;
  int16_t    payloadLen = htonl(pShowReq->payloadLen);
S
Shengliang Guan 已提交
124 125 126

  if (type <= TSDB_MGMT_TABLE_START || type >= TSDB_MGMT_TABLE_MAX) {
    terrno = TSDB_CODE_MND_INVALID_MSG_TYPE;
S
Shengliang Guan 已提交
127
    mError("failed to process show-meta req since %s", terrstr());
S
Shengliang Guan 已提交
128 129 130 131 132 133
    return -1;
  }

  ShowMetaFp metaFp = pMgmt->metaFps[type];
  if (metaFp == NULL) {
    terrno = TSDB_CODE_MND_INVALID_MSG_TYPE;
S
Shengliang Guan 已提交
134
    mError("failed to process show-meta req:%s since %s", mndShowStr(type), terrstr());
S
Shengliang Guan 已提交
135 136 137
    return -1;
  }

S
Shengliang Guan 已提交
138
  SShowObj *pShow = mndCreateShowObj(pMnode, pShowReq);
S
Shengliang Guan 已提交
139
  if (pShow == NULL) {
S
Shengliang Guan 已提交
140
    mError("failed to process show-meta req:%s since %s", mndShowStr(type), terrstr());
S
Shengliang Guan 已提交
141 142 143
    return -1;
  }

S
Shengliang Guan 已提交
144
  int32_t   size = sizeof(SShowRsp) + sizeof(SSchema) * TSDB_MAX_COLUMNS + TSDB_EXTRA_PAYLOAD_SIZE;
S
Shengliang Guan 已提交
145 146 147 148
  SShowRsp *pRsp = rpcMallocCont(size);
  if (pRsp == NULL) {
    mndReleaseShowObj(pShow, true);
    terrno = TSDB_CODE_OUT_OF_MEMORY;
S
Shengliang Guan 已提交
149
    mError("show:0x%" PRIx64 ", failed to process show-meta req:%s since malloc rsp error", pShow->id,
S
Shengliang Guan 已提交
150
           mndShowStr(type));
S
Shengliang Guan 已提交
151 152 153
    return -1;
  }

S
Shengliang Guan 已提交
154
  int32_t code = (*metaFp)(pReq, pShow, &pRsp->tableMeta);
S
Shengliang Guan 已提交
155
  mDebug("show:0x%" PRIx64 ", get meta finished, numOfRows:%d cols:%d type:%s, result:%s", pShow->id, pShow->numOfRows,
S
Shengliang Guan 已提交
156
         pShow->numOfColumns, mndShowStr(type), tstrerror(code));
S
Shengliang Guan 已提交
157 158

  if (code == TSDB_CODE_SUCCESS) {
S
Shengliang Guan 已提交
159 160
    pReq->contLen = sizeof(SShowRsp) + sizeof(SSchema) * pShow->numOfColumns;
    pReq->pCont = pRsp;
S
Shengliang Guan 已提交
161
    pRsp->showId = htobe64(pShow->id);
S
Shengliang Guan 已提交
162 163 164 165 166 167 168 169 170
    mndReleaseShowObj(pShow, false);
    return TSDB_CODE_SUCCESS;
  } else {
    rpcFreeCont(pRsp);
    mndReleaseShowObj(pShow, true);
    return code;
  }
}

S
Shengliang Guan 已提交
171 172
static int32_t mndProcessRetrieveReq(SMnodeMsg *pReq) {
  SMnode    *pMnode = pReq->pMnode;
S
Shengliang Guan 已提交
173 174 175 176 177
  SShowMgmt *pMgmt = &pMnode->showMgmt;
  int32_t    rowsToRead = 0;
  int32_t    size = 0;
  int32_t    rowsRead = 0;

S
Shengliang Guan 已提交
178
  SRetrieveTableReq *pRetrieve = pReq->rpcMsg.pCont;
S
Shengliang Guan 已提交
179
  int64_t            showId = htobe64(pRetrieve->showId);
S
Shengliang Guan 已提交
180

S
Shengliang Guan 已提交
181 182
  SShowObj *pShow = mndAcquireShowObj(pMnode, showId);
  if (pShow == NULL) {
S
Shengliang Guan 已提交
183
    terrno = TSDB_CODE_MND_INVALID_SHOWOBJ;
S
Shengliang Guan 已提交
184
    mError("failed to process show-retrieve req:%p since %s", pShow, terrstr());
S
Shengliang Guan 已提交
185 186 187 188 189 190 191
    return -1;
  }

  ShowRetrieveFp retrieveFp = pMgmt->retrieveFps[pShow->type];
  if (retrieveFp == NULL) {
    mndReleaseShowObj(pShow, false);
    terrno = TSDB_CODE_MSG_NOT_PROCESSED;
S
Shengliang Guan 已提交
192
    mError("show:0x%" PRIx64 ", failed to retrieve data since %s", pShow->id, terrstr());
S
Shengliang Guan 已提交
193 194 195
    return -1;
  }

S
Shengliang Guan 已提交
196
  mDebug("show:0x%" PRIx64 ", start retrieve data, numOfReads:%d numOfRows:%d type:%s", pShow->id, pShow->numOfReads,
S
Shengliang Guan 已提交
197
         pShow->numOfRows, mndShowStr(pShow->type));
S
Shengliang Guan 已提交
198 199

  if (mndCheckRetrieveFinished(pShow)) {
S
Shengliang Guan 已提交
200 201
    mDebug("show:0x%" PRIx64 ", read finished, numOfReads:%d numOfRows:%d", pShow->id, pShow->numOfReads,
           pShow->numOfRows);
S
Shengliang Guan 已提交
202 203 204 205 206 207 208 209
    pShow->numOfReads = pShow->numOfRows;
  }

  if ((pRetrieve->free & TSDB_QUERY_TYPE_FREE_RESOURCE) != TSDB_QUERY_TYPE_FREE_RESOURCE) {
    rowsToRead = pShow->numOfRows - pShow->numOfReads;
  }

  /* return no more than 100 tables in one round trip */
S
Shengliang Guan 已提交
210
  if (rowsToRead > SHOW_STEP_SIZE) rowsToRead = SHOW_STEP_SIZE;
S
Shengliang Guan 已提交
211 212 213 214 215 216 217 218

  /*
   * the actual number of table may be larger than the value of pShow->numOfRows, if a query is
   * issued during a continuous create table operation. Therefore, rowToRead may be less than 0.
   */
  if (rowsToRead < 0) rowsToRead = 0;
  size = pShow->rowSize * rowsToRead;

S
Shengliang Guan 已提交
219
  size += SHOW_STEP_SIZE;
S
Shengliang Guan 已提交
220 221 222 223
  SRetrieveTableRsp *pRsp = rpcMallocCont(size);
  if (pRsp == NULL) {
    mndReleaseShowObj(pShow, false);
    terrno = TSDB_CODE_OUT_OF_MEMORY;
S
Shengliang Guan 已提交
224
    mError("show:0x%" PRIx64 ", failed to retrieve data since %s", pShow->id, terrstr());
S
Shengliang Guan 已提交
225 226 227 228 229
    return -1;
  }

  // if free flag is set, client wants to clean the resources
  if ((pRetrieve->free & TSDB_QUERY_TYPE_FREE_RESOURCE) != TSDB_QUERY_TYPE_FREE_RESOURCE) {
S
Shengliang Guan 已提交
230
    rowsRead = (*retrieveFp)(pReq, pShow, pRsp->data, rowsToRead);
S
Shengliang Guan 已提交
231 232
  }

S
Shengliang Guan 已提交
233
  mDebug("show:0x%" PRIx64 ", stop retrieve data, rowsRead:%d rowsToRead:%d", pShow->id, rowsRead, rowsToRead);
S
Shengliang Guan 已提交
234 235

  pRsp->numOfRows = htonl(rowsRead);
S
Shengliang Guan 已提交
236
  pRsp->precision = TSDB_TIME_PRECISION_MILLI;  // millisecond time precision
S
Shengliang Guan 已提交
237

S
Shengliang Guan 已提交
238 239
  pReq->pCont = pRsp;
  pReq->contLen = size;
S
Shengliang Guan 已提交
240

S
Shengliang Guan 已提交
241
  if (rowsRead == 0 || rowsToRead == 0 || (rowsRead == rowsToRead && pShow->numOfRows == pShow->numOfReads)) {
S
Shengliang Guan 已提交
242
    pRsp->completed = 1;
S
Shengliang Guan 已提交
243
    mDebug("show:0x%" PRIx64 ", retrieve completed", pShow->id);
S
Shengliang Guan 已提交
244 245
    mndReleaseShowObj(pShow, true);
  } else {
S
Shengliang Guan 已提交
246
    mDebug("show:0x%" PRIx64 ", retrieve not completed yet", pShow->id);
S
Shengliang Guan 已提交
247 248 249 250 251 252
    mndReleaseShowObj(pShow, false);
  }

  return TSDB_CODE_SUCCESS;
}

S
Shengliang Guan 已提交
253
char *mndShowStr(int32_t showType) {
S
Shengliang Guan 已提交
254 255 256 257 258 259 260 261 262 263 264 265 266
  switch (showType) {
    case TSDB_MGMT_TABLE_ACCT:
      return "show accounts";
    case TSDB_MGMT_TABLE_USER:
      return "show users";
    case TSDB_MGMT_TABLE_DB:
      return "show databases";
    case TSDB_MGMT_TABLE_TABLE:
      return "show tables";
    case TSDB_MGMT_TABLE_DNODE:
      return "show dnodes";
    case TSDB_MGMT_TABLE_MNODE:
      return "show mnodes";
S
Shengliang Guan 已提交
267 268 269 270 271 272
    case TSDB_MGMT_TABLE_QNODE:
      return "show qnodes";
    case TSDB_MGMT_TABLE_SNODE:
      return "show snodes";
    case TSDB_MGMT_TABLE_BNODE:
      return "show bnodes";
S
Shengliang Guan 已提交
273 274
    case TSDB_MGMT_TABLE_VGROUP:
      return "show vgroups";
S
Shengliang Guan 已提交
275
    case TSDB_MGMT_TABLE_STB:
S
Shengliang Guan 已提交
276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293
      return "show stables";
    case TSDB_MGMT_TABLE_MODULE:
      return "show modules";
    case TSDB_MGMT_TABLE_QUERIES:
      return "show queries";
    case TSDB_MGMT_TABLE_STREAMS:
      return "show streams";
    case TSDB_MGMT_TABLE_VARIABLES:
      return "show configs";
    case TSDB_MGMT_TABLE_CONNS:
      return "show connections";
    case TSDB_MGMT_TABLE_SCORES:
      return "show scores";
    case TSDB_MGMT_TABLE_GRANTS:
      return "show grants";
    case TSDB_MGMT_TABLE_VNODES:
      return "show vnodes";
    case TSDB_MGMT_TABLE_CLUSTER:
S
Shengliang Guan 已提交
294
      return "show cluster";
S
Shengliang Guan 已提交
295 296 297 298
    case TSDB_MGMT_TABLE_STREAMTABLES:
      return "show streamtables";
    case TSDB_MGMT_TABLE_TP:
      return "show topics";
S
Shengliang 已提交
299
    case TSDB_MGMT_TABLE_FUNC:
S
Shengliang Guan 已提交
300
      return "show functions";
S
Shengliang Guan 已提交
301 302 303 304 305 306 307 308
    default:
      return "undefined";
  }
}

static bool mndCheckRetrieveFinished(SShowObj *pShow) {
  if (pShow->pIter == NULL && pShow->numOfReads != 0) {
    return true;
S
Shengliang Guan 已提交
309
  }
S
Shengliang Guan 已提交
310 311 312
  return false;
}

S
Shengliang Guan 已提交
313
void mndVacuumResult(char *data, int32_t numOfCols, int32_t rows, int32_t capacity, SShowObj *pShow) {
S
Shengliang Guan 已提交
314 315 316 317 318 319 320
  if (rows < capacity) {
    for (int32_t i = 0; i < numOfCols; ++i) {
      memmove(data + pShow->offset[i] * rows, data + pShow->offset[i] * capacity, pShow->bytes[i] * rows);
    }
  }
}

S
Shengliang Guan 已提交
321
void mndAddShowMetaHandle(SMnode *pMnode, EShowType showType, ShowMetaFp fp) {
S
Shengliang Guan 已提交
322 323 324 325
  SShowMgmt *pMgmt = &pMnode->showMgmt;
  pMgmt->metaFps[showType] = fp;
}

S
Shengliang Guan 已提交
326
void mndAddShowRetrieveHandle(SMnode *pMnode, EShowType showType, ShowRetrieveFp fp) {
S
Shengliang Guan 已提交
327 328 329 330
  SShowMgmt *pMgmt = &pMnode->showMgmt;
  pMgmt->retrieveFps[showType] = fp;
}

S
Shengliang Guan 已提交
331
void mndAddShowFreeIterHandle(SMnode *pMnode, EShowType showType, ShowFreeIterFp fp) {
S
Shengliang Guan 已提交
332 333 334
  SShowMgmt *pMgmt = &pMnode->showMgmt;
  pMgmt->freeIterFps[showType] = fp;
}