clientHb.c 18.0 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/>.
 */

L
Liu Jicong 已提交
16
#include "clientInt.h"
L
Liu Jicong 已提交
17
#include "trpc.h"
D
dapan1121 已提交
18 19
#include "catalog.h"
#include "clientLog.h"
L
Liu Jicong 已提交
20 21 22 23 24 25

static SClientHbMgr clientHbMgr = {0};

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

D
dapan1121 已提交
26
static int32_t hbMqHbRspHandle(struct SAppHbMgr *pAppHbMgr, SClientHbRsp* pRsp) {
L
Liu Jicong 已提交
27 28 29
  return 0;
}

D
dapan1121 已提交
30 31 32 33 34 35 36 37 38 39
static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog *pCatalog) {
  int32_t msgLen = 0;
  int32_t code = 0;
  
  while (msgLen < valueLen) {
    SUseDbRsp *rsp = (SUseDbRsp *)((char *)value + msgLen);

    rsp->vgVersion = ntohl(rsp->vgVersion);
    rsp->vgNum = ntohl(rsp->vgNum);
    rsp->uid = be64toh(rsp->uid);
D
dapan1121 已提交
40 41

    tscDebug("hb db rsp, db:%s, vgVersion:%d, uid:%"PRIx64, rsp->db, rsp->vgVersion, rsp->uid);
D
dapan1121 已提交
42 43
    
    if (rsp->vgVersion < 0) {
D
dapan1121 已提交
44
      code = catalogRemoveDB(pCatalog, rsp->db, rsp->uid);
D
dapan1121 已提交
45 46 47 48
    } else {
      SDBVgroupInfo vgInfo = {0};
      vgInfo.vgVersion = rsp->vgVersion;
      vgInfo.hashMethod = rsp->hashMethod;
D
dapan1121 已提交
49 50
      vgInfo.vgHash = taosHashInit(rsp->vgNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
      if (NULL == vgInfo.vgHash) {
D
dapan1121 已提交
51 52 53 54 55 56 57 58 59
        tscError("hash init[%d] failed", rsp->vgNum);
        return TSDB_CODE_TSC_OUT_OF_MEMORY;
      }

      for (int32_t i = 0; i < rsp->vgNum; ++i) {
        rsp->vgroupInfo[i].vgId = ntohl(rsp->vgroupInfo[i].vgId);
        rsp->vgroupInfo[i].hashBegin = ntohl(rsp->vgroupInfo[i].hashBegin);
        rsp->vgroupInfo[i].hashEnd = ntohl(rsp->vgroupInfo[i].hashEnd);

D
dapan1121 已提交
60 61
        for (int32_t n = 0; n < rsp->vgroupInfo[i].epset.numOfEps; ++n) {
          rsp->vgroupInfo[i].epset.eps[n].port = ntohs(rsp->vgroupInfo[i].epset.eps[n].port);
D
dapan1121 已提交
62 63
        }

D
dapan1121 已提交
64
        if (0 != taosHashPut(vgInfo.vgHash, &rsp->vgroupInfo[i].vgId, sizeof(rsp->vgroupInfo[i].vgId), &rsp->vgroupInfo[i], sizeof(rsp->vgroupInfo[i]))) {
D
dapan1121 已提交
65
          tscError("hash push failed, errno:%d", errno);
D
dapan1121 已提交
66
          taosHashCleanup(vgInfo.vgHash);
D
dapan1121 已提交
67 68 69 70
          return TSDB_CODE_TSC_OUT_OF_MEMORY;
        }
      }  
      
D
dapan1121 已提交
71
      code = catalogUpdateDBVgroup(pCatalog, rsp->db, rsp->uid, &vgInfo);
D
dapan1121 已提交
72
      if (code) {
D
dapan1121 已提交
73
        taosHashCleanup(vgInfo.vgHash);
D
dapan1121 已提交
74 75 76 77 78 79 80 81 82 83 84 85 86
      }
    }

    if (code) {
      return code;
    }

    msgLen += sizeof(SUseDbRsp) + rsp->vgNum * sizeof(SVgroupInfo);
  }

  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102
static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalog *pCatalog) {
  int32_t msgLen = 0;
  int32_t code = 0;
  int32_t schemaNum = 0;
  
  while (msgLen < valueLen) {
    STableMetaRsp *rsp = (STableMetaRsp *)((char *)value + msgLen);

    rsp->numOfColumns = ntohl(rsp->numOfColumns);
    rsp->suid = be64toh(rsp->suid);
    
    if (rsp->numOfColumns < 0) {
      schemaNum = 0;
      
      tscDebug("hb remove stb, db:%s, stb:%s", rsp->dbFName, rsp->stbName);

D
dapan1121 已提交
103
      catalogRemoveSTableMeta(pCatalog, rsp->dbFName, rsp->stbName, rsp->suid);
D
dapan 已提交
104
    } else {
D
dapan1121 已提交
105 106
      tscDebug("hb update stb, db:%s, stb:%s", rsp->dbFName, rsp->stbName);

D
dapan 已提交
107
      rsp->numOfTags = ntohl(rsp->numOfTags);
D
dapan1121 已提交
108 109 110 111 112 113
      rsp->sversion = ntohl(rsp->sversion);
      rsp->tversion = ntohl(rsp->tversion);
      rsp->tuid = be64toh(rsp->tuid);
      rsp->vgId = ntohl(rsp->vgId);

      SSchema* pSchema = rsp->pSchema;
D
dapan 已提交
114 115 116
      
      schemaNum = rsp->numOfColumns + rsp->numOfTags;

D
dapan1121 已提交
117 118 119
      for (int i = 0; i < schemaNum; ++i) {
        pSchema->bytes = ntohl(pSchema->bytes);
        pSchema->colId = ntohl(pSchema->colId);
D
dapan 已提交
120

D
dapan1121 已提交
121
        pSchema++;
D
dapan 已提交
122 123
      }

D
dapan1121 已提交
124 125 126 127 128 129
      if (rsp->pSchema[0].colId != PRIMARYKEY_TIMESTAMP_COL_ID) {
        tscError("invalid colId[%d] for the first column in table meta rsp msg", rsp->pSchema[0].colId);
        return TSDB_CODE_TSC_INVALID_VALUE;
      }      

      catalogUpdateSTableMeta(pCatalog, rsp);
D
dapan 已提交
130 131 132 133 134 135 136 137 138
    }

    msgLen += sizeof(STableMetaRsp) + schemaNum * sizeof(SSchema);
  }

  return TSDB_CODE_SUCCESS;
}


D
dapan1121 已提交
139 140 141 142 143 144 145 146
static int32_t hbQueryHbRspHandle(struct SAppHbMgr *pAppHbMgr, SClientHbRsp* pRsp) {
  SHbConnInfo * info = taosHashGet(pAppHbMgr->connInfo, &pRsp->connKey, sizeof(SClientHbKey));
  if (NULL == info) {
    tscWarn("fail to get connInfo, may be dropped, connId:%d, type:%d", pRsp->connKey.connId, pRsp->connKey.hbType);
    return TSDB_CODE_SUCCESS;
  }

  int32_t kvNum = pRsp->info ? taosArrayGetSize(pRsp->info) : 0;
D
dapan1121 已提交
147 148 149

  tscDebug("hb got %d rsp kv", kvNum);
  
D
dapan1121 已提交
150 151 152
  for (int32_t i = 0; i < kvNum; ++i) {
    SKv *kv = taosArrayGet(pRsp->info, i);
    switch (kv->key) {
D
dapan1121 已提交
153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168
      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;
        }

        int64_t *clusterId = (int64_t *)info->param;
        struct SCatalog *pCatalog = NULL;
        
        int32_t code = catalogGetHandle(*clusterId, &pCatalog);
        if (code != TSDB_CODE_SUCCESS) {
          tscWarn("catalogGetHandle failed, clusterId:%"PRIx64", error:%s", *clusterId, tstrerror(code));
          break;
        }

        hbProcessDBInfoRsp(kv->value, kv->valueLen, pCatalog);
D
dapan1121 已提交
169
        break;
D
dapan1121 已提交
170
      }
D
dapan 已提交
171 172 173 174 175
      case HEARTBEAT_KEY_STBINFO:{
        if (kv->valueLen <= 0 || NULL == kv->value) {
          tscError("invalid hb stb info, len:%d, value:%p", kv->valueLen, kv->value);
          break;
        }
D
dapan1121 已提交
176

D
dapan 已提交
177 178 179 180 181 182 183 184 185 186
        int64_t *clusterId = (int64_t *)info->param;
        struct SCatalog *pCatalog = NULL;
        
        int32_t code = catalogGetHandle(*clusterId, &pCatalog);
        if (code != TSDB_CODE_SUCCESS) {
          tscWarn("catalogGetHandle failed, clusterId:%"PRIx64", error:%s", *clusterId, tstrerror(code));
          break;
        }

        hbProcessStbInfoRsp(kv->value, kv->valueLen, pCatalog);
D
dapan1121 已提交
187
        break;
D
dapan 已提交
188
      }
D
dapan1121 已提交
189 190 191 192 193 194 195 196 197
      default:
        tscError("invalid hb key type:%d", kv->key);
        break;
    }
  }

  return TSDB_CODE_SUCCESS;
}

L
Liu Jicong 已提交
198
static int32_t hbMqAsyncCallBack(void* param, const SDataBuf* pMsg, int32_t code) {
D
dapan1121 已提交
199
  static int32_t emptyRspNum = 0;
L
Liu Jicong 已提交
200
  if (code != 0) {
D
dapan1121 已提交
201
    tfree(param);
L
Liu Jicong 已提交
202 203
    return -1;
  }
D
dapan1121 已提交
204
  char *key = (char *)param;
D
dapan1121 已提交
205 206 207
  SClientHbBatchRsp pRsp = {0};
  tDeserializeSClientHbBatchRsp(pMsg->pData, &pRsp);
  
D
dapan1121 已提交
208
  int32_t rspNum = taosArrayGetSize(pRsp.rsps);
D
dapan1121 已提交
209 210 211

  SAppInstInfo** pInst = taosHashGet(appInfo.pInstMap, key, strlen(key));
  if (pInst == NULL || NULL == *pInst) {
D
dapan1121 已提交
212 213 214
    tscError("cluster not exist, key:%s", key);    
    tfree(param);
    tFreeClientHbBatchRsp(&pRsp);
D
dapan1121 已提交
215 216 217
    return -1;
  }

D
dapan1121 已提交
218 219 220
  tfree(param);

  if (rspNum) {
D
dapan 已提交
221
    tscDebug("hb got %d rsp, %d empty rsp received before", rspNum, atomic_val_compare_exchange_32(&emptyRspNum, emptyRspNum, 0));
D
dapan1121 已提交
222 223 224 225 226
  } else {
    atomic_add_fetch_32(&emptyRspNum, 1);
  }

  for (int32_t i = 0; i < rspNum; ++i) {
D
dapan1121 已提交
227
    SClientHbRsp* rsp = taosArrayGet(pRsp.rsps, i);
D
dapan1121 已提交
228 229 230 231 232
    code = (*clientHbMgr.rspHandle[rsp->connKey.hbType])((*pInst)->pAppHbMgr, rsp);
    if (code) {
      break;
    }
  }
D
dapan1121 已提交
233 234

  tFreeClientHbBatchRsp(&pRsp);
D
dapan1121 已提交
235 236
  
  return code;
L
Liu Jicong 已提交
237 238
}

D
dapan1121 已提交
239 240 241 242 243 244 245 246 247 248
int32_t hbGetExpiredDBInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SClientHbReq *req) {
  SDbVgVersion *dbs = NULL;
  uint32_t dbNum = 0;
  int32_t code = 0;

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

D
dapan1121 已提交
249 250 251 252
  if (dbNum <= 0) {
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
253 254 255 256 257 258 259
  for (int32_t i = 0; i < dbNum; ++i) {
    SDbVgVersion *db = &dbs[i];
    db->dbId = htobe64(db->dbId);
    db->vgVersion = htonl(db->vgVersion);
  }

  SKv kv = {.key = HEARTBEAT_KEY_DBINFO, .valueLen = sizeof(SDbVgVersion) * dbNum, .value = dbs};
D
dapan1121 已提交
260 261 262

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

D
dapan1121 已提交
263 264 265 266 267
  taosHashPut(req->info, &kv.key, sizeof(kv.key), &kv, sizeof(kv));

  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298
int32_t hbGetExpiredStbInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SClientHbReq *req) {
  SSTableMetaVersion *stbs = NULL;
  uint32_t stbNum = 0;
  int32_t code = 0;

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

  if (stbNum <= 0) {
    return TSDB_CODE_SUCCESS;
  }

  for (int32_t i = 0; i < stbNum; ++i) {
    SSTableMetaVersion *stb = &stbs[i];
    stb->suid = htobe64(stb->suid);
    stb->sversion = htons(stb->sversion);
    stb->tversion = htons(stb->tversion);    
  }

  SKv kv = {.key = HEARTBEAT_KEY_STBINFO, .valueLen = sizeof(SSTableMetaVersion) * stbNum, .value = stbs};

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

  taosHashPut(req->info, &kv.key, sizeof(kv.key), &kv, sizeof(kv));

  return TSDB_CODE_SUCCESS;
}


D
dapan1121 已提交
299 300 301 302 303 304 305 306 307 308 309 310 311 312 313
int32_t hbQueryHbReqHandle(SClientHbKey *connKey, void* param, SClientHbReq *req) {
  int64_t *clusterId = (int64_t *)param;
  struct SCatalog *pCatalog = NULL;

  int32_t code = catalogGetHandle(*clusterId, &pCatalog);
  if (code != TSDB_CODE_SUCCESS) {
    tscWarn("catalogGetHandle failed, clusterId:%"PRIx64", error:%s", *clusterId, tstrerror(code));
    return code;
  }
  
  code = hbGetExpiredDBInfo(connKey, pCatalog, req);
  if (TSDB_CODE_SUCCESS != code) {
    return code;
  }

D
dapan 已提交
314 315 316 317 318
  code = hbGetExpiredStbInfo(connKey, pCatalog, req);
  if (TSDB_CODE_SUCCESS != code) {
    return code;
  }

D
dapan1121 已提交
319 320 321 322 323 324 325 326 327 328 329 330 331

  return TSDB_CODE_SUCCESS;
}

int32_t hbMqHbReqHandle(SClientHbKey *connKey, void* param, SClientHbReq *req) {

}

void hbMgrInitMqHbHandle() {
  clientHbMgr.reqHandle[HEARTBEAT_TYPE_QUERY] = hbQueryHbReqHandle;
  clientHbMgr.reqHandle[HEARTBEAT_TYPE_MQ] = hbMqHbReqHandle;
  clientHbMgr.rspHandle[HEARTBEAT_TYPE_QUERY] = hbQueryHbRspHandle;
  clientHbMgr.rspHandle[HEARTBEAT_TYPE_MQ] = hbMqHbRspHandle;
L
Liu Jicong 已提交
332 333
}

L
Liu Jicong 已提交
334 335
static FORCE_INLINE void hbMgrInitHandle() {
  // init all handle
D
dapan1121 已提交
336
  hbMgrInitMqHbHandle();
L
Liu Jicong 已提交
337 338
}

D
dapan1121 已提交
339 340
void hbFreeReq(void *req) {
  SClientHbReq *pReq = (SClientHbReq *)req;
D
dapan1121 已提交
341
  tFreeReqKvHash(pReq->info);
D
dapan1121 已提交
342 343 344 345
}



L
Liu Jicong 已提交
346
SClientHbBatchReq* hbGatherAllInfo(SAppHbMgr *pAppHbMgr) {
L
Liu Jicong 已提交
347 348
  SClientHbBatchReq* pBatchReq = malloc(sizeof(SClientHbBatchReq));
  if (pBatchReq == NULL) {
L
Liu Jicong 已提交
349 350 351
    terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
    return NULL;
  }
L
Liu Jicong 已提交
352
  int32_t connKeyCnt = atomic_load_32(&pAppHbMgr->connKeyCnt);
L
Liu Jicong 已提交
353
  pBatchReq->reqs = taosArrayInit(connKeyCnt, sizeof(SClientHbReq));
L
Liu Jicong 已提交
354

D
dapan1121 已提交
355
  int32_t code = 0;
L
Liu Jicong 已提交
356
  void *pIter = taosHashIterate(pAppHbMgr->activeInfo, NULL);
L
Liu Jicong 已提交
357 358
  while (pIter != NULL) {
    SClientHbReq* pOneReq = pIter;
D
dapan1121 已提交
359 360 361 362 363 364 365 366 367 368

    SHbConnInfo * info = taosHashGet(pAppHbMgr->connInfo, &pOneReq->connKey, sizeof(SClientHbKey));
    if (info) {
      code = (*clientHbMgr.reqHandle[pOneReq->connKey.hbType])(&pOneReq->connKey, info->param, pOneReq);
      if (code) {
        taosHashCancelIterate(pAppHbMgr->activeInfo, pIter);
        break;
      }
    }

L
Liu Jicong 已提交
369
    taosArrayPush(pBatchReq->reqs, pOneReq);
L
Liu Jicong 已提交
370

L
Liu Jicong 已提交
371
    pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter);
L
Liu Jicong 已提交
372 373
  }

D
dapan1121 已提交
374 375 376
  if (code) {
    taosArrayDestroyEx(pBatchReq->reqs, hbFreeReq);
    tfree(pBatchReq);
L
Liu Jicong 已提交
377 378
  }

L
Liu Jicong 已提交
379
  return pBatchReq;
L
Liu Jicong 已提交
380 381
}

D
dapan1121 已提交
382 383 384 385 386 387

void hbClearReqInfo(SAppHbMgr *pAppHbMgr) {
  void *pIter = taosHashIterate(pAppHbMgr->activeInfo, NULL);
  while (pIter != NULL) {
    SClientHbReq* pOneReq = pIter;

D
dapan1121 已提交
388
    tFreeReqKvHash(pOneReq->info);
D
dapan1121 已提交
389 390 391 392 393 394 395 396
    taosHashClear(pOneReq->info);

    pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter);
  }
}



L
Liu Jicong 已提交
397 398 399
static void* hbThreadFunc(void* param) {
  setThreadName("hb");
  while (1) {
400 401
    int8_t threadStop = atomic_val_compare_exchange_8(&clientHbMgr.threadStop, 1, 2);
    if(1 == threadStop) {
L
Liu Jicong 已提交
402 403 404
      break;
    }

405 406
    pthread_mutex_lock(&clientHbMgr.lock);

L
Liu Jicong 已提交
407 408
    int sz = taosArrayGetSize(clientHbMgr.appHbMgrs);
    for(int i = 0; i < sz; i++) {
L
Liu Jicong 已提交
409 410
      SAppHbMgr* pAppHbMgr = taosArrayGetP(clientHbMgr.appHbMgrs, i);

L
Liu Jicong 已提交
411 412 413 414
      int32_t connCnt = atomic_load_32(&pAppHbMgr->connKeyCnt);
      if (connCnt == 0) {
        continue;
      }
L
Liu Jicong 已提交
415
      SClientHbBatchReq* pReq = hbGatherAllInfo(pAppHbMgr);
L
Liu Jicong 已提交
416 417 418
      if (pReq == NULL) {
        continue;
      }
L
Liu Jicong 已提交
419 420 421
      int tlen = tSerializeSClientHbBatchReq(NULL, pReq);
      void *buf = malloc(tlen);
      if (buf == NULL) {
D
dapan1121 已提交
422 423 424
        terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
        tFreeClientHbBatchReq(pReq, false);
        hbClearReqInfo(pAppHbMgr);
L
Liu Jicong 已提交
425 426
        break;
      }
L
Liu Jicong 已提交
427 428
      void *abuf = buf;
      tSerializeSClientHbBatchReq(&abuf, pReq);
L
Liu Jicong 已提交
429 430 431
      SMsgSendInfo *pInfo = malloc(sizeof(SMsgSendInfo));
      if (pInfo == NULL) {
        terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
L
Liu Jicong 已提交
432
        tFreeClientHbBatchReq(pReq, false);
D
dapan1121 已提交
433
        hbClearReqInfo(pAppHbMgr);
L
Liu Jicong 已提交
434 435 436 437 438 439 440
        free(buf);
        break;
      }
      pInfo->fp = hbMqAsyncCallBack;
      pInfo->msgInfo.pData = buf;
      pInfo->msgInfo.len = tlen;
      pInfo->msgType = TDMT_MND_HEARTBEAT;
D
dapan1121 已提交
441
      pInfo->param = strdup(pAppHbMgr->key);
L
Liu Jicong 已提交
442 443
      pInfo->requestId = generateRequestId();
      pInfo->requestObjRefId = 0;
L
Liu Jicong 已提交
444 445

      SAppInstInfo *pAppInstInfo = pAppHbMgr->pAppInstInfo;
L
Liu Jicong 已提交
446
      int64_t transporterId = 0;
L
Liu Jicong 已提交
447
      SEpSet epSet = getEpSet_s(&pAppInstInfo->mgmtEp);
L
Liu Jicong 已提交
448
      asyncSendMsgToServer(pAppInstInfo->pTransporter, &epSet, &transporterId, pInfo);
D
dapan1121 已提交
449 450
      tFreeClientHbBatchReq(pReq, false);      
      hbClearReqInfo(pAppHbMgr);
L
Liu Jicong 已提交
451 452 453

      atomic_add_fetch_32(&pAppHbMgr->reportCnt, 1);
    }
454 455 456

    pthread_mutex_unlock(&clientHbMgr.lock);
    
L
Liu Jicong 已提交
457
    taosMsleep(HEARTBEAT_INTERVAL);
L
Liu Jicong 已提交
458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474
  }
  return NULL;
}

static int32_t hbCreateThread() {
  pthread_attr_t thAttr;
  pthread_attr_init(&thAttr);
  pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE);

  if (pthread_create(&clientHbMgr.thread, &thAttr, hbThreadFunc, NULL) != 0) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }
  pthread_attr_destroy(&thAttr);
  return 0;
}

L
Liu Jicong 已提交
475
static void hbStopThread() {
476
  if (atomic_val_compare_exchange_8(&clientHbMgr.threadStop, 0, 1)) {
D
dapan1121 已提交
477
    tscDebug("hb thread already stopped");
478 479 480 481 482 483
    return;
  }
  
  while (2 != atomic_load_8(&clientHbMgr.threadStop)) {
    usleep(10);
  }
D
dapan1121 已提交
484 485

  tscDebug("hb thread stopped");  
L
Liu Jicong 已提交
486 487
}

D
dapan1121 已提交
488
SAppHbMgr* appHbMgrInit(SAppInstInfo* pAppInstInfo, char *key) {
L
Liu Jicong 已提交
489
  hbMgrInit();
L
Liu Jicong 已提交
490 491 492 493 494 495 496
  SAppHbMgr* pAppHbMgr = malloc(sizeof(SAppHbMgr)); 
  if (pAppHbMgr == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }
  // init stat
  pAppHbMgr->startTime = taosGetTimestampMs();
L
Liu Jicong 已提交
497 498 499
  pAppHbMgr->connKeyCnt = 0;
  pAppHbMgr->reportCnt = 0;
  pAppHbMgr->reportBytes = 0;
D
dapan1121 已提交
500
  pAppHbMgr->key = strdup(key);
L
Liu Jicong 已提交
501

L
Liu Jicong 已提交
502 503
  // init app info
  pAppHbMgr->pAppInstInfo = pAppInstInfo;
L
Liu Jicong 已提交
504 505 506

  // init hash info
  pAppHbMgr->activeInfo = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
L
Liu Jicong 已提交
507 508 509 510 511 512

  if (pAppHbMgr->activeInfo == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    free(pAppHbMgr);
    return NULL;
  }
L
Liu Jicong 已提交
513 514
  pAppHbMgr->activeInfo->freeFp = tFreeClientHbReq;
  // init getInfoFunc
D
dapan1121 已提交
515
  pAppHbMgr->connInfo = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
L
Liu Jicong 已提交
516

D
dapan1121 已提交
517
  if (pAppHbMgr->connInfo == NULL) {
L
Liu Jicong 已提交
518 519 520 521 522
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    free(pAppHbMgr);
    return NULL;
  }

523
  pthread_mutex_lock(&clientHbMgr.lock);
L
Liu Jicong 已提交
524
  taosArrayPush(clientHbMgr.appHbMgrs, &pAppHbMgr);
525 526
  pthread_mutex_unlock(&clientHbMgr.lock);
  
L
Liu Jicong 已提交
527 528 529
  return pAppHbMgr;
}

D
dapan1121 已提交
530
void appHbMgrCleanup(void) {
L
Liu Jicong 已提交
531 532 533 534
  pthread_mutex_lock(&clientHbMgr.lock);

  int sz = taosArrayGetSize(clientHbMgr.appHbMgrs);
  for (int i = 0; i < sz; i++) {
L
Liu Jicong 已提交
535
    SAppHbMgr* pTarget = taosArrayGetP(clientHbMgr.appHbMgrs, i);
D
dapan1121 已提交
536 537 538 539
    taosHashCleanup(pTarget->activeInfo);
    pTarget->activeInfo = NULL;
    taosHashCleanup(pTarget->connInfo);
    pTarget->connInfo = NULL;
L
Liu Jicong 已提交
540 541 542 543 544 545
  }

  pthread_mutex_unlock(&clientHbMgr.lock);
}

int hbMgrInit() {
L
Liu Jicong 已提交
546 547 548 549
  // init once
  int8_t old = atomic_val_compare_exchange_8(&clientHbMgr.inited, 0, 1);
  if (old == 1) return 0;

L
Liu Jicong 已提交
550 551
  clientHbMgr.appHbMgrs = taosArrayInit(0, sizeof(void*));
  pthread_mutex_init(&clientHbMgr.lock, NULL);
L
Liu Jicong 已提交
552 553 554 555 556

  // init handle funcs
  hbMgrInitHandle();

  // init backgroud thread
L
Liu Jicong 已提交
557 558
  hbCreateThread();

L
Liu Jicong 已提交
559 560 561 562
  return 0;
}

void hbMgrCleanUp() {
L
Liu Jicong 已提交
563
  return;
564 565
  hbStopThread();
  
L
Liu Jicong 已提交
566
  // destroy all appHbMgr
L
Liu Jicong 已提交
567 568 569
  int8_t old = atomic_val_compare_exchange_8(&clientHbMgr.inited, 1, 0);
  if (old == 0) return;

570
  pthread_mutex_lock(&clientHbMgr.lock);
D
dapan1121 已提交
571
  appHbMgrCleanup();
572 573 574 575
  taosArrayDestroy(clientHbMgr.appHbMgrs);  
  pthread_mutex_unlock(&clientHbMgr.lock);
  
  clientHbMgr.appHbMgrs = NULL;
L
Liu Jicong 已提交
576 577
}

D
dapan1121 已提交
578
int hbRegisterConnImpl(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, SHbConnInfo *info) {
L
Liu Jicong 已提交
579
  // init hash in activeinfo
L
Liu Jicong 已提交
580
  void* data = taosHashGet(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey));
L
Liu Jicong 已提交
581 582 583 584 585 586
  if (data != NULL) {
    return 0;
  }
  SClientHbReq hbReq;
  hbReq.connKey = connKey;
  hbReq.info = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
D
dapan1121 已提交
587
  
L
Liu Jicong 已提交
588
  taosHashPut(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey), &hbReq, sizeof(SClientHbReq));
D
dapan1121 已提交
589
 
L
Liu Jicong 已提交
590
  // init hash
D
dapan1121 已提交
591 592 593 594
  if (info != NULL) {
    SClientHbReq * pReq = taosHashGet(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey));
    info->req = pReq;
    taosHashPut(pAppHbMgr->connInfo, &connKey, sizeof(SClientHbKey), info, sizeof(SHbConnInfo));
L
Liu Jicong 已提交
595 596
  }

L
Liu Jicong 已提交
597
  atomic_add_fetch_32(&pAppHbMgr->connKeyCnt, 1);
L
Liu Jicong 已提交
598 599 600
  return 0;
}

D
dapan1121 已提交
601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622
int hbRegisterConn(SAppHbMgr* pAppHbMgr, int32_t connId, int64_t clusterId, int32_t hbType) {
  SClientHbKey connKey = {.connId = connId, .hbType = HEARTBEAT_TYPE_QUERY};
  SHbConnInfo info = {0};

  switch (hbType) {
    case HEARTBEAT_TYPE_QUERY: {
      int64_t *pClusterId = malloc(sizeof(int64_t));
      *pClusterId = clusterId;

      info.param = pClusterId;
      break;
    }
    case HEARTBEAT_TYPE_MQ: {
      break;
    }
    default:
      break;
  }
  
  return hbRegisterConnImpl(pAppHbMgr, connKey, &info);
}

L
Liu Jicong 已提交
623
void hbDeregisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey) {
D
dapan1121 已提交
624 625 626 627 628 629
  int32_t code = 0;
  code = taosHashRemove(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey));
  code = taosHashRemove(pAppHbMgr->connInfo, &connKey, sizeof(SClientHbKey));
  if (code) {
    return;
  }
L
Liu Jicong 已提交
630
  atomic_sub_fetch_32(&pAppHbMgr->connKeyCnt, 1);
L
Liu Jicong 已提交
631 632
}

L
Liu Jicong 已提交
633
int hbAddConnInfo(SAppHbMgr *pAppHbMgr, SClientHbKey connKey, void* key, void* value, int32_t keyLen, int32_t valueLen) {
L
Liu Jicong 已提交
634
  // find req by connection id
L
Liu Jicong 已提交
635
  SClientHbReq* pReq = taosHashGet(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey));
L
Liu Jicong 已提交
636 637 638 639 640 641
  ASSERT(pReq != NULL);

  taosHashPut(pReq->info, key, keyLen, value, valueLen);

  return 0;
}