transComm.c 12.4 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
/*
 * 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"

dengyihao's avatar
dengyihao 已提交
19 20
// static TdThreadOnce transModuleInit = PTHREAD_ONCE_INIT;

dengyihao's avatar
dengyihao 已提交
21 22 23 24 25 26 27 28 29 30 31 32 33 34
int transAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey) {
  T_MD5_CTX context;
  int       ret = -1;

  tMD5Init(&context);
  tMD5Update(&context, (uint8_t*)pKey, TSDB_PASSWORD_LEN);
  tMD5Update(&context, (uint8_t*)pMsg, msgLen);
  tMD5Update(&context, (uint8_t*)pKey, TSDB_PASSWORD_LEN);
  tMD5Final(&context);

  if (memcmp(context.digest, pAuth, sizeof(context.digest)) == 0) ret = 0;

  return ret;
}
dengyihao's avatar
dengyihao 已提交
35

dengyihao's avatar
dengyihao 已提交
36 37 38 39 40 41 42 43 44 45 46
void transBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey) {
  T_MD5_CTX context;

  tMD5Init(&context);
  tMD5Update(&context, (uint8_t*)pKey, TSDB_PASSWORD_LEN);
  tMD5Update(&context, (uint8_t*)pMsg, msgLen);
  tMD5Update(&context, (uint8_t*)pKey, TSDB_PASSWORD_LEN);
  tMD5Final(&context);

  memcpy(pAuth, context.digest, sizeof(context.digest));
}
dengyihao's avatar
dengyihao 已提交
47

dengyihao's avatar
dengyihao 已提交
48
bool transCompressMsg(char* msg, int32_t len, int32_t* flen) {
dengyihao's avatar
dengyihao 已提交
49
  return false;
dengyihao's avatar
dengyihao 已提交
50 51 52 53 54 55 56
  // SRpcHead* pHead = rpcHeadFromCont(pCont);
  bool succ = false;
  int  overhead = sizeof(STransCompMsg);
  if (!NEEDTO_COMPRESSS_MSG(len)) {
    return succ;
  }

wafwerar's avatar
wafwerar 已提交
57
  char* buf = taosMemoryMalloc(len + overhead + 8);  // 8 extra bytes
dengyihao's avatar
dengyihao 已提交
58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82
  if (buf == NULL) {
    tError("failed to allocate memory for rpc msg compression, contLen:%d", len);
    *flen = len;
    return succ;
  }

  int32_t clen = LZ4_compress_default(msg, buf, len, len + overhead);
  tDebug("compress rpc msg, before:%d, after:%d, overhead:%d", len, clen, overhead);
  /*
   * only the compressed size is less than the value of contLen - overhead, the compression is applied
   * The first four bytes is set to 0, the second four bytes are utilized to keep the original length of message
   */
  if (clen > 0 && clen < len - overhead) {
    STransCompMsg* pComp = (STransCompMsg*)msg;
    pComp->reserved = 0;
    pComp->contLen = htonl(len);
    memcpy(msg + overhead, buf, clen);

    tDebug("compress rpc msg, before:%d, after:%d", len, clen);
    *flen = clen + overhead;
    succ = true;
  } else {
    *flen = len;
    succ = false;
  }
wafwerar's avatar
wafwerar 已提交
83
  taosMemoryFree(buf);
dengyihao's avatar
dengyihao 已提交
84 85 86 87 88 89 90 91 92 93 94 95
  return succ;
}
bool transDecompressMsg(char* msg, int32_t len, int32_t* flen) {
  // impl later
  return false;
  STransCompMsg* pComp = (STransCompMsg*)msg;

  int overhead = sizeof(STransCompMsg);
  int clen = 0;
  return false;
}

dengyihao's avatar
dengyihao 已提交
96 97 98 99
void transFreeMsg(void* msg) {
  if (msg == NULL) {
    return;
  }
wafwerar's avatar
wafwerar 已提交
100
  taosMemoryFree((char*)msg - sizeof(STransMsgHead));
dengyihao's avatar
dengyihao 已提交
101
}
dengyihao's avatar
dengyihao 已提交
102 103 104 105 106 107 108

int transInitBuffer(SConnBuffer* buf) {
  transClearBuffer(buf);
  return 0;
}
int transClearBuffer(SConnBuffer* buf) {
  memset(buf, 0, sizeof(*buf));
dengyihao's avatar
fix bug  
dengyihao 已提交
109
  buf->total = -1;
dengyihao's avatar
dengyihao 已提交
110 111 112 113 114 115
  return 0;
}
int transAllocBuffer(SConnBuffer* connBuf, uv_buf_t* uvBuf) {
  /*
   * formate of data buffer:
   * |<--------------------------data from socket------------------------------->|
dengyihao's avatar
dengyihao 已提交
116 117
   * |<------STransMsgHead------->|<-------------------userdata--------------->|<-----auth data----->|<----user
   * info--->|
dengyihao's avatar
dengyihao 已提交
118
   */
dengyihao's avatar
fix bug  
dengyihao 已提交
119
  static const int CAPACITY = sizeof(STransMsgHead);
dengyihao's avatar
dengyihao 已提交
120 121 122

  SConnBuffer* p = connBuf;
  if (p->cap == 0) {
wafwerar's avatar
wafwerar 已提交
123
    p->buf = (char*)taosMemoryCalloc(CAPACITY, sizeof(char));
dengyihao's avatar
dengyihao 已提交
124 125
    p->len = 0;
    p->cap = CAPACITY;
dengyihao's avatar
dengyihao 已提交
126
    p->total = -1;
dengyihao's avatar
dengyihao 已提交
127 128 129

    uvBuf->base = p->buf;
    uvBuf->len = CAPACITY;
dengyihao's avatar
dengyihao 已提交
130 131 132
  } else if (p->total == -1 && p->len < CAPACITY) {
    uvBuf->base = p->buf + p->len;
    uvBuf->len = CAPACITY - p->len;
dengyihao's avatar
dengyihao 已提交
133
  } else {
dengyihao's avatar
dengyihao 已提交
134
    p->cap = p->total;
wafwerar's avatar
wafwerar 已提交
135
    p->buf = taosMemoryRealloc(p->buf, p->cap);
dengyihao's avatar
dengyihao 已提交
136
    tTrace("internal malloc mem: %p, size: %d", p->buf, p->cap);
dengyihao's avatar
dengyihao 已提交
137

dengyihao's avatar
dengyihao 已提交
138 139 140 141 142
    uvBuf->base = p->buf + p->len;
    uvBuf->len = p->cap - p->len;
  }
  return 0;
}
dengyihao's avatar
dengyihao 已提交
143 144 145 146 147 148 149 150 151 152 153 154 155
// check whether already read complete
bool transReadComplete(SConnBuffer* connBuf) {
  if (connBuf->total == -1 && connBuf->len >= sizeof(STransMsgHead)) {
    STransMsgHead head;
    memcpy((char*)&head, connBuf->buf, sizeof(head));
    int32_t msgLen = (int32_t)htonl(head.msgLen);
    connBuf->total = msgLen;
  }
  if (connBuf->len == connBuf->cap && connBuf->total == connBuf->cap) {
    return true;
  }
  return false;
}
dengyihao's avatar
dengyihao 已提交
156
int transPackMsg(STransMsgHead* msgHead, bool sercured, bool auth) { return 0; }
dengyihao's avatar
dengyihao 已提交
157

dengyihao's avatar
dengyihao 已提交
158
int transUnpackMsg(STransMsgHead* msgHead) { return 0; }
dengyihao's avatar
dengyihao 已提交
159 160
int transDestroyBuffer(SConnBuffer* buf) {
  if (buf->cap > 0) {
wafwerar's avatar
wafwerar 已提交
161
    taosMemoryFreeClear(buf->buf);
dengyihao's avatar
dengyihao 已提交
162 163
  }
  transClearBuffer(buf);
164 165

  return 0;
dengyihao's avatar
dengyihao 已提交
166
}
dengyihao's avatar
dengyihao 已提交
167

dengyihao's avatar
dengyihao 已提交
168 169 170 171 172 173
int transSetConnOption(uv_tcp_t* stream) {
  uv_tcp_nodelay(stream, 1);
  int ret = uv_tcp_keepalive(stream, 5, 5);
  return ret;
}

dengyihao's avatar
dengyihao 已提交
174
SAsyncPool* transCreateAsyncPool(uv_loop_t* loop, int sz, void* arg, AsyncCB cb) {
wafwerar's avatar
wafwerar 已提交
175
  SAsyncPool* pool = taosMemoryCalloc(1, sizeof(SAsyncPool));
dengyihao's avatar
dengyihao 已提交
176 177
  pool->index = 0;
  pool->nAsync = sz;
wafwerar's avatar
wafwerar 已提交
178
  pool->asyncs = taosMemoryCalloc(1, sizeof(uv_async_t) * pool->nAsync);
dengyihao's avatar
dengyihao 已提交
179 180 181 182

  for (int i = 0; i < pool->nAsync; i++) {
    uv_async_t* async = &(pool->asyncs[i]);
    uv_async_init(loop, async, cb);
dengyihao's avatar
dengyihao 已提交
183

wafwerar's avatar
wafwerar 已提交
184
    SAsyncItem* item = taosMemoryCalloc(1, sizeof(SAsyncItem));
dengyihao's avatar
dengyihao 已提交
185 186
    item->pThrd = arg;
    QUEUE_INIT(&item->qmsg);
wafwerar's avatar
wafwerar 已提交
187
    taosThreadMutexInit(&item->mtx, NULL);
dengyihao's avatar
dengyihao 已提交
188 189

    async->data = item;
dengyihao's avatar
dengyihao 已提交
190 191 192
  }
  return pool;
}
dengyihao's avatar
dengyihao 已提交
193

dengyihao's avatar
dengyihao 已提交
194 195 196
void transDestroyAsyncPool(SAsyncPool* pool) {
  for (int i = 0; i < pool->nAsync; i++) {
    uv_async_t* async = &(pool->asyncs[i]);
dengyihao's avatar
dengyihao 已提交
197
    // uv_close((uv_handle_t*)async, NULL);
dengyihao's avatar
dengyihao 已提交
198
    SAsyncItem* item = async->data;
wafwerar's avatar
wafwerar 已提交
199
    taosThreadMutexDestroy(&item->mtx);
wafwerar's avatar
wafwerar 已提交
200
    taosMemoryFree(item);
dengyihao's avatar
dengyihao 已提交
201
  }
wafwerar's avatar
wafwerar 已提交
202 203
  taosMemoryFree(pool->asyncs);
  taosMemoryFree(pool);
dengyihao's avatar
dengyihao 已提交
204
}
dengyihao's avatar
dengyihao 已提交
205
int transSendAsync(SAsyncPool* pool, queue* q) {
dengyihao's avatar
dengyihao 已提交
206 207 208 209 210 211
  int idx = pool->index;
  idx = idx % pool->nAsync;
  // no need mutex here
  if (pool->index++ > pool->nAsync) {
    pool->index = 0;
  }
dengyihao's avatar
dengyihao 已提交
212 213 214 215
  uv_async_t* async = &(pool->asyncs[idx]);
  SAsyncItem* item = async->data;

  int64_t st = taosGetTimestampUs();
wafwerar's avatar
wafwerar 已提交
216
  taosThreadMutexLock(&item->mtx);
dengyihao's avatar
dengyihao 已提交
217
  QUEUE_PUSH(&item->qmsg, q);
wafwerar's avatar
wafwerar 已提交
218
  taosThreadMutexUnlock(&item->mtx);
dengyihao's avatar
dengyihao 已提交
219 220 221 222 223
  int64_t el = taosGetTimestampUs() - st;
  if (el > 50) {
    // tInfo("lock and unlock cost: %d", (int)el);
  }
  return uv_async_send(async);
dengyihao's avatar
dengyihao 已提交
224
}
dengyihao's avatar
dengyihao 已提交
225

dengyihao's avatar
dengyihao 已提交
226 227 228 229
void transCtxInit(STransCtx* ctx) {
  // init transCtx
  ctx->args = taosHashInit(2, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UINT), true, HASH_NO_LOCK);
}
dengyihao's avatar
dengyihao 已提交
230
void transCtxCleanup(STransCtx* ctx) {
dengyihao's avatar
dengyihao 已提交
231 232 233 234 235 236
  if (ctx->args == NULL) {
    return;
  }

  STransCtxVal* iter = taosHashIterate(ctx->args, NULL);
  while (iter) {
dengyihao's avatar
dengyihao 已提交
237
    ctx->freeFunc(iter->val);
dengyihao's avatar
dengyihao 已提交
238 239
    iter = taosHashIterate(ctx->args, iter);
  }
dengyihao's avatar
dengyihao 已提交
240

dengyihao's avatar
dengyihao 已提交
241
  taosHashCleanup(ctx->args);
U
ubuntu 已提交
242
  ctx->args = NULL;
dengyihao's avatar
dengyihao 已提交
243 244 245 246 247
}

void transCtxMerge(STransCtx* dst, STransCtx* src) {
  if (dst->args == NULL) {
    dst->args = src->args;
dengyihao's avatar
dengyihao 已提交
248
    dst->brokenVal = src->brokenVal;
dengyihao's avatar
dengyihao 已提交
249
    dst->freeFunc = src->freeFunc;
dengyihao's avatar
dengyihao 已提交
250 251 252 253 254 255 256 257 258 259 260 261
    src->args = NULL;
    return;
  }
  void*  key = NULL;
  size_t klen = 0;
  void*  iter = taosHashIterate(src->args, NULL);
  while (iter) {
    STransCtxVal* sVal = (STransCtxVal*)iter;
    key = taosHashGetKey(sVal, &klen);

    STransCtxVal* dVal = taosHashGet(dst->args, key, klen);
    if (dVal) {
dengyihao's avatar
dengyihao 已提交
262
      dst->freeFunc(dVal->val);
dengyihao's avatar
dengyihao 已提交
263 264 265 266 267 268 269 270 271 272 273 274 275 276
    }
    taosHashPut(dst->args, key, klen, sVal, sizeof(*sVal));
    iter = taosHashIterate(src->args, iter);
  }
  taosHashCleanup(src->args);
}
void* transCtxDumpVal(STransCtx* ctx, int32_t key) {
  if (ctx->args == NULL) {
    return NULL;
  }
  STransCtxVal* cVal = taosHashGet(ctx->args, (const void*)&key, sizeof(key));
  if (cVal == NULL) {
    return NULL;
  }
dengyihao's avatar
dengyihao 已提交
277
  void* ret = NULL;
D
dapan1121 已提交
278 279
  (*cVal->clone)(cVal->val, &ret);
  return ret;
dengyihao's avatar
dengyihao 已提交
280
}
dengyihao's avatar
dengyihao 已提交
281
void* transCtxDumpBrokenlinkVal(STransCtx* ctx, int32_t* msgType) {
dengyihao's avatar
dengyihao 已提交
282
  void* ret = NULL;
dengyihao's avatar
dengyihao 已提交
283
  if (ctx->brokenVal.clone == NULL) {
dengyihao's avatar
dengyihao 已提交
284
    return ret;
dengyihao's avatar
dengyihao 已提交
285
  }
D
dapan1121 已提交
286
  (*ctx->brokenVal.clone)(ctx->brokenVal.val, &ret);
dengyihao's avatar
dengyihao 已提交
287 288 289

  *msgType = ctx->brokenVal.msgType;

D
dapan1121 已提交
290
  return ret;
dengyihao's avatar
dengyihao 已提交
291
}
dengyihao's avatar
dengyihao 已提交
292

wafwerar's avatar
wafwerar 已提交
293
void transQueueInit(STransQueue* queue, void (*freeFunc)(const void* arg)) {
dengyihao's avatar
dengyihao 已提交
294
  queue->q = taosArrayInit(2, sizeof(void*));
wafwerar's avatar
wafwerar 已提交
295
  queue->freeFunc = (void (*)(const void*))freeFunc;
dengyihao's avatar
dengyihao 已提交
296 297
}
bool transQueuePush(STransQueue* queue, void* arg) {
dengyihao's avatar
dengyihao 已提交
298 299 300
  if (queue->q == NULL) {
    return true;
  }
dengyihao's avatar
dengyihao 已提交
301 302 303 304 305 306 307
  taosArrayPush(queue->q, &arg);
  if (taosArrayGetSize(queue->q) > 1) {
    return false;
  }
  return true;
}
void* transQueuePop(STransQueue* queue) {
dengyihao's avatar
dengyihao 已提交
308
  if (queue->q == NULL || taosArrayGetSize(queue->q) == 0) {
dengyihao's avatar
dengyihao 已提交
309 310 311 312 313 314
    return NULL;
  }
  void* ptr = taosArrayGetP(queue->q, 0);
  taosArrayRemove(queue->q, 0);
  return ptr;
}
dengyihao's avatar
dengyihao 已提交
315
int32_t transQueueSize(STransQueue* queue) {
dengyihao's avatar
dengyihao 已提交
316 317 318
  if (queue->q == NULL) {
    return 0;
  }
dengyihao's avatar
dengyihao 已提交
319 320 321
  return taosArrayGetSize(queue->q);
}
void* transQueueGet(STransQueue* queue, int i) {
dengyihao's avatar
dengyihao 已提交
322
  if (queue->q == NULL || taosArrayGetSize(queue->q) == 0) {
dengyihao's avatar
dengyihao 已提交
323 324 325 326 327
    return NULL;
  }
  if (i >= taosArrayGetSize(queue->q)) {
    return NULL;
  }
dengyihao's avatar
dengyihao 已提交
328

dengyihao's avatar
dengyihao 已提交
329 330 331 332 333
  void* ptr = taosArrayGetP(queue->q, i);
  return ptr;
}

void* transQueueRm(STransQueue* queue, int i) {
dengyihao's avatar
dengyihao 已提交
334
  if (queue->q == NULL || taosArrayGetSize(queue->q) == 0) {
dengyihao's avatar
dengyihao 已提交
335 336
    return NULL;
  }
dengyihao's avatar
dengyihao 已提交
337 338 339 340 341
  if (i >= taosArrayGetSize(queue->q)) {
    return NULL;
  }
  void* ptr = taosArrayGetP(queue->q, i);
  taosArrayRemove(queue->q, i);
dengyihao's avatar
dengyihao 已提交
342 343
  return ptr;
}
dengyihao's avatar
dengyihao 已提交
344

dengyihao's avatar
dengyihao 已提交
345
bool transQueueEmpty(STransQueue* queue) {
dengyihao's avatar
dengyihao 已提交
346 347 348
  if (queue->q == NULL) {
    return true;
  }
dengyihao's avatar
dengyihao 已提交
349 350 351
  return taosArrayGetSize(queue->q) == 0;
}
void transQueueClear(STransQueue* queue) {
wafwerar's avatar
wafwerar 已提交
352
  if (queue->freeFunc != NULL) {
dengyihao's avatar
dengyihao 已提交
353 354
    for (int i = 0; i < taosArrayGetSize(queue->q); i++) {
      void* p = taosArrayGetP(queue->q, i);
wafwerar's avatar
wafwerar 已提交
355
      queue->freeFunc(p);
dengyihao's avatar
dengyihao 已提交
356 357 358 359 360 361 362 363
    }
  }
  taosArrayClear(queue->q);
}
void transQueueDestroy(STransQueue* queue) {
  transQueueClear(queue);
  taosArrayDestroy(queue->q);
}
dengyihao's avatar
dengyihao 已提交
364 365 366 367 368

static int32_t timeCompare(const HeapNode* a, const HeapNode* b) {
  SDelayTask* arg1 = container_of(a, SDelayTask, node);
  SDelayTask* arg2 = container_of(b, SDelayTask, node);
  if (arg1->execTime > arg2->execTime) {
dengyihao's avatar
dengyihao 已提交
369
    return 0;
dengyihao's avatar
dengyihao 已提交
370 371 372 373 374
  } else {
    return 1;
  }
}

dengyihao's avatar
dengyihao 已提交
375
static void transDQTimeout(uv_timer_t* timer) {
dengyihao's avatar
dengyihao 已提交
376
  SDelayQueue* queue = timer->data;
dengyihao's avatar
dengyihao 已提交
377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393
  tTrace("timer %p timeout", timer);
  uint64_t timeout = 0;
  do {
    HeapNode* minNode = heapMin(queue->heap);
    if (minNode == NULL) break;
    SDelayTask* task = container_of(minNode, SDelayTask, node);
    if (task->execTime <= taosGetTimestampMs()) {
      heapRemove(queue->heap, minNode);
      task->func(task->arg);
      taosMemoryFree(task);
      timeout = 0;
    } else {
      timeout = task->execTime - taosGetTimestampMs();
      break;
    }
  } while (1);
  if (timeout != 0) {
dengyihao's avatar
dengyihao 已提交
394
    uv_timer_start(queue->timer, transDQTimeout, timeout, 0);
dengyihao's avatar
dengyihao 已提交
395 396
  }
}
dengyihao's avatar
dengyihao 已提交
397
int transDQCreate(uv_loop_t* loop, SDelayQueue** queue) {
dengyihao's avatar
dengyihao 已提交
398 399 400 401 402 403 404 405 406 407 408 409 410 411 412
  uv_timer_t* timer = taosMemoryCalloc(1, sizeof(uv_timer_t));
  uv_timer_init(loop, timer);

  Heap* heap = heapCreate(timeCompare);

  SDelayQueue* q = taosMemoryCalloc(1, sizeof(SDelayQueue));
  q->heap = heap;
  q->timer = timer;
  q->loop = loop;
  q->timer->data = q;

  *queue = q;
  return 0;
}

dengyihao's avatar
dengyihao 已提交
413
void transDQDestroy(SDelayQueue* queue) {
dengyihao's avatar
dengyihao 已提交
414
  taosMemoryFree(queue->timer);
dengyihao's avatar
dengyihao 已提交
415 416 417 418 419 420 421 422 423 424 425

  while (heapSize(queue->heap) > 0) {
    HeapNode* minNode = heapMin(queue->heap);
    if (minNode == NULL) {
      return;
    }
    heapRemove(queue->heap, minNode);

    SDelayTask* task = container_of(minNode, SDelayTask, node);
    taosMemoryFree(task);
  }
dengyihao's avatar
dengyihao 已提交
426 427 428 429
  heapDestroy(queue->heap);
  taosMemoryFree(queue);
}

dengyihao's avatar
dengyihao 已提交
430
int transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg, uint64_t timeoutMs) {
dengyihao's avatar
dengyihao 已提交
431
  uint64_t    now = taosGetTimestampMs();
dengyihao's avatar
dengyihao 已提交
432 433 434
  SDelayTask* task = taosMemoryCalloc(1, sizeof(SDelayTask));
  task->func = func;
  task->arg = arg;
dengyihao's avatar
dengyihao 已提交
435 436 437 438 439 440
  task->execTime = now + timeoutMs;

  HeapNode* minNode = heapMin(queue->heap);
  if (minNode) {
    SDelayTask* minTask = container_of(minNode, SDelayTask, node);
    if (minTask->execTime < task->execTime) {
dengyihao's avatar
dengyihao 已提交
441
      timeoutMs = minTask->execTime <= now ? 0 : minTask->execTime - now;
dengyihao's avatar
dengyihao 已提交
442 443
    }
  }
dengyihao's avatar
dengyihao 已提交
444

dengyihao's avatar
dengyihao 已提交
445
  tTrace("timer %p put task into queue, timeoutMs: %" PRIu64 "", queue->timer, timeoutMs);
dengyihao's avatar
dengyihao 已提交
446
  heapInsert(queue->heap, &task->node);
dengyihao's avatar
dengyihao 已提交
447
  uv_timer_start(queue->timer, transDQTimeout, timeoutMs, 0);
dengyihao's avatar
dengyihao 已提交
448 449
  return 0;
}
dengyihao's avatar
dengyihao 已提交
450 451 452 453 454 455 456 457 458 459 460 461

void transPrintEpSet(SEpSet* pEpSet) {
  if (pEpSet == NULL) {
    tTrace("NULL epset");
    return;
  }
  tTrace("epset begin: inUse: %d", pEpSet->inUse);
  for (int i = 0; i < pEpSet->numOfEps; i++) {
    tTrace("ip: %s, port: %d", pEpSet->eps[i].fqdn, pEpSet->eps[i].port);
  }
  tTrace("epset end");
}
dengyihao's avatar
dengyihao 已提交
462
#endif