clientHb.c 36.7 KB
Newer Older
L
Liu Jicong 已提交
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/>.
 */

D
dapan1121 已提交
16
#include "catalog.h"
L
fix  
Liu Jicong 已提交
17
#include "clientInt.h"
D
dapan1121 已提交
18
#include "clientLog.h"
L
Liu Jicong 已提交
19
#include "scheduler.h"
L
fix  
Liu Jicong 已提交
20
#include "trpc.h"
K
kailixu 已提交
21
#include "tsimplehash.h"
L
Liu Jicong 已提交
22

K
kailixu 已提交
23 24
typedef struct {
  union {
K
kailixu 已提交
25
    struct {
K
kailixu 已提交
26 27 28 29
      SAppHbMgr *pAppHbMgr;
      int64_t    clusterId;
      int32_t    reqCnt;
      int8_t     connHbFlag;
K
kailixu 已提交
30
    };
K
kailixu 已提交
31 32 33
  };
} SHbParam;

K
kailixu 已提交
34 35 36 37 38
static SClientHbMgr clientHbMgr = {0};

static int32_t hbCreateThread();
static void    hbStopThread();

K
kailixu 已提交
39
static int32_t hbMqHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq *req) { return 0; }
L
Liu Jicong 已提交
40

L
Liu Jicong 已提交
41
static int32_t hbMqHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) { return 0; }
L
Liu Jicong 已提交
42

K
kailixu 已提交
43
static int32_t hbUpdateUserAuthInfo(SAppHbMgr *pAppHbMgr, SUserAuthBatchRsp *batchRsp) {
K
kailixu 已提交
44 45 46 47 48 49
  SClientHbReq *pReq = NULL;
  while ((pReq = taosHashIterate(pAppHbMgr->activeInfo, pReq))) {
    STscObj *pTscObj = (STscObj *)acquireTscObj(pReq->connKey.tscRid);
    if (!pTscObj) {
      continue;
    }
K
kailixu 已提交
50 51
    for (int32_t i = 0; i < TARRAY_SIZE(batchRsp->pArray); ++i) {
      SGetUserAuthRsp *rsp = taosArrayGet(batchRsp->pArray, i);
K
kailixu 已提交
52

K
kailixu 已提交
53
      if (0 == strncmp(rsp->user, pTscObj->user, TSDB_USER_LEN)) {
K
kailixu 已提交
54 55 56
        
        pTscObj->authVer = rsp->version;

K
kailixu 已提交
57
#if 0 // make jenkins happy temporarily. After PR pass, enable these lines again.
K
kailixu 已提交
58
        if (pTscObj->sysInfo != rsp->sysInfo) {
K
kailixu 已提交
59 60
          tscDebug("update sysInfo of user %s from %" PRIi8 " to %" PRIi8 ", tscRid:%" PRIi64, rsp->user,
                   pTscObj->sysInfo, rsp->sysInfo, pTscObj->id);
K
kailixu 已提交
61 62
          pTscObj->sysInfo = rsp->sysInfo;
        }
K
kailixu 已提交
63
#endif
K
kailixu 已提交
64 65 66
        if (pTscObj->passInfo.fp) {
          SPassInfo *passInfo = &pTscObj->passInfo;
          int32_t    oldVer = atomic_load_32(&passInfo->ver);
K
kailixu 已提交
67 68 69
          if (oldVer < rsp->passVer) {
            atomic_store_32(&passInfo->ver, rsp->passVer);
            if (passInfo->fp) {
K
kailixu 已提交
70
              (*passInfo->fp)(passInfo->param, &rsp->passVer, TAOS_NOTIFY_PASSVER);
K
kailixu 已提交
71
            }
K
kailixu 已提交
72 73
            tscDebug("update passVer of user %s from %d to %d, tscRid:%" PRIi64, rsp->user, oldVer,
                     atomic_load_32(&passInfo->ver), pTscObj->id);
K
kailixu 已提交
74
          }
K
kailixu 已提交
75
        }
K
kailixu 已提交
76

K
kailixu 已提交
77
        break;
K
kailixu 已提交
78 79
      }
    }
K
kailixu 已提交
80
    releaseTscObj(pReq->connKey.tscRid);
K
kailixu 已提交
81
  }
K
kailixu 已提交
82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105
  return 0;
}

static int32_t hbProcessUserAuthInfoRsp(void *value, int32_t valueLen, struct SCatalog *pCatalog,
                                        SAppHbMgr *pAppHbMgr) {
  int32_t code = 0;

  SUserAuthBatchRsp batchRsp = {0};
  if (tDeserializeSUserAuthBatchRsp(value, valueLen, &batchRsp) != 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

  int32_t numOfBatchs = taosArrayGetSize(batchRsp.pArray);
  for (int32_t i = 0; i < numOfBatchs; ++i) {
    SGetUserAuthRsp *rsp = taosArrayGet(batchRsp.pArray, i);
    tscDebug("hb user auth rsp, user:%s, version:%d", rsp->user, rsp->version);

    catalogUpdateUserAuthInfo(pCatalog, rsp);
  }

  if (numOfBatchs > 0) hbUpdateUserAuthInfo(pAppHbMgr, &batchRsp);

  atomic_val_compare_exchange_8(&pAppHbMgr->connHbFlag, 1, 2);
K
kailixu 已提交
106 107

  taosArrayDestroy(batchRsp.pArray);
K
kailixu 已提交
108
  return TSDB_CODE_SUCCESS;
K
kailixu 已提交
109 110
}

D
dapan1121 已提交
111
static int32_t hbGenerateVgInfoFromRsp(SDBVgInfo **pInfo, SUseDbRsp *rsp) {
112
  int32_t    code = 0;
D
dapan1121 已提交
113 114 115 116 117
  SDBVgInfo *vgInfo = taosMemoryCalloc(1, sizeof(SDBVgInfo));
  if (NULL == vgInfo) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    return code;
  }
118

D
dapan1121 已提交
119 120 121 122 123 124 125 126 127 128 129 130
  vgInfo->vgVersion = rsp->vgVersion;
  vgInfo->stateTs = rsp->stateTs;
  vgInfo->hashMethod = rsp->hashMethod;
  vgInfo->hashPrefix = rsp->hashPrefix;
  vgInfo->hashSuffix = rsp->hashSuffix;
  vgInfo->vgHash = taosHashInit(rsp->vgNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
  if (NULL == vgInfo->vgHash) {
    taosMemoryFree(vgInfo);
    tscError("hash init[%d] failed", rsp->vgNum);
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _return;
  }
131

D
dapan1121 已提交
132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152
  for (int32_t j = 0; j < rsp->vgNum; ++j) {
    SVgroupInfo *pInfo = taosArrayGet(rsp->pVgroupInfos, j);
    if (taosHashPut(vgInfo->vgHash, &pInfo->vgId, sizeof(int32_t), pInfo, sizeof(SVgroupInfo)) != 0) {
      tscError("hash push failed, errno:%d", errno);
      taosHashCleanup(vgInfo->vgHash);
      taosMemoryFree(vgInfo);
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _return;
    }
  }

_return:
  if (code) {
    taosHashCleanup(vgInfo->vgHash);
    taosMemoryFreeClear(vgInfo);
  }

  *pInfo = vgInfo;
  return code;
}

D
dapan1121 已提交
153 154 155
static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog *pCatalog) {
  int32_t code = 0;

D
dapan1121 已提交
156 157
  SDbHbBatchRsp batchRsp = {0};
  if (tDeserializeSDbHbBatchRsp(value, valueLen, &batchRsp) != 0) {
S
Shengliang Guan 已提交
158
    terrno = TSDB_CODE_INVALID_MSG;
159 160
    code = terrno;
    goto _return;
S
Shengliang Guan 已提交
161
  }
D
dapan1121 已提交
162

D
dapan1121 已提交
163
  int32_t numOfBatchs = taosArrayGetSize(batchRsp.pArray);
S
Shengliang Guan 已提交
164
  for (int32_t i = 0; i < numOfBatchs; ++i) {
D
dapan1121 已提交
165 166
    SDbHbRsp *rsp = taosArrayGet(batchRsp.pArray, i);
    if (rsp->useDbRsp) {
H
Haojun Liao 已提交
167
      tscDebug("hb use db rsp, db:%s, vgVersion:%d, stateTs:%" PRId64 ", uid:%" PRIx64,
D
dapan1121 已提交
168
        rsp->useDbRsp->db, rsp->useDbRsp->vgVersion, rsp->useDbRsp->stateTs, rsp->useDbRsp->uid);
H
Haojun Liao 已提交
169

D
dapan1121 已提交
170 171 172 173 174
      if (rsp->useDbRsp->vgVersion < 0) {
        code = catalogRemoveDB(pCatalog, rsp->useDbRsp->db, rsp->useDbRsp->uid);
      } else {
        SDBVgInfo *vgInfo = NULL;
        code = hbGenerateVgInfoFromRsp(&vgInfo, rsp->useDbRsp);
D
dapan1121 已提交
175
        if (TSDB_CODE_SUCCESS != code) {
D
dapan1121 已提交
176
          goto _return;
D
dapan1121 已提交
177
        }
L
fix  
Liu Jicong 已提交
178

D
dapan1121 已提交
179
        catalogUpdateDBVgInfo(pCatalog, rsp->useDbRsp->db, rsp->useDbRsp->uid, vgInfo);
H
Haojun Liao 已提交
180

D
dapan1121 已提交
181 182 183 184 185
        if (IS_SYS_DBNAME(rsp->useDbRsp->db)) {
          code = hbGenerateVgInfoFromRsp(&vgInfo, rsp->useDbRsp);
          if (TSDB_CODE_SUCCESS != code) {
            goto _return;
          }
H
Haojun Liao 已提交
186

D
dapan1121 已提交
187 188
          catalogUpdateDBVgInfo(pCatalog, (rsp->useDbRsp->db[0] == 'i') ? TSDB_PERFORMANCE_SCHEMA_DB : TSDB_INFORMATION_SCHEMA_DB, rsp->useDbRsp->uid, vgInfo);
        }
D
dapan1121 已提交
189
      }
D
dapan1121 已提交
190 191
    }

D
dapan1121 已提交
192 193 194 195
    if (rsp->cfgRsp) {
      tscDebug("hb db cfg rsp, db:%s, cfgVersion:%d", rsp->cfgRsp->db, rsp->cfgRsp->cfgVersion);
      catalogUpdateDbCfg(pCatalog, rsp->cfgRsp->db, rsp->cfgRsp->dbId, rsp->cfgRsp);
      rsp->cfgRsp = NULL;
D
dapan1121 已提交
196 197 198
    }
  }

D
dapan1121 已提交
199 200
_return:

D
dapan1121 已提交
201
  tFreeSDbHbBatchRsp(&batchRsp);
D
dapan1121 已提交
202
  return code;
D
dapan1121 已提交
203 204
}

D
dapan 已提交
205 206 207
static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalog *pCatalog) {
  int32_t code = 0;

D
dapan1121 已提交
208 209
  SSTbHbRsp hbRsp = {0};
  if (tDeserializeSSTbHbRsp(value, valueLen, &hbRsp) != 0) {
S
Shengliang Guan 已提交
210 211 212 213
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

D
dapan1121 已提交
214 215 216
  int32_t numOfMeta = taosArrayGetSize(hbRsp.pMetaRsp);
  for (int32_t i = 0; i < numOfMeta; ++i) {
    STableMetaRsp *rsp = taosArrayGet(hbRsp.pMetaRsp, i);
S
Shengliang Guan 已提交
217

D
dapan 已提交
218 219
    if (rsp->numOfColumns < 0) {
      tscDebug("hb remove stb, db:%s, stb:%s", rsp->dbFName, rsp->stbName);
D
dapan1121 已提交
220
      catalogRemoveStbMeta(pCatalog, rsp->dbFName, rsp->dbId, rsp->stbName, rsp->suid);
D
dapan 已提交
221
    } else {
D
dapan1121 已提交
222
      tscDebug("hb update stb, db:%s, stb:%s", rsp->dbFName, rsp->stbName);
S
Shengliang Guan 已提交
223
      if (rsp->pSchemas[0].colId != PRIMARYKEY_TIMESTAMP_COL_ID) {
224
        tscError("invalid colId[%" PRIi16 "] for the first column in table meta rsp msg", rsp->pSchemas[0].colId);
D
dapan1121 已提交
225
        tFreeSSTbHbRsp(&hbRsp);
D
dapan1121 已提交
226
        return TSDB_CODE_TSC_INVALID_VALUE;
D
dapan 已提交
227 228
      }

D
dapan1121 已提交
229
      catalogAsyncUpdateTableMeta(pCatalog, rsp);
D
dapan 已提交
230 231 232
    }
  }

D
dapan1121 已提交
233 234 235 236 237 238 239 240 241 242
  int32_t numOfIndex = taosArrayGetSize(hbRsp.pIndexRsp);
  for (int32_t i = 0; i < numOfIndex; ++i) {
    STableIndexRsp *rsp = taosArrayGet(hbRsp.pIndexRsp, i);

    catalogUpdateTableIndex(pCatalog, rsp);
  }

  taosArrayDestroy(hbRsp.pIndexRsp);
  hbRsp.pIndexRsp = NULL;

D
dapan1121 已提交
243
  tFreeSSTbHbRsp(&hbRsp);
D
dapan 已提交
244 245 246
  return TSDB_CODE_SUCCESS;
}

L
Liu Jicong 已提交
247
static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
D
dapan1121 已提交
248
  SClientHbReq *pReq = taosHashAcquire(pAppHbMgr->activeInfo, &pRsp->connKey, sizeof(SClientHbKey));
D
dapan1121 已提交
249 250
  if (NULL == pReq) {
    tscWarn("pReq to get activeInfo, may be dropped, refId:%" PRIx64 ", type:%d", pRsp->connKey.tscRid,
L
Liu Jicong 已提交
251
            pRsp->connKey.connType);
D
dapan1121 已提交
252 253 254
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
255 256 257 258
  if (pRsp->query) {
    STscObj *pTscObj = (STscObj *)acquireTscObj(pRsp->connKey.tscRid);
    if (NULL == pTscObj) {
      tscDebug("tscObj rid %" PRIx64 " not exist", pRsp->connKey.tscRid);
259
    } else {
D
dapan1121 已提交
260
      if (pRsp->query->totalDnodes > 1 && !isEpsetEqual(&pTscObj->pAppInfo->mgmtEp.epSet, &pRsp->query->epSet)) {
261 262 263 264 265 266 267
        SEpSet *pOrig = &pTscObj->pAppInfo->mgmtEp.epSet;
        SEp    *pOrigEp = &pOrig->eps[pOrig->inUse];
        SEp    *pNewEp = &pRsp->query->epSet.eps[pRsp->query->epSet.inUse];
        tscDebug("mnode epset updated from %d/%d=>%s:%d to %d/%d=>%s:%d in hb", pOrig->inUse, pOrig->numOfEps,
                 pOrigEp->fqdn, pOrigEp->port, pRsp->query->epSet.inUse, pRsp->query->epSet.numOfEps, pNewEp->fqdn,
                 pNewEp->port);

D
dapan1121 已提交
268 269
        updateEpSet_s(&pTscObj->pAppInfo->mgmtEp, &pRsp->query->epSet);
      }
270

D
dapan1121 已提交
271 272
      pTscObj->pAppInfo->totalDnodes = pRsp->query->totalDnodes;
      pTscObj->pAppInfo->onlineDnodes = pRsp->query->onlineDnodes;
D
dapan1121 已提交
273
      pTscObj->connId = pRsp->query->connId;
wmmhello's avatar
wmmhello 已提交
274
      tscTrace("conn %u hb rsp, dnodes %d/%d", pTscObj->connId, pTscObj->pAppInfo->onlineDnodes,
dengyihao's avatar
dengyihao 已提交
275
               pTscObj->pAppInfo->totalDnodes);
L
Liu Jicong 已提交
276

D
dapan1121 已提交
277
      if (pRsp->query->killRid) {
D
dapan1121 已提交
278
        tscDebug("request rid %" PRIx64 " need to be killed now", pRsp->query->killRid);
D
dapan1121 已提交
279 280 281 282 283 284 285 286
        SRequestObj *pRequest = acquireRequest(pRsp->query->killRid);
        if (NULL == pRequest) {
          tscDebug("request 0x%" PRIx64 " not exist to kill", pRsp->query->killRid);
        } else {
          taos_stop_query((TAOS_RES *)pRequest);
          releaseRequest(pRsp->query->killRid);
        }
      }
L
Liu Jicong 已提交
287

D
dapan1121 已提交
288
      if (pRsp->query->killConnection) {
289
        taos_close_internal(pTscObj);
D
dapan1121 已提交
290 291
      }

D
dapan1121 已提交
292 293 294 295
      if (pRsp->query->pQnodeList) {
        updateQnodeList(pTscObj->pAppInfo, pRsp->query->pQnodeList);
      }

D
dapan1121 已提交
296 297 298
      releaseTscObj(pRsp->connKey.tscRid);
    }
  }
L
Liu Jicong 已提交
299

D
dapan1121 已提交
300
  int32_t kvNum = pRsp->info ? taosArrayGetSize(pRsp->info) : 0;
D
dapan1121 已提交
301 302

  tscDebug("hb got %d rsp kv", kvNum);
L
fix  
Liu Jicong 已提交
303

D
dapan1121 已提交
304 305 306
  for (int32_t i = 0; i < kvNum; ++i) {
    SKv *kv = taosArrayGet(pRsp->info, i);
    switch (kv->key) {
D
dapan 已提交
307 308 309 310 311 312 313 314
      case HEARTBEAT_KEY_USER_AUTHINFO: {
        if (kv->valueLen <= 0 || NULL == kv->value) {
          tscError("invalid hb user auth info, len:%d, value:%p", kv->valueLen, kv->value);
          break;
        }

        struct SCatalog *pCatalog = NULL;

D
dapan1121 已提交
315
        int32_t code = catalogGetHandle(pReq->clusterId, &pCatalog);
D
dapan 已提交
316
        if (code != TSDB_CODE_SUCCESS) {
D
dapan1121 已提交
317
          tscWarn("catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", pReq->clusterId, tstrerror(code));
D
dapan 已提交
318 319 320
          break;
        }

K
kailixu 已提交
321
        hbProcessUserAuthInfoRsp(kv->value, kv->valueLen, pCatalog, pAppHbMgr);
D
dapan 已提交
322 323
        break;
      }
D
dapan1121 已提交
324 325 326 327 328 329 330
      case HEARTBEAT_KEY_DBINFO: {
        if (kv->valueLen <= 0 || NULL == kv->value) {
          tscError("invalid hb db info, len:%d, value:%p", kv->valueLen, kv->value);
          break;
        }

        struct SCatalog *pCatalog = NULL;
L
fix  
Liu Jicong 已提交
331

D
dapan1121 已提交
332
        int32_t code = catalogGetHandle(pReq->clusterId, &pCatalog);
D
dapan1121 已提交
333
        if (code != TSDB_CODE_SUCCESS) {
D
dapan1121 已提交
334
          tscWarn("catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", pReq->clusterId, tstrerror(code));
D
dapan1121 已提交
335 336 337 338
          break;
        }

        hbProcessDBInfoRsp(kv->value, kv->valueLen, pCatalog);
D
dapan1121 已提交
339
        break;
D
dapan1121 已提交
340
      }
L
fix  
Liu Jicong 已提交
341
      case HEARTBEAT_KEY_STBINFO: {
D
dapan 已提交
342 343 344 345
        if (kv->valueLen <= 0 || NULL == kv->value) {
          tscError("invalid hb stb info, len:%d, value:%p", kv->valueLen, kv->value);
          break;
        }
D
dapan1121 已提交
346

D
dapan 已提交
347
        struct SCatalog *pCatalog = NULL;
L
fix  
Liu Jicong 已提交
348

D
dapan1121 已提交
349
        int32_t code = catalogGetHandle(pReq->clusterId, &pCatalog);
D
dapan 已提交
350
        if (code != TSDB_CODE_SUCCESS) {
D
dapan1121 已提交
351
          tscWarn("catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", pReq->clusterId, tstrerror(code));
D
dapan 已提交
352 353 354 355
          break;
        }

        hbProcessStbInfoRsp(kv->value, kv->valueLen, pCatalog);
D
dapan1121 已提交
356
        break;
D
dapan 已提交
357
      }
D
dapan1121 已提交
358 359 360 361 362 363
      default:
        tscError("invalid hb key type:%d", kv->key);
        break;
    }
  }

D
dapan1121 已提交
364 365
  taosHashRelease(pAppHbMgr->activeInfo, pReq);

D
dapan1121 已提交
366 367 368
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
369
static int32_t hbAsyncCallBack(void *param, SDataBuf *pMsg, int32_t code) {
D
dapan1121 已提交
370 371 372 373
  if (0 == atomic_load_8(&clientHbMgr.inited)) {
    goto _return;
  }

374
  static int32_t    emptyRspNum = 0;
375
  int32_t           idx = *(int32_t *)param;
D
dapan1121 已提交
376
  SClientHbBatchRsp pRsp = {0};
D
dapan1121 已提交
377 378
  if (TSDB_CODE_SUCCESS == code) {
    tDeserializeSClientHbBatchRsp(pMsg->pData, pMsg->len, &pRsp);
379

dengyihao's avatar
dengyihao 已提交
380 381 382 383 384 385
    int32_t now = taosGetTimestampSec();
    int32_t delta = abs(now - pRsp.svrTimestamp);
    if (delta > timestampDeltaLimit) {
      code = TSDB_CODE_TIME_UNSYNCED;
      tscError("time diff: %ds is too big", delta);
    }
dengyihao's avatar
dengyihao 已提交
386 387
  }

D
dapan1121 已提交
388
  int32_t rspNum = taosArrayGetSize(pRsp.rsps);
D
dapan1121 已提交
389

D
dapan1121 已提交
390
  taosThreadMutexLock(&clientHbMgr.lock);
D
dapan1121 已提交
391

D
dapan1121 已提交
392 393 394 395
  SAppHbMgr *pAppHbMgr = taosArrayGetP(clientHbMgr.appHbMgrs, idx);
  if (pAppHbMgr == NULL) {
    taosThreadMutexUnlock(&clientHbMgr.lock);
    tscError("appHbMgr not exist, idx:%d", idx);
D
dapan1121 已提交
396
    taosMemoryFree(pMsg->pData);
dengyihao's avatar
dengyihao 已提交
397
    taosMemoryFree(pMsg->pEpSet);
D
dapan1121 已提交
398
    tFreeClientHbBatchRsp(&pRsp);
D
dapan1121 已提交
399 400 401
    return -1;
  }

D
dapan1121 已提交
402 403
  SAppInstInfo *pInst = pAppHbMgr->pAppInstInfo;

D
dapan1121 已提交
404
  if (code != 0) {
D
dapan1121 已提交
405
    pInst->onlineDnodes = pInst->totalDnodes ? 0 : -1;
406
    tscDebug("hb rsp error %s, update server status %d/%d", tstrerror(code), pInst->onlineDnodes, pInst->totalDnodes);
D
dapan1121 已提交
407 408
  }

D
dapan1121 已提交
409
  if (rspNum) {
L
fix  
Liu Jicong 已提交
410 411
    tscDebug("hb got %d rsp, %d empty rsp received before", rspNum,
             atomic_val_compare_exchange_32(&emptyRspNum, emptyRspNum, 0));
D
dapan1121 已提交
412 413 414 415 416
  } else {
    atomic_add_fetch_32(&emptyRspNum, 1);
  }

  for (int32_t i = 0; i < rspNum; ++i) {
L
fix  
Liu Jicong 已提交
417
    SClientHbRsp *rsp = taosArrayGet(pRsp.rsps, i);
D
dapan1121 已提交
418
    code = (*clientHbMgr.rspHandle[rsp->connKey.connType])(pAppHbMgr, rsp);
D
dapan1121 已提交
419 420 421 422
    if (code) {
      break;
    }
  }
D
dapan1121 已提交
423

D
dapan1121 已提交
424
  taosThreadMutexUnlock(&clientHbMgr.lock);
D
dapan1121 已提交
425

D
dapan1121 已提交
426
  tFreeClientHbBatchRsp(&pRsp);
D
dapan1121 已提交
427 428

_return:
H
Haojun Liao 已提交
429
  taosMemoryFree(pMsg->pData);
dengyihao's avatar
dengyihao 已提交
430
  taosMemoryFree(pMsg->pEpSet);
D
dapan1121 已提交
431
  return code;
L
Liu Jicong 已提交
432 433
}

D
dapan1121 已提交
434
int32_t hbBuildQueryDesc(SQueryHbReqBasic *hbBasic, STscObj *pObj) {
L
Liu Jicong 已提交
435
  int64_t    now = taosGetTimestampUs();
D
dapan1121 已提交
436
  SQueryDesc desc = {0};
L
Liu Jicong 已提交
437
  int32_t    code = 0;
D
dapan1121 已提交
438

L
Liu Jicong 已提交
439
  void *pIter = taosHashIterate(pObj->pRequests, NULL);
D
dapan1121 已提交
440
  while (pIter != NULL) {
L
Liu Jicong 已提交
441
    int64_t     *rid = pIter;
D
dapan1121 已提交
442
    SRequestObj *pRequest = acquireRequest(*rid);
D
dapan1121 已提交
443 444 445 446 447
    if (NULL == pRequest) {
      pIter = taosHashIterate(pObj->pRequests, pIter);
      continue;
    }

448
    if (pRequest->killed || 0 == pRequest->body.queryJob) {
D
dapan1121 已提交
449
      releaseRequest(*rid);
D
dapan1121 已提交
450
      pIter = taosHashIterate(pObj->pRequests, pIter);
D
dapan1121 已提交
451 452 453 454
      continue;
    }

    tstrncpy(desc.sql, pRequest->sqlstr, sizeof(desc.sql));
455
    desc.stime = pRequest->metric.start / 1000;
L
Liu Jicong 已提交
456
    desc.queryId = pRequest->requestId;
D
dapan1121 已提交
457
    desc.useconds = now - pRequest->metric.start;
L
Liu Jicong 已提交
458
    desc.reqRid = pRequest->self;
459
    desc.stableQuery = pRequest->stableQuery;
D
dapan1121 已提交
460
    taosGetFqdn(desc.fqdn);
D
dapan1121 已提交
461
    desc.subPlanNum = pRequest->body.subplanNum;
D
dapan1121 已提交
462 463 464 465 466

    if (desc.subPlanNum) {
      desc.subDesc = taosArrayInit(desc.subPlanNum, sizeof(SQuerySubDesc));
      if (NULL == desc.subDesc) {
        releaseRequest(*rid);
S
Shengliang Guan 已提交
467
        return TSDB_CODE_OUT_OF_MEMORY;
D
dapan1121 已提交
468 469 470 471 472 473
      }

      code = schedulerGetTasksStatus(pRequest->body.queryJob, desc.subDesc);
      if (code) {
        taosArrayDestroy(desc.subDesc);
        desc.subDesc = NULL;
474
        desc.subPlanNum = 0;
D
dapan1121 已提交
475
      }
D
dapan1121 已提交
476
      desc.subPlanNum = taosArrayGetSize(desc.subDesc);
D
dapan1121 已提交
477 478
    } else {
      desc.subDesc = NULL;
D
dapan1121 已提交
479 480
    }

L
Liu Jicong 已提交
481
    releaseRequest(*rid);
D
dapan1121 已提交
482
    taosArrayPush(hbBasic->queryDesc, &desc);
L
Liu Jicong 已提交
483

D
dapan1121 已提交
484 485 486 487 488 489 490 491 492 493
    pIter = taosHashIterate(pObj->pRequests, pIter);
  }

  return TSDB_CODE_SUCCESS;
}

int32_t hbGetQueryBasicInfo(SClientHbKey *connKey, SClientHbReq *req) {
  STscObj *pTscObj = (STscObj *)acquireTscObj(connKey->tscRid);
  if (NULL == pTscObj) {
    tscWarn("tscObj rid %" PRIx64 " not exist", connKey->tscRid);
S
Shengliang Guan 已提交
494
    return TSDB_CODE_APP_ERROR;
D
dapan1121 已提交
495
  }
L
Liu Jicong 已提交
496

D
dapan1121 已提交
497 498 499 500
  SQueryHbReqBasic *hbBasic = (SQueryHbReqBasic *)taosMemoryCalloc(1, sizeof(SQueryHbReqBasic));
  if (NULL == hbBasic) {
    tscError("calloc %d failed", (int32_t)sizeof(SQueryHbReqBasic));
    releaseTscObj(connKey->tscRid);
S
Shengliang Guan 已提交
501
    return TSDB_CODE_OUT_OF_MEMORY;
D
dapan1121 已提交
502
  }
503

504 505 506 507 508 509 510 511 512
  hbBasic->connId = pTscObj->connId;

  int32_t numOfQueries = pTscObj->pRequests ? taosHashGetSize(pTscObj->pRequests) : 0;
  if (numOfQueries <= 0) {
    req->query = hbBasic;
    releaseTscObj(connKey->tscRid);
    tscDebug("no queries on connection");
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
513 514 515 516 517 518

  hbBasic->queryDesc = taosArrayInit(numOfQueries, sizeof(SQueryDesc));
  if (NULL == hbBasic->queryDesc) {
    tscWarn("taosArrayInit %d queryDesc failed", numOfQueries);
    releaseTscObj(connKey->tscRid);
    taosMemoryFree(hbBasic);
S
Shengliang Guan 已提交
519
    return TSDB_CODE_OUT_OF_MEMORY;
D
dapan1121 已提交
520
  }
L
Liu Jicong 已提交
521

D
dapan1121 已提交
522 523 524
  int32_t code = hbBuildQueryDesc(hbBasic, pTscObj);
  if (code) {
    releaseTscObj(connKey->tscRid);
D
dapan1121 已提交
525 526 527
    if (hbBasic->queryDesc) {
      taosArrayDestroyEx(hbBasic->queryDesc, tFreeClientHbQueryDesc);
    }
D
dapan1121 已提交
528 529 530 531 532 533 534 535 536 537
    taosMemoryFree(hbBasic);
    return code;
  }

  req->query = hbBasic;
  releaseTscObj(connKey->tscRid);

  return TSDB_CODE_SUCCESS;
}

K
kailixu 已提交
538 539 540 541 542 543 544 545 546 547 548 549 550
static int32_t hbGetUserAuthInfo(SClientHbKey *connKey, SHbParam *param, SClientHbReq *req) {
  STscObj *pTscObj = (STscObj *)acquireTscObj(connKey->tscRid);
  if (!pTscObj) {
    tscWarn("tscObj rid %" PRIx64 " not exist", connKey->tscRid);
    return TSDB_CODE_APP_ERROR;
  }

  int32_t code = 0;

  SKv  kv = {.key = HEARTBEAT_KEY_USER_AUTHINFO};
  SKv *pKv = NULL;
  if ((pKv = taosHashGet(req->info, &kv.key, sizeof(kv.key)))) {
    int32_t           userNum = pKv->valueLen / sizeof(SUserAuthVersion);
K
kailixu 已提交
551
    SUserAuthVersion *userAuths = (SUserAuthVersion *)pKv->value;
K
kailixu 已提交
552
    for (int32_t i = 0; i < userNum; ++i) {
K
kailixu 已提交
553 554
      SUserAuthVersion *pUserAuth = userAuths + i;
      // both key and user exist, update version
K
kailixu 已提交
555
      if (strncmp(pUserAuth->user, pTscObj->user, TSDB_USER_LEN) == 0) {
K
kailixu 已提交
556
        pUserAuth->version = htonl(-1);  // force get userAuthInfo
K
kailixu 已提交
557 558 559
        goto _return;
      }
    }
K
kailixu 已提交
560
    // key exists, user not exist, append user
K
kailixu 已提交
561 562 563 564
    SUserAuthVersion *qUserAuth =
        (SUserAuthVersion *)taosMemoryRealloc(pKv->value, (userNum + 1) * sizeof(SUserAuthVersion));
    if (qUserAuth) {
      strncpy((qUserAuth + userNum)->user, pTscObj->user, TSDB_USER_LEN);
K
kailixu 已提交
565
      (qUserAuth + userNum)->version = htonl(-1);  // force get userAuthInfo
K
kailixu 已提交
566 567 568 569 570 571 572 573
      pKv->value = qUserAuth;
      pKv->valueLen += sizeof(SUserAuthVersion);
    } else {
      code = TSDB_CODE_OUT_OF_MEMORY;
    }
    goto _return;
  }

K
kailixu 已提交
574
  // key/user not exist, add user
K
kailixu 已提交
575 576 577 578 579 580
  SUserAuthVersion *user = taosMemoryMalloc(sizeof(SUserAuthVersion));
  if (!user) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _return;
  }
  strncpy(user->user, pTscObj->user, TSDB_USER_LEN);
K
kailixu 已提交
581
  user->version = htonl(-1);  // force get userAuthInfo
K
kailixu 已提交
582 583 584 585 586 587 588 589 590 591 592
  kv.valueLen = sizeof(SUserAuthVersion);
  kv.value = user;

  tscDebug("hb got user auth info, valueLen:%d, user:%s, authVer:%d, tscRid:%" PRIi64, kv.valueLen, user->user,
           pTscObj->authVer, connKey->tscRid);

  if (!req->info) {
    req->info = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
  }

  if (taosHashPut(req->info, &kv.key, sizeof(kv.key), &kv, sizeof(kv)) < 0) {
K
kailixu 已提交
593
    taosMemoryFree(user);
K
kailixu 已提交
594 595 596 597 598 599 600 601 602 603 604 605 606
    code = terrno ? terrno : TSDB_CODE_APP_ERROR;
    goto _return;
  }

_return:
  releaseTscObj(connKey->tscRid);
  if (code) {
    tscError("hb got user auth info failed since %s", terrstr(code));
  }

  return code;
}

D
dapan 已提交
607 608 609 610 611 612 613 614 615
int32_t hbGetExpiredUserInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SClientHbReq *req) {
  SUserAuthVersion *users = NULL;
  uint32_t          userNum = 0;
  int32_t           code = 0;

  code = catalogGetExpiredUsers(pCatalog, &users, &userNum);
  if (TSDB_CODE_SUCCESS != code) {
    return code;
  }
K
kailixu 已提交
616 617 618 619 620
  STscObj *pTscObj = (STscObj *)acquireTscObj(connKey->tscRid);
  if (NULL == pTscObj) {
    tscWarn("tscObj rid %" PRIx64 " not exist", connKey->tscRid);
    return TSDB_CODE_APP_ERROR;
  }
D
dapan 已提交
621 622

  if (userNum <= 0) {
K
kailixu 已提交
623 624
    // printf("%s:%d hb get expired: [User:%s] NNNNo user since user num:  %d, second:%d\n", __func__, __LINE__, pTscObj->user, userNum,
    //        taosGetTimestampSec());
D
dapan1121 已提交
625
    taosMemoryFree(users);
K
kailixu 已提交
626
    releaseTscObj(connKey->tscRid);
D
dapan 已提交
627
    return TSDB_CODE_SUCCESS;
K
kailixu 已提交
628 629 630
  } 

  releaseTscObj(connKey->tscRid);
D
dapan 已提交
631 632 633 634 635 636 637 638 639 640 641 642 643 644

  for (int32_t i = 0; i < userNum; ++i) {
    SUserAuthVersion *user = &users[i];
    user->version = htonl(user->version);
  }

  SKv kv = {
      .key = HEARTBEAT_KEY_USER_AUTHINFO,
      .valueLen = sizeof(SUserAuthVersion) * userNum,
      .value = users,
  };

  tscDebug("hb got %d expired users, valueLen:%d", userNum, kv.valueLen);

D
dapan1121 已提交
645 646 647
  if (NULL == req->info) {
    req->info = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
  }
648

D
dapan 已提交
649 650 651 652 653
  taosHashPut(req->info, &kv.key, sizeof(kv.key), &kv, sizeof(kv));

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
654
int32_t hbGetExpiredDBInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SClientHbReq *req) {
D
dapan1121 已提交
655
  SDbCacheInfo *dbs = NULL;
L
fix  
Liu Jicong 已提交
656 657
  uint32_t      dbNum = 0;
  int32_t       code = 0;
D
dapan1121 已提交
658 659 660 661 662 663

  code = catalogGetExpiredDBs(pCatalog, &dbs, &dbNum);
  if (TSDB_CODE_SUCCESS != code) {
    return code;
  }

D
dapan1121 已提交
664
  if (dbNum <= 0) {
D
dapan1121 已提交
665
    taosMemoryFree(dbs);
D
dapan1121 已提交
666 667 668
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
669
  for (int32_t i = 0; i < dbNum; ++i) {
D
dapan1121 已提交
670
    SDbCacheInfo *db = &dbs[i];
H
Haojun Liao 已提交
671
    tscDebug("the %dth expired dbFName:%s, dbId:%" PRId64 ", vgVersion:%d, cfgVersion:%d, numOfTable:%d, startTs:%" PRId64,
D
dapan1121 已提交
672
      i, db->dbFName, db->dbId, db->vgVersion, db->cfgVersion, db->numOfTable, db->stateTs);
D
dapan1121 已提交
673

D
dapan1121 已提交
674 675
    db->dbId = htobe64(db->dbId);
    db->vgVersion = htonl(db->vgVersion);
D
dapan1121 已提交
676
    db->cfgVersion = htonl(db->cfgVersion);
D
dapan 已提交
677
    db->numOfTable = htonl(db->numOfTable);
D
dapan1121 已提交
678
    db->stateTs = htobe64(db->stateTs);
D
dapan1121 已提交
679 680
  }

L
Liu Jicong 已提交
681 682
  SKv kv = {
      .key = HEARTBEAT_KEY_DBINFO,
D
dapan1121 已提交
683
      .valueLen = sizeof(SDbCacheInfo) * dbNum,
L
Liu Jicong 已提交
684 685
      .value = dbs,
  };
D
dapan1121 已提交
686 687 688

  tscDebug("hb got %d expired db, valueLen:%d", dbNum, kv.valueLen);

D
dapan1121 已提交
689 690 691 692
  if (NULL == req->info) {
    req->info = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
  }

D
dapan1121 已提交
693 694 695 696 697
  taosHashPut(req->info, &kv.key, sizeof(kv.key), &kv, sizeof(kv));

  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
698
int32_t hbGetExpiredStbInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SClientHbReq *req) {
D
dapan1121 已提交
699
  SSTableVersion *stbs = NULL;
700 701
  uint32_t        stbNum = 0;
  int32_t         code = 0;
D
dapan 已提交
702 703 704 705 706 707 708

  code = catalogGetExpiredSTables(pCatalog, &stbs, &stbNum);
  if (TSDB_CODE_SUCCESS != code) {
    return code;
  }

  if (stbNum <= 0) {
D
dapan1121 已提交
709
    taosMemoryFree(stbs);
D
dapan 已提交
710 711 712 713
    return TSDB_CODE_SUCCESS;
  }

  for (int32_t i = 0; i < stbNum; ++i) {
D
dapan1121 已提交
714
    SSTableVersion *stb = &stbs[i];
D
dapan 已提交
715 716
    stb->suid = htobe64(stb->suid);
    stb->sversion = htons(stb->sversion);
L
fix  
Liu Jicong 已提交
717
    stb->tversion = htons(stb->tversion);
D
dapan1121 已提交
718
    stb->smaVer = htonl(stb->smaVer);
D
dapan 已提交
719 720
  }

L
Liu Jicong 已提交
721 722
  SKv kv = {
      .key = HEARTBEAT_KEY_STBINFO,
D
dapan1121 已提交
723
      .valueLen = sizeof(SSTableVersion) * stbNum,
L
Liu Jicong 已提交
724 725
      .value = stbs,
  };
D
dapan 已提交
726 727 728

  tscDebug("hb got %d expired stb, valueLen:%d", stbNum, kv.valueLen);

D
dapan1121 已提交
729 730 731 732
  if (NULL == req->info) {
    req->info = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
  }

D
dapan 已提交
733 734 735 736 737
  taosHashPut(req->info, &kv.key, sizeof(kv.key), &kv, sizeof(kv));

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
738
int32_t hbGetAppInfo(int64_t clusterId, SClientHbReq *req) {
739
  SAppHbReq *pApp = taosHashGet(clientHbMgr.appSummary, &clusterId, sizeof(clusterId));
D
dapan1121 已提交
740 741 742 743 744 745
  if (NULL != pApp) {
    memcpy(&req->app, pApp, sizeof(*pApp));
  } else {
    memset(&req->app.summary, 0, sizeof(req->app.summary));
    req->app.pid = taosGetPId();
    req->app.appId = clientHbMgr.appId;
746
    taosGetAppName(req->app.name, NULL);
D
dapan1121 已提交
747 748 749 750 751
  }

  return TSDB_CODE_SUCCESS;
}

K
kailixu 已提交
752
int32_t hbQueryHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq *req) {
753 754 755 756 757 758 759 760 761 762
  int32_t   code = 0;
  SHbParam *hbParam = (SHbParam *)param;
  SCatalog *pCatalog = NULL;

  if (hbParam->reqCnt == 0) {
    code = catalogGetHandle(hbParam->clusterId, &pCatalog);
    if (code != TSDB_CODE_SUCCESS) {
      tscWarn("catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", hbParam->clusterId, tstrerror(code));
      return code;
    }
D
dapan1121 已提交
763
  }
L
fix  
Liu Jicong 已提交
764

K
kailixu 已提交
765
  hbGetAppInfo(hbParam->clusterId, req);
D
dapan1121 已提交
766

D
dapan1121 已提交
767
  hbGetQueryBasicInfo(connKey, req);
L
Liu Jicong 已提交
768

K
kailixu 已提交
769 770 771 772 773 774 775
  if (hbParam->reqCnt == 0) {
    if(!tSimpleHashGet(clientHbMgr.appHbHash, &hbParam->clusterId, sizeof(hbParam->clusterId))) {
      code = hbGetExpiredUserInfo(connKey, pCatalog, req);
      if (TSDB_CODE_SUCCESS != code) {
        return code;
      }
    }
K
kailixu 已提交
776

K
kailixu 已提交
777
    // invoke after hbGetExpiredUserInfo
K
kailixu 已提交
778
    if (2 != atomic_load_8(&hbParam->pAppHbMgr->connHbFlag)) {
K
kailixu 已提交
779 780 781 782
      code = hbGetUserAuthInfo(connKey, hbParam, req);
      if (TSDB_CODE_SUCCESS != code) {
        return code;
      }
K
kailixu 已提交
783
      atomic_store_8(&hbParam->pAppHbMgr->connHbFlag, 1);
784
    }
D
dapan 已提交
785

786 787 788 789
    code = hbGetExpiredDBInfo(connKey, pCatalog, req);
    if (TSDB_CODE_SUCCESS != code) {
      return code;
    }
D
dapan1121 已提交
790

791 792 793 794
    code = hbGetExpiredStbInfo(connKey, pCatalog, req);
    if (TSDB_CODE_SUCCESS != code) {
      return code;
    }
D
dapan 已提交
795
  }
K
kailixu 已提交
796
  ++hbParam->reqCnt;  // success to get catalog info
K
kailixu 已提交
797

D
dapan1121 已提交
798 799 800
  return TSDB_CODE_SUCCESS;
}

801 802
static FORCE_INLINE void hbMgrInitHandle() {
  // init all handle
D
dapan1121 已提交
803 804
  clientHbMgr.reqHandle[CONN_TYPE__QUERY] = hbQueryHbReqHandle;
  clientHbMgr.reqHandle[CONN_TYPE__TMQ] = hbMqHbReqHandle;
L
Liu Jicong 已提交
805

D
dapan1121 已提交
806 807
  clientHbMgr.rspHandle[CONN_TYPE__QUERY] = hbQueryHbRspHandle;
  clientHbMgr.rspHandle[CONN_TYPE__TMQ] = hbMqHbRspHandle;
L
Liu Jicong 已提交
808 809
}

L
fix  
Liu Jicong 已提交
810
SClientHbBatchReq *hbGatherAllInfo(SAppHbMgr *pAppHbMgr) {
wafwerar's avatar
wafwerar 已提交
811
  SClientHbBatchReq *pBatchReq = taosMemoryCalloc(1, sizeof(SClientHbBatchReq));
L
Liu Jicong 已提交
812
  if (pBatchReq == NULL) {
S
Shengliang Guan 已提交
813
    terrno = TSDB_CODE_OUT_OF_MEMORY;
L
Liu Jicong 已提交
814 815
    return NULL;
  }
L
Liu Jicong 已提交
816
  int32_t connKeyCnt = atomic_load_32(&pAppHbMgr->connKeyCnt);
L
Liu Jicong 已提交
817
  pBatchReq->reqs = taosArrayInit(connKeyCnt, sizeof(SClientHbReq));
818
  if (!pBatchReq->reqs) {
dengyihao's avatar
dengyihao 已提交
819 820 821
    tFreeClientHbBatchReq(pBatchReq);
    return NULL;
  }
D
dapan1121 已提交
822

K
kailixu 已提交
823
  void    *pIter = NULL;
K
kailixu 已提交
824
  SHbParam param = {.pAppHbMgr = pAppHbMgr};
K
kailixu 已提交
825 826 827 828 829 830 831 832
  while ((pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter))) {
    SClientHbReq *pOneReq = pIter;
    SClientHbKey *connKey = &pOneReq->connKey;
    STscObj      *pTscObj = (STscObj *)acquireTscObj(connKey->tscRid);

    if (!pTscObj) {
      continue;
    }
833

dengyihao's avatar
dengyihao 已提交
834
    pOneReq = taosArrayPush(pBatchReq->reqs, pOneReq);
835

K
kailixu 已提交
836
    switch (connKey->connType) {
K
kailixu 已提交
837
      case CONN_TYPE__QUERY: {
838 839 840
        if (param.clusterId == 0) {
          // init
          param.clusterId = pOneReq->clusterId;
K
kailixu 已提交
841
          param.connHbFlag = atomic_load_8(&pAppHbMgr->connHbFlag);
842
        }
K
kailixu 已提交
843 844 845 846 847
        break;
      }
      default:
        break;
    }
K
kailixu 已提交
848 849
    if (clientHbMgr.reqHandle[connKey->connType]) {
      int32_t code = (*clientHbMgr.reqHandle[connKey->connType])(connKey, &param, pOneReq);
K
kailixu 已提交
850 851
      if (code) {
        tscWarn("hbGatherAllInfo failed since %s, tscRid:%" PRIi64 ", connType:%" PRIi8, tstrerror(code),
K
kailixu 已提交
852
                connKey->tscRid, connKey->connType);
K
kailixu 已提交
853 854
      }
    }
K
kailixu 已提交
855

K
kailixu 已提交
856
    releaseTscObj(connKey->tscRid);
L
Liu Jicong 已提交
857
  }
L
Liu Jicong 已提交
858

L
Liu Jicong 已提交
859
  return pBatchReq;
L
Liu Jicong 已提交
860 861
}

862
void hbThreadFuncUnexpectedStopped(void) { atomic_store_8(&clientHbMgr.threadStop, 2); }
wafwerar's avatar
wafwerar 已提交
863

864
void hbMergeSummary(SAppClusterSummary *dst, SAppClusterSummary *src) {
D
dapan1121 已提交
865 866 867 868 869 870 871 872 873 874 875 876 877
  dst->numOfInsertsReq += src->numOfInsertsReq;
  dst->numOfInsertRows += src->numOfInsertRows;
  dst->insertElapsedTime += src->insertElapsedTime;
  dst->insertBytes += src->insertBytes;
  dst->fetchBytes += src->fetchBytes;
  dst->queryElapsedTime += src->queryElapsedTime;
  dst->numOfSlowQueries += src->numOfSlowQueries;
  dst->totalRequests += src->totalRequests;
  dst->currentRequests += src->currentRequests;
}

int32_t hbGatherAppInfo(void) {
  SAppHbReq req = {0};
878
  int       sz = taosArrayGetSize(clientHbMgr.appHbMgrs);
D
dapan1121 已提交
879 880 881 882 883
  if (sz > 0) {
    req.pid = taosGetPId();
    req.appId = clientHbMgr.appId;
    taosGetAppName(req.name, NULL);
  }
D
dapan1121 已提交
884 885

  taosHashClear(clientHbMgr.appSummary);
886

D
dapan1121 已提交
887 888
  for (int32_t i = 0; i < sz; ++i) {
    SAppHbMgr *pAppHbMgr = taosArrayGetP(clientHbMgr.appHbMgrs, i);
dengyihao's avatar
dengyihao 已提交
889 890
    if (pAppHbMgr == NULL) continue;

891 892
    uint64_t   clusterId = pAppHbMgr->pAppInstInfo->clusterId;
    SAppHbReq *pApp = taosHashGet(clientHbMgr.appSummary, &clusterId, sizeof(clusterId));
D
dapan1121 已提交
893 894
    if (NULL == pApp) {
      memcpy(&req.summary, &pAppHbMgr->pAppInstInfo->summary, sizeof(req.summary));
D
dapan1121 已提交
895
      req.startTime = pAppHbMgr->startTime;
D
dapan1121 已提交
896 897 898 899 900
      taosHashPut(clientHbMgr.appSummary, &clusterId, sizeof(clusterId), &req, sizeof(req));
    } else {
      if (pAppHbMgr->startTime < pApp->startTime) {
        pApp->startTime = pAppHbMgr->startTime;
      }
901

D
dapan1121 已提交
902 903 904 905 906 907 908
      hbMergeSummary(&pApp->summary, &pAppHbMgr->pAppInstInfo->summary);
    }
  }

  return TSDB_CODE_SUCCESS;
}

L
fix  
Liu Jicong 已提交
909
static void *hbThreadFunc(void *param) {
L
Liu Jicong 已提交
910
  setThreadName("hb");
wafwerar's avatar
wafwerar 已提交
911
#ifdef WINDOWS
wafwerar's avatar
wafwerar 已提交
912 913 914
  if (taosCheckCurrentInDll()) {
    atexit(hbThreadFuncUnexpectedStopped);
  }
wafwerar's avatar
wafwerar 已提交
915
#endif
L
Liu Jicong 已提交
916
  while (1) {
917
    if (1 == clientHbMgr.threadStop) {
L
Liu Jicong 已提交
918 919 920
      break;
    }

wafwerar's avatar
wafwerar 已提交
921
    taosThreadMutexLock(&clientHbMgr.lock);
922

L
Liu Jicong 已提交
923
    int sz = taosArrayGetSize(clientHbMgr.appHbMgrs);
D
dapan1121 已提交
924 925
    if (sz > 0) {
      hbGatherAppInfo();
K
kailixu 已提交
926
      if (sz > 1 && !clientHbMgr.appHbHash) {
K
kailixu 已提交
927
        clientHbMgr.appHbHash = tSimpleHashInit(0, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT));
K
kailixu 已提交
928
      }
K
kailixu 已提交
929
      tSimpleHashClear(clientHbMgr.appHbHash);
D
dapan1121 已提交
930
    }
931

L
fix  
Liu Jicong 已提交
932 933
    for (int i = 0; i < sz; i++) {
      SAppHbMgr *pAppHbMgr = taosArrayGetP(clientHbMgr.appHbMgrs, i);
dengyihao's avatar
dengyihao 已提交
934 935 936
      if (pAppHbMgr == NULL) {
        continue;
      }
L
Liu Jicong 已提交
937

L
Liu Jicong 已提交
938 939 940 941
      int32_t connCnt = atomic_load_32(&pAppHbMgr->connKeyCnt);
      if (connCnt == 0) {
        continue;
      }
L
fix  
Liu Jicong 已提交
942
      SClientHbBatchReq *pReq = hbGatherAllInfo(pAppHbMgr);
dengyihao's avatar
dengyihao 已提交
943 944
      if (pReq == NULL || taosArrayGetP(clientHbMgr.appHbMgrs, i) == NULL) {
        tFreeClientHbBatchReq(pReq);
L
Liu Jicong 已提交
945 946
        continue;
      }
L
fix  
Liu Jicong 已提交
947
      int   tlen = tSerializeSClientHbBatchReq(NULL, 0, pReq);
wafwerar's avatar
wafwerar 已提交
948
      void *buf = taosMemoryMalloc(tlen);
L
Liu Jicong 已提交
949
      if (buf == NULL) {
S
Shengliang Guan 已提交
950
        terrno = TSDB_CODE_OUT_OF_MEMORY;
D
dapan1121 已提交
951
        tFreeClientHbBatchReq(pReq);
dengyihao's avatar
dengyihao 已提交
952
        // hbClearReqInfo(pAppHbMgr);
L
Liu Jicong 已提交
953 954
        break;
      }
L
fix  
Liu Jicong 已提交
955

S
Shengliang Guan 已提交
956
      tSerializeSClientHbBatchReq(buf, tlen, pReq);
wafwerar's avatar
wafwerar 已提交
957
      SMsgSendInfo *pInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
958

L
Liu Jicong 已提交
959
      if (pInfo == NULL) {
S
Shengliang Guan 已提交
960
        terrno = TSDB_CODE_OUT_OF_MEMORY;
D
dapan1121 已提交
961
        tFreeClientHbBatchReq(pReq);
dengyihao's avatar
dengyihao 已提交
962
        // hbClearReqInfo(pAppHbMgr);
wafwerar's avatar
wafwerar 已提交
963
        taosMemoryFree(buf);
L
Liu Jicong 已提交
964 965
        break;
      }
L
Liu Jicong 已提交
966
      pInfo->fp = hbAsyncCallBack;
L
Liu Jicong 已提交
967 968 969
      pInfo->msgInfo.pData = buf;
      pInfo->msgInfo.len = tlen;
      pInfo->msgType = TDMT_MND_HEARTBEAT;
D
dapan1121 已提交
970 971
      pInfo->param = taosMemoryMalloc(sizeof(int32_t));
      *(int32_t *)pInfo->param = i;
dengyihao's avatar
dengyihao 已提交
972
      pInfo->paramFreeFp = taosMemoryFree;
L
Liu Jicong 已提交
973 974
      pInfo->requestId = generateRequestId();
      pInfo->requestObjRefId = 0;
L
Liu Jicong 已提交
975 976

      SAppInstInfo *pAppInstInfo = pAppHbMgr->pAppInstInfo;
L
fix  
Liu Jicong 已提交
977 978
      int64_t       transporterId = 0;
      SEpSet        epSet = getEpSet_s(&pAppInstInfo->mgmtEp);
L
Liu Jicong 已提交
979
      asyncSendMsgToServer(pAppInstInfo->pTransporter, &epSet, &transporterId, pInfo);
D
dapan1121 已提交
980
      tFreeClientHbBatchReq(pReq);
dengyihao's avatar
dengyihao 已提交
981
      // hbClearReqInfo(pAppHbMgr);
K
kailixu 已提交
982
      tSimpleHashPut(clientHbMgr.appHbHash, &pAppHbMgr->pAppInstInfo->clusterId, sizeof(uint64_t), NULL, 0);
L
Liu Jicong 已提交
983 984
      atomic_add_fetch_32(&pAppHbMgr->reportCnt, 1);
    }
985

wafwerar's avatar
wafwerar 已提交
986
    taosThreadMutexUnlock(&clientHbMgr.lock);
L
fix  
Liu Jicong 已提交
987

L
Liu Jicong 已提交
988
    taosMsleep(HEARTBEAT_INTERVAL);
L
Liu Jicong 已提交
989
  }
K
kailixu 已提交
990

L
Liu Jicong 已提交
991 992 993 994
  return NULL;
}

static int32_t hbCreateThread() {
wafwerar's avatar
wafwerar 已提交
995 996 997
  TdThreadAttr thAttr;
  taosThreadAttrInit(&thAttr);
  taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
L
Liu Jicong 已提交
998

D
dapan1121 已提交
999 1000 1001
  if (taosThreadCreate(&clientHbMgr.thread, &thAttr, hbThreadFunc, NULL) != 0) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
L
Liu Jicong 已提交
1002
  }
D
dapan1121 已提交
1003
  taosThreadAttrDestroy(&thAttr);
L
Liu Jicong 已提交
1004 1005 1006
  return 0;
}

L
Liu Jicong 已提交
1007
static void hbStopThread() {
D
dapan1121 已提交
1008 1009 1010
  if (0 == atomic_load_8(&clientHbMgr.inited)) {
    return;
  }
1011
  if (atomic_val_compare_exchange_8(&clientHbMgr.threadStop, 0, 1)) {
D
dapan1121 已提交
1012
    tscDebug("hb thread already stopped");
1013 1014
    return;
  }
L
fix  
Liu Jicong 已提交
1015

1016 1017 1018 1019 1020 1021
  // thread quit mode kill or inner exit from self-thread
  if (clientHbMgr.quitByKill) {
    taosThreadKill(clientHbMgr.thread, 0);
  } else {
    taosThreadJoin(clientHbMgr.thread, NULL);
  }
D
dapan1121 已提交
1022

L
fix  
Liu Jicong 已提交
1023
  tscDebug("hb thread stopped");
L
Liu Jicong 已提交
1024 1025
}

L
fix  
Liu Jicong 已提交
1026
SAppHbMgr *appHbMgrInit(SAppInstInfo *pAppInstInfo, char *key) {
1027
  if (hbMgrInit() != 0) {
1028 1029 1030
    terrno = TSDB_CODE_TSC_INTERNAL_ERROR;
    return NULL;
  }
wafwerar's avatar
wafwerar 已提交
1031
  SAppHbMgr *pAppHbMgr = taosMemoryMalloc(sizeof(SAppHbMgr));
L
Liu Jicong 已提交
1032 1033 1034 1035 1036 1037
  if (pAppHbMgr == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }
  // init stat
  pAppHbMgr->startTime = taosGetTimestampMs();
L
Liu Jicong 已提交
1038
  pAppHbMgr->connKeyCnt = 0;
K
kailixu 已提交
1039
  pAppHbMgr->connHbFlag = 0;
L
Liu Jicong 已提交
1040 1041
  pAppHbMgr->reportCnt = 0;
  pAppHbMgr->reportBytes = 0;
1042
  pAppHbMgr->key = taosStrdup(key);
L
Liu Jicong 已提交
1043

L
Liu Jicong 已提交
1044 1045
  // init app info
  pAppHbMgr->pAppInstInfo = pAppInstInfo;
L
Liu Jicong 已提交
1046 1047 1048

  // init hash info
  pAppHbMgr->activeInfo = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
L
Liu Jicong 已提交
1049 1050 1051

  if (pAppHbMgr->activeInfo == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
wafwerar's avatar
wafwerar 已提交
1052
    taosMemoryFree(pAppHbMgr);
L
Liu Jicong 已提交
1053 1054
    return NULL;
  }
H
Haojun Liao 已提交
1055

1056
  // taosHashSetFreeFp(pAppHbMgr->activeInfo, tFreeClientHbReq);
L
Liu Jicong 已提交
1057

wafwerar's avatar
wafwerar 已提交
1058
  taosThreadMutexLock(&clientHbMgr.lock);
L
Liu Jicong 已提交
1059
  taosArrayPush(clientHbMgr.appHbMgrs, &pAppHbMgr);
D
dapan1121 已提交
1060
  pAppHbMgr->idx = taosArrayGetSize(clientHbMgr.appHbMgrs) - 1;
wafwerar's avatar
wafwerar 已提交
1061
  taosThreadMutexUnlock(&clientHbMgr.lock);
L
fix  
Liu Jicong 已提交
1062

L
Liu Jicong 已提交
1063 1064 1065
  return pAppHbMgr;
}

D
dapan1121 已提交
1066 1067 1068 1069 1070 1071 1072 1073 1074
void hbFreeAppHbMgr(SAppHbMgr *pTarget) {
  void *pIter = taosHashIterate(pTarget->activeInfo, NULL);
  while (pIter != NULL) {
    SClientHbReq *pOneReq = pIter;
    tFreeClientHbReq(pOneReq);
    pIter = taosHashIterate(pTarget->activeInfo, pIter);
  }
  taosHashCleanup(pTarget->activeInfo);
  pTarget->activeInfo = NULL;
1075

D
dapan1121 已提交
1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087
  taosMemoryFree(pTarget->key);
  taosMemoryFree(pTarget);
}

void hbRemoveAppHbMrg(SAppHbMgr **pAppHbMgr) {
  taosThreadMutexLock(&clientHbMgr.lock);
  int32_t mgrSize = taosArrayGetSize(clientHbMgr.appHbMgrs);
  for (int32_t i = 0; i < mgrSize; ++i) {
    SAppHbMgr *pItem = taosArrayGetP(clientHbMgr.appHbMgrs, i);
    if (pItem == *pAppHbMgr) {
      hbFreeAppHbMgr(*pAppHbMgr);
      *pAppHbMgr = NULL;
dengyihao's avatar
dengyihao 已提交
1088
      taosArraySet(clientHbMgr.appHbMgrs, i, pAppHbMgr);
D
dapan1121 已提交
1089 1090 1091 1092 1093 1094
      break;
    }
  }
  taosThreadMutexUnlock(&clientHbMgr.lock);
}

D
dapan1121 已提交
1095
void appHbMgrCleanup(void) {
L
Liu Jicong 已提交
1096 1097
  int sz = taosArrayGetSize(clientHbMgr.appHbMgrs);
  for (int i = 0; i < sz; i++) {
L
fix  
Liu Jicong 已提交
1098
    SAppHbMgr *pTarget = taosArrayGetP(clientHbMgr.appHbMgrs, i);
dengyihao's avatar
dengyihao 已提交
1099
    if (pTarget == NULL) continue;
D
dapan1121 已提交
1100
    hbFreeAppHbMgr(pTarget);
L
Liu Jicong 已提交
1101
  }
K
kailixu 已提交
1102 1103 1104 1105 1106
  clientHbMgr.appHbMgrs = taosArrayDestroy(clientHbMgr.appHbMgrs);
  tSimpleHashCleanup(clientHbMgr.appHbHash);
  clientHbMgr.appHbHash = NULL;
  taosHashCleanup(clientHbMgr.appSummary);
  clientHbMgr.appSummary = NULL;
L
Liu Jicong 已提交
1107 1108 1109
}

int hbMgrInit() {
L
Liu Jicong 已提交
1110 1111 1112 1113
  // init once
  int8_t old = atomic_val_compare_exchange_8(&clientHbMgr.inited, 0, 1);
  if (old == 1) return 0;

D
dapan1121 已提交
1114 1115
  clientHbMgr.appId = tGenIdPI64();
  tscDebug("app %" PRIx64 " initialized", clientHbMgr.appId);
1116

D
dapan1121 已提交
1117
  clientHbMgr.appSummary = taosHashInit(10, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
L
fix  
Liu Jicong 已提交
1118
  clientHbMgr.appHbMgrs = taosArrayInit(0, sizeof(void *));
dengyihao's avatar
dengyihao 已提交
1119 1120

  TdThreadMutexAttr attr = {0};
dengyihao's avatar
dengyihao 已提交
1121

dengyihao's avatar
dengyihao 已提交
1122
  int ret = taosThreadMutexAttrInit(&attr);
1123 1124
  if (ret != 0) {
    uError("hbMgrInit:taosThreadMutexAttrInit error") return ret;
1125
  }
dengyihao's avatar
dengyihao 已提交
1126

dengyihao's avatar
dengyihao 已提交
1127
  ret = taosThreadMutexAttrSetType(&attr, PTHREAD_MUTEX_RECURSIVE);
1128 1129
  if (ret != 0) {
    uError("hbMgrInit:taosThreadMutexAttrSetType error") return ret;
1130
  }
dengyihao's avatar
dengyihao 已提交
1131 1132

  ret = taosThreadMutexInit(&clientHbMgr.lock, &attr);
1133 1134
  if (ret != 0) {
    uError("hbMgrInit:taosThreadMutexInit error") return ret;
1135
  }
dengyihao's avatar
dengyihao 已提交
1136 1137

  ret = taosThreadMutexAttrDestroy(&attr);
1138 1139
  if (ret != 0) {
    uError("hbMgrInit:taosThreadMutexAttrDestroy error") return ret;
1140
  }
L
Liu Jicong 已提交
1141 1142 1143 1144 1145

  // init handle funcs
  hbMgrInitHandle();

  // init backgroud thread
D
dapan1121 已提交
1146
  hbCreateThread();
L
Liu Jicong 已提交
1147

L
Liu Jicong 已提交
1148 1149 1150 1151
  return 0;
}

void hbMgrCleanUp() {
D
dapan1121 已提交
1152
  hbStopThread();
L
fix  
Liu Jicong 已提交
1153

L
Liu Jicong 已提交
1154
  // destroy all appHbMgr
L
Liu Jicong 已提交
1155 1156 1157
  int8_t old = atomic_val_compare_exchange_8(&clientHbMgr.inited, 1, 0);
  if (old == 0) return;

wafwerar's avatar
wafwerar 已提交
1158
  taosThreadMutexLock(&clientHbMgr.lock);
D
dapan1121 已提交
1159
  appHbMgrCleanup();
wafwerar's avatar
wafwerar 已提交
1160
  taosThreadMutexUnlock(&clientHbMgr.lock);
L
Liu Jicong 已提交
1161 1162
}

D
dapan1121 已提交
1163
int hbRegisterConnImpl(SAppHbMgr *pAppHbMgr, SClientHbKey connKey, int64_t clusterId) {
L
Liu Jicong 已提交
1164
  // init hash in activeinfo
L
fix  
Liu Jicong 已提交
1165
  void *data = taosHashGet(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey));
L
Liu Jicong 已提交
1166 1167 1168
  if (data != NULL) {
    return 0;
  }
D
dapan1121 已提交
1169
  SClientHbReq hbReq = {0};
L
Liu Jicong 已提交
1170
  hbReq.connKey = connKey;
D
dapan1121 已提交
1171
  hbReq.clusterId = clusterId;
1172
  // hbReq.info = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
L
fix  
Liu Jicong 已提交
1173

L
Liu Jicong 已提交
1174
  taosHashPut(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey), &hbReq, sizeof(SClientHbReq));
L
fix  
Liu Jicong 已提交
1175

L
Liu Jicong 已提交
1176
  atomic_add_fetch_32(&pAppHbMgr->connKeyCnt, 1);
L
Liu Jicong 已提交
1177 1178 1179
  return 0;
}

D
dapan1121 已提交
1180
int hbRegisterConn(SAppHbMgr *pAppHbMgr, int64_t tscRefId, int64_t clusterId, int8_t connType) {
L
Liu Jicong 已提交
1181
  SClientHbKey connKey = {
D
dapan1121 已提交
1182
      .tscRid = tscRefId,
D
dapan1121 已提交
1183
      .connType = connType,
L
Liu Jicong 已提交
1184
  };
D
dapan1121 已提交
1185

D
dapan1121 已提交
1186 1187
  switch (connType) {
    case CONN_TYPE__QUERY: {
D
dapan1121 已提交
1188
      return hbRegisterConnImpl(pAppHbMgr, connKey, clusterId);
D
dapan1121 已提交
1189
    }
D
dapan1121 已提交
1190
    case CONN_TYPE__TMQ: {
L
Liu Jicong 已提交
1191
      return 0;
D
dapan1121 已提交
1192 1193
    }
    default:
L
Liu Jicong 已提交
1194
      return 0;
D
dapan1121 已提交
1195 1196 1197
  }
}

K
kailixu 已提交
1198 1199
void hbDeregisterConn(STscObj *pTscObj, SClientHbKey connKey) {
  SAppHbMgr    *pAppHbMgr = pTscObj->pAppInfo->pAppHbMgr;
D
dapan1121 已提交
1200
  SClientHbReq *pReq = taosHashAcquire(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey));
D
dapan1121 已提交
1201
  if (pReq) {
D
dapan1121 已提交
1202
    tFreeClientHbReq(pReq);
D
dapan1121 已提交
1203
    taosHashRemove(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey));
D
dapan1121 已提交
1204
    taosHashRelease(pAppHbMgr->activeInfo, pReq);
D
dapan1121 已提交
1205 1206
  }

D
dapan1121 已提交
1207
  if (NULL == pReq) {
D
dapan1121 已提交
1208 1209
    return;
  }
L
Liu Jicong 已提交
1210

L
Liu Jicong 已提交
1211
  atomic_sub_fetch_32(&pAppHbMgr->connKeyCnt, 1);
L
Liu Jicong 已提交
1212
}
1213 1214 1215 1216 1217

// set heart beat thread quit mode , if quicByKill 1 then kill thread else quit from inner
void taos_set_hb_quit(int8_t quitByKill) {
  clientHbMgr.quitByKill = quitByKill;
}