trans.c 5.6 KB
Newer Older
dengyihao's avatar
dengyihao 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
/*
 * 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/>.
 */

#ifdef USE_UV

#include "transComm.h"

20
void* (*taosInitHandle[])(uint32_t ip, uint32_t port, char* label, int32_t numOfThreads, void* fp, void* shandle) = {
U
ubuntu 已提交
21 22 23
    transInitServer, transInitClient};

void (*taosCloseHandle[])(void* arg) = {transCloseServer, transCloseClient};
dengyihao's avatar
dengyihao 已提交
24

dengyihao's avatar
dengyihao 已提交
25 26 27
void (*taosRefHandle[])(void* handle) = {transRefSrvHandle, transRefCliHandle};
void (*taosUnRefHandle[])(void* handle) = {transUnrefSrvHandle, transUnrefCliHandle};

dengyihao's avatar
dengyihao 已提交
28
int (*transReleaseHandle[])(void* handle) = {transReleaseSrvHandle, transReleaseCliHandle};
dengyihao's avatar
dengyihao 已提交
29

dengyihao's avatar
dengyihao 已提交
30 31
static int32_t transValidLocalFqdn(const char* localFqdn, uint32_t* ip) {
  *ip = taosGetIpv4FromFqdn(localFqdn);
dengyihao's avatar
dengyihao 已提交
32
  if (*ip == 0xFFFFFFFF) {
dengyihao's avatar
dengyihao 已提交
33
    terrno = TSDB_CODE_RPC_FQDN_ERROR;
dengyihao's avatar
dengyihao 已提交
34
    return -1;
dengyihao's avatar
dengyihao 已提交
35
  }
dengyihao's avatar
dengyihao 已提交
36
  return 0;
dengyihao's avatar
dengyihao 已提交
37
}
dengyihao's avatar
dengyihao 已提交
38
void* rpcOpen(const SRpcInit* pInit) {
dengyihao's avatar
dengyihao 已提交
39
  rpcInit();
dengyihao's avatar
dengyihao 已提交
40

wafwerar's avatar
wafwerar 已提交
41
  SRpcInfo* pRpc = taosMemoryCalloc(1, sizeof(SRpcInfo));
dengyihao's avatar
dengyihao 已提交
42 43 44 45
  if (pRpc == NULL) {
    return NULL;
  }
  if (pInit->label) {
dengyihao's avatar
dengyihao 已提交
46
    tstrncpy(pRpc->label, pInit->label, sizeof(pRpc->label));
dengyihao's avatar
dengyihao 已提交
47
  }
dengyihao's avatar
dengyihao 已提交
48 49

  pRpc->compressSize = pInit->compressSize;
dengyihao's avatar
dengyihao 已提交
50 51 52 53
  if (pRpc->compressSize < 0) {
    pRpc->compressSize = -1;
  }

dengyihao's avatar
dengyihao 已提交
54 55
  pRpc->encryption = pInit->encryption;

dengyihao's avatar
dengyihao 已提交
56 57 58 59 60
  pRpc->retryMinInterval = pInit->retryMinInterval;  // retry init interval
  pRpc->retryStepFactor = pInit->retryStepFactor;
  pRpc->retryMaxInterval = pInit->retryMaxInterval;
  pRpc->retryMaxTimouet = pInit->retryMaxTimouet;

dengyihao's avatar
dengyihao 已提交
61 62 63
  pRpc->failFastThreshold = pInit->failFastThreshold;
  pRpc->failFastInterval = pInit->failFastInterval;

dengyihao's avatar
dengyihao 已提交
64
  // register callback handle
dengyihao's avatar
dengyihao 已提交
65
  pRpc->cfp = pInit->cfp;
dengyihao's avatar
dengyihao 已提交
66
  pRpc->retry = pInit->rfp;
dengyihao's avatar
dengyihao 已提交
67
  pRpc->startTimer = pInit->tfp;
dengyihao's avatar
dengyihao 已提交
68
  pRpc->destroyFp = pInit->dfp;
dengyihao's avatar
dengyihao 已提交
69
  pRpc->failFastFp = pInit->ffp;
dengyihao's avatar
dengyihao 已提交
70

dengyihao's avatar
dengyihao 已提交
71
  pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads;
dengyihao's avatar
dengyihao 已提交
72 73 74
  if (pRpc->numOfThreads <= 0) {
    pRpc->numOfThreads = 1;
  }
dengyihao's avatar
fix bug  
dengyihao 已提交
75

76 77
  uint32_t ip = 0;
  if (pInit->connType == TAOS_CONN_SERVER) {
dengyihao's avatar
dengyihao 已提交
78
    if (transValidLocalFqdn(pInit->localFqdn, &ip) != 0) {
S
Shengliang Guan 已提交
79
      tError("invalid fqdn:%s, errmsg:%s", pInit->localFqdn, terrstr());
80 81 82 83 84
      taosMemoryFree(pRpc);
      return NULL;
    }
  }

dengyihao's avatar
dengyihao 已提交
85
  pRpc->connType = pInit->connType;
86
  pRpc->idleTime = pInit->idleTime;
87 88
  pRpc->tcphandle =
      (*taosInitHandle[pRpc->connType])(ip, pInit->localPort, pRpc->label, pRpc->numOfThreads, NULL, pRpc);
dengyihao's avatar
dengyihao 已提交
89

dengyihao's avatar
dengyihao 已提交
90 91 92 93
  if (pRpc->tcphandle == NULL) {
    taosMemoryFree(pRpc);
    return NULL;
  }
dengyihao's avatar
dengyihao 已提交
94
  pRpc->parent = pInit->parent;
dengyihao's avatar
dengyihao 已提交
95
  if (pInit->user) {
dengyihao's avatar
dengyihao 已提交
96
    tstrncpy(pRpc->user, pInit->user, sizeof(pRpc->user));
dengyihao's avatar
dengyihao 已提交
97
  }
dengyihao's avatar
dengyihao 已提交
98 99 100 101

  int64_t refId = transAddExHandle(transGetInstMgt(), pRpc);
  transAcquireExHandle(transGetInstMgt(), refId);
  pRpc->refId = refId;
dengyihao's avatar
dengyihao 已提交
102
  return (void*)refId;
dengyihao's avatar
dengyihao 已提交
103
}
dengyihao's avatar
dengyihao 已提交
104
void rpcClose(void* arg) {
dengyihao's avatar
dengyihao 已提交
105
  tInfo("start to close rpc");
dengyihao's avatar
dengyihao 已提交
106 107
  transRemoveExHandle(transGetInstMgt(), (int64_t)arg);
  transReleaseExHandle(transGetInstMgt(), (int64_t)arg);
dengyihao's avatar
dengyihao 已提交
108
  tInfo("end to close rpc");
dengyihao's avatar
dengyihao 已提交
109 110 111
  return;
}
void rpcCloseImpl(void* arg) {
dengyihao's avatar
dengyihao 已提交
112 113
  SRpcInfo* pRpc = (SRpcInfo*)arg;
  (*taosCloseHandle[pRpc->connType])(pRpc->tcphandle);
wafwerar's avatar
wafwerar 已提交
114
  taosMemoryFree(pRpc);
dengyihao's avatar
dengyihao 已提交
115
}
dengyihao's avatar
dengyihao 已提交
116

wafwerar's avatar
wafwerar 已提交
117 118
void* rpcMallocCont(int64_t contLen) {
  int64_t size = contLen + TRANS_MSG_OVERHEAD;
119
  char*   start = taosMemoryCalloc(1, size);
dengyihao's avatar
dengyihao 已提交
120
  if (start == NULL) {
121
    tError("failed to malloc msg, size:%" PRId64, size);
122
    terrno = TSDB_CODE_OUT_OF_MEMORY;
dengyihao's avatar
dengyihao 已提交
123 124
    return NULL;
  } else {
125
    tTrace("malloc mem:%p size:%" PRId64, start, size);
dengyihao's avatar
dengyihao 已提交
126
  }
127

dengyihao's avatar
dengyihao 已提交
128
  return start + sizeof(STransMsgHead);
dengyihao's avatar
dengyihao 已提交
129
}
dengyihao's avatar
dengyihao 已提交
130

131 132
void rpcFreeCont(void* cont) {
  if (cont == NULL) return;
wafwerar's avatar
wafwerar 已提交
133
  taosMemoryFree((char*)cont - TRANS_MSG_OVERHEAD);
D
dapan1121 已提交
134
  tTrace("rpc free cont:%p", (char*)cont - TRANS_MSG_OVERHEAD);
dengyihao's avatar
dengyihao 已提交
135
}
136

wafwerar's avatar
wafwerar 已提交
137
void* rpcReallocCont(void* ptr, int64_t contLen) {
138 139 140
  if (ptr == NULL) return rpcMallocCont(contLen);

  char*   st = (char*)ptr - TRANS_MSG_OVERHEAD;
wafwerar's avatar
wafwerar 已提交
141
  int64_t sz = contLen + TRANS_MSG_OVERHEAD;
wafwerar's avatar
wafwerar 已提交
142
  st = taosMemoryRealloc(st, sz);
dengyihao's avatar
dengyihao 已提交
143
  if (st == NULL) {
144
    terrno = TSDB_CODE_OUT_OF_MEMORY;
dengyihao's avatar
dengyihao 已提交
145 146
    return NULL;
  }
147

dengyihao's avatar
dengyihao 已提交
148 149 150
  return st + TRANS_MSG_OVERHEAD;
}

dengyihao's avatar
dengyihao 已提交
151 152
int rpcSendRequest(void* shandle, const SEpSet* pEpSet, SRpcMsg* pMsg, int64_t* pRid) {
  return transSendRequest(shandle, pEpSet, pMsg, NULL);
dengyihao's avatar
dengyihao 已提交
153
}
dengyihao's avatar
dengyihao 已提交
154 155
int rpcSendRequestWithCtx(void* shandle, const SEpSet* pEpSet, SRpcMsg* pMsg, int64_t* pRid, SRpcCtx* pCtx) {
  return transSendRequest(shandle, pEpSet, pMsg, pCtx);
U
ubuntu 已提交
156
}
dengyihao's avatar
dengyihao 已提交
157 158
int rpcSendRecv(void* shandle, SEpSet* pEpSet, SRpcMsg* pMsg, SRpcMsg* pRsp) {
  return transSendRecv(shandle, pEpSet, pMsg, pRsp);
U
ubuntu 已提交
159 160
}

dengyihao's avatar
dengyihao 已提交
161
int rpcSendResponse(const SRpcMsg* pMsg) { return transSendResponse(pMsg); }
162

dengyihao's avatar
dengyihao 已提交
163
void rpcRefHandle(void* handle, int8_t type) { (*taosRefHandle[type])(handle); }
U
ubuntu 已提交
164

dengyihao's avatar
dengyihao 已提交
165
void rpcUnrefHandle(void* handle, int8_t type) { (*taosUnRefHandle[type])(handle); }
dengyihao's avatar
dengyihao 已提交
166

dengyihao's avatar
dengyihao 已提交
167
int rpcRegisterBrokenLinkArg(SRpcMsg* msg) { return transRegisterMsg(msg); }
dengyihao's avatar
dengyihao 已提交
168
int rpcReleaseHandle(void* handle, int8_t type) { return (*transReleaseHandle[type])(handle); }
dengyihao's avatar
dengyihao 已提交
169

dengyihao's avatar
dengyihao 已提交
170
int rpcSetDefaultAddr(void* thandle, const char* ip, const char* fqdn) {
dengyihao's avatar
dengyihao 已提交
171
  // later
dengyihao's avatar
dengyihao 已提交
172
  return transSetDefaultAddr(thandle, ip, fqdn);
dengyihao's avatar
dengyihao 已提交
173
}
dengyihao's avatar
dengyihao 已提交
174

D
dapan1121 已提交
175
void* rpcAllocHandle() { return (void*)transAllocHandle(); }
dengyihao's avatar
dengyihao 已提交
176

dengyihao's avatar
dengyihao 已提交
177
int32_t rpcInit() {
dengyihao's avatar
dengyihao 已提交
178
  transInit();
dengyihao's avatar
dengyihao 已提交
179 180 181
  return 0;
}
void rpcCleanup(void) {
dengyihao's avatar
dengyihao 已提交
182
  transCleanup();
dengyihao's avatar
dengyihao 已提交
183 184
  transHttpEnvDestroy();

dengyihao's avatar
dengyihao 已提交
185 186 187
  return;
}

dengyihao's avatar
dengyihao 已提交
188
#endif