cliBench.c 6.3 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 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34
/*
 * 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 "os.h"
#include "taoserror.h"
#include "tglobal.h"
#include "transLog.h"
#include "trpc.h"
#include "tutil.h"

typedef struct {
  int      index;
  SEpSet   epSet;
  int      num;
  int      numOfReqs;
  int      msgSize;
  tsem_t   rspSem;
  tsem_t  *pOverSem;
  TdThread thread;
  void    *pRpc;
} SInfo;

G
bench  
gccgdb1234 已提交
35
void initLogEnv() {
dengyihao's avatar
dengyihao 已提交
36 37
  const char   *logDir = "/tmp/trans_cli";
  const char   *defaultLogFileNamePrefix = "taoslog";
G
bench  
gccgdb1234 已提交
38 39
  const int32_t maxLogFileNum = 10000;
  tsAsyncLog = 0;
dengyihao's avatar
dengyihao 已提交
40
  // idxDebugFlag = 143;
G
bench  
gccgdb1234 已提交
41 42
  strcpy(tsLogDir, (char *)logDir);
  taosRemoveDir(tsLogDir);
dengyihao's avatar
dengyihao 已提交
43 44
  taosMkDir(tsLogDir);

G
bench  
gccgdb1234 已提交
45
  if (taosInitLog(defaultLogFileNamePrefix, maxLogFileNum) < 0) {
dengyihao's avatar
dengyihao 已提交
46
    printf("failed to open log file in directory:%s\n", tsLogDir);
G
bench  
gccgdb1234 已提交
47 48
  }
}
dengyihao's avatar
dengyihao 已提交
49

dengyihao's avatar
dengyihao 已提交
50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73
static void processResponse(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
  SInfo *pInfo = (SInfo *)pMsg->info.ahandle;
  tDebug("thread:%d, response is received, type:%d contLen:%d code:0x%x", pInfo->index, pMsg->msgType, pMsg->contLen,
         pMsg->code);

  if (pEpSet) pInfo->epSet = *pEpSet;

  rpcFreeCont(pMsg->pCont);
  tsem_post(&pInfo->rspSem);
}

static int tcount = 0;

static void *sendRequest(void *param) {
  SInfo  *pInfo = (SInfo *)param;
  SRpcMsg rpcMsg = {0};

  tDebug("thread:%d, start to send request", pInfo->index);

  while (pInfo->numOfReqs == 0 || pInfo->num < pInfo->numOfReqs) {
    pInfo->num++;
    rpcMsg.pCont = rpcMallocCont(pInfo->msgSize);
    rpcMsg.contLen = pInfo->msgSize;
    rpcMsg.info.ahandle = pInfo;
dengyihao's avatar
dengyihao 已提交
74
    rpcMsg.info.noResp = 1;
dengyihao's avatar
dengyihao 已提交
75 76 77 78
    rpcMsg.msgType = 1;
    tDebug("thread:%d, send request, contLen:%d num:%d", pInfo->index, pInfo->msgSize, pInfo->num);
    rpcSendRequest(pInfo->pRpc, &pInfo->epSet, &rpcMsg, NULL);
    if (pInfo->num % 20000 == 0) tInfo("thread:%d, %d requests have been sent", pInfo->index, pInfo->num);
dengyihao's avatar
dengyihao 已提交
79
    // tsem_wait(&pInfo->rspSem);
dengyihao's avatar
dengyihao 已提交
80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114
  }

  tDebug("thread:%d, it is over", pInfo->index);
  tcount++;

  return NULL;
}

int main(int argc, char *argv[]) {
  SRpcInit       rpcInit;
  SEpSet         epSet;
  int            msgSize = 128;
  int            numOfReqs = 0;
  int            appThreads = 1;
  char           serverIp[40] = "127.0.0.1";
  struct timeval systemTime;
  int64_t        startTime, endTime;

  // server info
  epSet.numOfEps = 1;
  epSet.inUse = 0;
  epSet.eps[0].port = 7000;
  epSet.eps[1].port = 7000;
  strcpy(epSet.eps[0].fqdn, serverIp);
  strcpy(epSet.eps[1].fqdn, "192.168.0.1");

  // client info
  memset(&rpcInit, 0, sizeof(rpcInit));
  rpcInit.localPort = 0;
  rpcInit.label = "APP";
  rpcInit.numOfThreads = 1;
  rpcInit.cfp = processResponse;
  rpcInit.sessions = 100;
  rpcInit.idleTime = tsShellActivityTimer * 1000;
  rpcInit.user = "michael";
dengyihao's avatar
dengyihao 已提交
115

dengyihao's avatar
dengyihao 已提交
116
  rpcInit.connType = TAOS_CONN_CLIENT;
dengyihao's avatar
dengyihao 已提交
117 118 119
  rpcInit.connLimitNum = 300;
  rpcInit.connLimitLock = 1;
  rpcInit.supportBatch = 0;
dengyihao's avatar
dengyihao 已提交
120

G
bench  
gccgdb1234 已提交
121
  rpcDebugFlag = 135;
dengyihao's avatar
dengyihao 已提交
122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154
  for (int i = 1; i < argc; ++i) {
    if (strcmp(argv[i], "-p") == 0 && i < argc - 1) {
    } else if (strcmp(argv[i], "-i") == 0 && i < argc - 1) {
    } else if (strcmp(argv[i], "-t") == 0 && i < argc - 1) {
      rpcInit.numOfThreads = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-m") == 0 && i < argc - 1) {
      msgSize = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-s") == 0 && i < argc - 1) {
      rpcInit.sessions = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-n") == 0 && i < argc - 1) {
      numOfReqs = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-a") == 0 && i < argc - 1) {
      appThreads = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-o") == 0 && i < argc - 1) {
      tsCompressMsgSize = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-u") == 0 && i < argc - 1) {
    } else if (strcmp(argv[i], "-k") == 0 && i < argc - 1) {
    } else if (strcmp(argv[i], "-spi") == 0 && i < argc - 1) {
    } else if (strcmp(argv[i], "-d") == 0 && i < argc - 1) {
      rpcDebugFlag = atoi(argv[++i]);
    } else {
      printf("\nusage: %s [options] \n", argv[0]);
      printf("  [-i ip]: first server IP address, default is:%s\n", serverIp);
      printf("  [-t threads]: number of rpc threads, default is:%d\n", rpcInit.numOfThreads);
      printf("  [-m msgSize]: message body size, default is:%d\n", msgSize);
      printf("  [-a threads]: number of app threads, default is:%d\n", appThreads);
      printf("  [-n requests]: number of requests per thread, default is:%d\n", numOfReqs);
      printf("  [-u user]: user name for the connection, default is:%s\n", rpcInit.user);
      printf("  [-d debugFlag]: debug flag, default:%d\n", rpcDebugFlag);
      printf("  [-h help]: print out this help\n\n");
      exit(0);
    }
  }
G
bench  
gccgdb1234 已提交
155 156

  initLogEnv();
dengyihao's avatar
dengyihao 已提交
157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203

  void *pRpc = rpcOpen(&rpcInit);
  if (pRpc == NULL) {
    tError("failed to initialize RPC");
    return -1;
  }

  tInfo("client is initialized");
  tInfo("threads:%d msgSize:%d requests:%d", appThreads, msgSize, numOfReqs);

  int64_t now = taosGetTimestampUs();

  SInfo *pInfo = (SInfo *)taosMemoryCalloc(1, sizeof(SInfo) * appThreads);
  SInfo *p = pInfo;
  for (int i = 0; i < appThreads; ++i) {
    pInfo->index = i;
    pInfo->epSet = epSet;
    pInfo->numOfReqs = numOfReqs;
    pInfo->msgSize = msgSize;
    tsem_init(&pInfo->rspSem, 0, 0);
    pInfo->pRpc = pRpc;

    taosThreadCreate(&pInfo->thread, NULL, sendRequest, pInfo);
    pInfo++;
  }

  do {
    taosUsleep(1);
  } while (tcount < appThreads);

  float usedTime = (taosGetTimestampUs() - now) / 1000.0f;

  tInfo("it takes %.3f mseconds to send %d requests to server", usedTime, numOfReqs * appThreads);
  tInfo("Performance: %.3f requests per second, msgSize:%d bytes", 1000.0 * numOfReqs * appThreads / usedTime, msgSize);

  for (int i = 0; i < appThreads; i++) {
    SInfo *pInfo = p;
    taosThreadJoin(pInfo->thread, NULL);
    p++;
  }
  int ch = getchar();
  UNUSED(ch);

  taosCloseLog();

  return 0;
}