syncUtil.c 4.3 KB
Newer Older
M
Minghao Li 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
/*
 * 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 "syncUtil.h"
M
Minghao Li 已提交
17 18 19
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>
M
Minghao Li 已提交
20
#include "syncEnv.h"
M
Minghao Li 已提交
21

M
Minghao Li 已提交
22 23 24 25
// ---- encode / decode
uint64_t syncUtilAddr2U64(const char* host, uint16_t port) {
  uint64_t u64;
  uint32_t hostU32 = (uint32_t)inet_addr(host);
M
Minghao Li 已提交
26
  // assert(hostU32 != (uint32_t)-1);
M
Minghao Li 已提交
27 28 29
  u64 = (((uint64_t)hostU32) << 32) | (((uint32_t)port) << 16);
  return u64;
}
M
Minghao Li 已提交
30

M
Minghao Li 已提交
31 32
void syncUtilU642Addr(uint64_t u64, char* host, size_t len, uint16_t* port) {
  uint32_t hostU32 = (uint32_t)((u64 >> 32) & 0x00000000FFFFFFFF);
M
Minghao Li 已提交
33

M
Minghao Li 已提交
34 35 36 37 38 39 40 41
  struct in_addr addr;
  addr.s_addr = hostU32;
  snprintf(host, len, "%s", inet_ntoa(addr));
  *port = (uint16_t)((u64 & 0x00000000FFFF0000) >> 16);
}

void syncUtilnodeInfo2EpSet(const SNodeInfo* pNodeInfo, SEpSet* pEpSet) {
  pEpSet->inUse = 0;
M
Minghao Li 已提交
42
  pEpSet->numOfEps = 0;
M
Minghao Li 已提交
43 44 45 46 47 48 49 50
  addEpIntoEpSet(pEpSet, pNodeInfo->nodeFqdn, pNodeInfo->nodePort);
}

void syncUtilraftId2EpSet(const SRaftId* raftId, SEpSet* pEpSet) {
  char     host[TSDB_FQDN_LEN];
  uint16_t port;

  syncUtilU642Addr(raftId->addr, host, sizeof(host), &port);
M
sync io  
Minghao Li 已提交
51 52 53 54 55

  /*
    pEpSet->numOfEps = 1;
    pEpSet->inUse = 0;
    pEpSet->eps[0].port = port;
M
Minghao Li 已提交
56
    snprintf(pEpSet->eps[0].fqdn, sizeof(pEpSet->eps[0].fqdn), "%s", host);
M
sync io  
Minghao Li 已提交
57
  */
M
Minghao Li 已提交
58
  pEpSet->inUse = 0;
M
Minghao Li 已提交
59
  pEpSet->numOfEps = 0;
M
Minghao Li 已提交
60 61 62 63
  addEpIntoEpSet(pEpSet, host, port);
}

void syncUtilnodeInfo2raftId(const SNodeInfo* pNodeInfo, SyncGroupId vgId, SRaftId* raftId) {
M
sync io  
Minghao Li 已提交
64 65 66 67 68
  uint32_t ipv4 = taosGetIpv4FromFqdn(pNodeInfo->nodeFqdn);
  assert(ipv4 != 0xFFFFFFFF);
  char ipbuf[128];
  tinet_ntoa(ipbuf, ipv4);
  raftId->addr = syncUtilAddr2U64(ipbuf, pNodeInfo->nodePort);
M
Minghao Li 已提交
69 70 71
  raftId->vgId = vgId;
}

M
Minghao Li 已提交
72 73 74 75 76
bool syncUtilSameId(const SRaftId* pId1, const SRaftId* pId2) {
  bool ret = pId1->addr == pId2->addr && pId1->vgId == pId2->vgId;
  return ret;
}

M
Minghao Li 已提交
77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94
// ---- SSyncBuffer -----
void syncUtilbufBuild(SSyncBuffer* syncBuf, size_t len) {
  syncBuf->len = len;
  syncBuf->data = malloc(syncBuf->len);
}

void syncUtilbufDestroy(SSyncBuffer* syncBuf) { free(syncBuf->data); }

void syncUtilbufCopy(const SSyncBuffer* src, SSyncBuffer* dest) {
  dest->len = src->len;
  dest->data = src->data;
}

void syncUtilbufCopyDeep(const SSyncBuffer* src, SSyncBuffer* dest) {
  dest->len = src->len;
  dest->data = malloc(dest->len);
  memcpy(dest->data, src->data, dest->len);
}
M
Minghao Li 已提交
95 96 97 98 99

// ---- misc ----

int32_t syncUtilRand(int32_t max) { return rand() % max; }

M
Minghao Li 已提交
100 101
int32_t syncUtilElectRandomMS() { ELECT_TIMER_MS_MIN + syncUtilRand(ELECT_TIMER_MS_RANGE); }

M
Minghao Li 已提交
102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129
int32_t syncUtilQuorum(int32_t replicaNum) { return replicaNum / 2 + 1; }

cJSON* syncUtilNodeInfo2Json(const SNodeInfo* p) {
  char   u64buf[128];
  cJSON* pRoot = cJSON_CreateObject();

  cJSON_AddStringToObject(pRoot, "nodeFqdn", p->nodeFqdn);
  cJSON_AddNumberToObject(pRoot, "nodePort", p->nodePort);

  cJSON* pJson = cJSON_CreateObject();
  cJSON_AddItemToObject(pJson, "SNodeInfo", pRoot);
  return pJson;
}

cJSON* syncUtilRaftId2Json(const SRaftId* p) {
  char   u64buf[128];
  cJSON* pRoot = cJSON_CreateObject();

  snprintf(u64buf, sizeof(u64buf), "%lu", p->addr);
  cJSON_AddStringToObject(pRoot, "addr", u64buf);
  char     host[128];
  uint16_t port;
  syncUtilU642Addr(p->addr, host, sizeof(host), &port);
  cJSON_AddStringToObject(pRoot, "host", host);
  cJSON_AddNumberToObject(pRoot, "port", port);
  cJSON_AddNumberToObject(pRoot, "vgId", p->vgId);

  cJSON* pJson = cJSON_CreateObject();
M
Minghao Li 已提交
130
  cJSON_AddItemToObject(pJson, "SRaftId", pRoot);
M
Minghao Li 已提交
131 132 133 134 135 136 137 138 139 140 141 142 143 144
  return pJson;
}

const char* syncUtilState2String(ESyncState state) {
  if (state == TAOS_SYNC_STATE_FOLLOWER) {
    return "TAOS_SYNC_STATE_FOLLOWER";
  } else if (state == TAOS_SYNC_STATE_CANDIDATE) {
    return "TAOS_SYNC_STATE_CANDIDATE";
  } else if (state == TAOS_SYNC_STATE_LEADER) {
    return "TAOS_SYNC_STATE_LEADER";
  } else {
    return "TAOS_SYNC_STATE_UNKNOWN";
  }
}