clientEnv.c 17.2 KB
Newer Older
H
Haojun Liao 已提交
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/>.
 */

16
#include "catalog.h"
H
Haojun Liao 已提交
17 18
#include "clientInt.h"
#include "clientLog.h"
dengyihao's avatar
dengyihao 已提交
19 20
#include "functionMgt.h"
#include "os.h"
21
#include "query.h"
X
Xiaoyu Wang 已提交
22
#include "scheduler.h"
H
Haojun Liao 已提交
23 24
#include "tcache.h"
#include "tglobal.h"
dengyihao's avatar
dengyihao 已提交
25
#include "tmsg.h"
H
Haojun Liao 已提交
26 27
#include "tref.h"
#include "trpc.h"
dengyihao's avatar
dengyihao 已提交
28
#include "tsched.h"
H
Haojun Liao 已提交
29 30 31
#include "ttime.h"

#define TSC_VAR_NOT_RELEASE 1
32
#define TSC_VAR_RELEASED    0
H
Haojun Liao 已提交
33

dengyihao's avatar
dengyihao 已提交
34
SAppInfo appInfo;
L
Liu Jicong 已提交
35
int32_t  clientReqRefPool = -1;
dengyihao's avatar
dengyihao 已提交
36
int32_t  clientConnRefPool = -1;
H
Haojun Liao 已提交
37

dengyihao's avatar
dengyihao 已提交
38 39
int32_t timestampDeltaLimit = 900;  // s

L
Liu Jicong 已提交
40 41
static TdThreadOnce tscinit = PTHREAD_ONCE_INIT;
volatile int32_t    tscInitRes = 0;
H
Haojun Liao 已提交
42

dengyihao's avatar
dengyihao 已提交
43
static int32_t registerRequest(SRequestObj *pRequest, STscObj *pTscObj) {
H
Haojun Liao 已提交
44
  // connection has been released already, abort creating request.
H
Haojun Liao 已提交
45
  pRequest->self = taosAddRef(clientReqRefPool, pRequest);
H
Haojun Liao 已提交
46 47 48 49

  int32_t num = atomic_add_fetch_32(&pTscObj->numOfReqs, 1);

  if (pTscObj->pAppInfo) {
D
dapan1121 已提交
50
    SAppClusterSummary *pSummary = &pTscObj->pAppInfo->summary;
H
Haojun Liao 已提交
51

L
Liu Jicong 已提交
52 53
    int32_t total = atomic_add_fetch_64((int64_t *)&pSummary->totalRequests, 1);
    int32_t currentInst = atomic_add_fetch_64((int64_t *)&pSummary->currentRequests, 1);
dengyihao's avatar
dengyihao 已提交
54 55
    tscDebug("0x%" PRIx64 " new Request from connObj:0x%" PRIx64
             ", current:%d, app current:%d, total:%d, reqId:0x%" PRIx64,
D
dapan1121 已提交
56
             pRequest->self, pRequest->pTscObj->id, num, currentInst, total, pRequest->requestId);
H
Haojun Liao 已提交
57
  }
D
dapan1121 已提交
58 59

  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
60 61
}

dengyihao's avatar
dengyihao 已提交
62
static void deregisterRequest(SRequestObj *pRequest) {
dengyihao's avatar
dengyihao 已提交
63
  const static int64_t SLOW_QUERY_INTERVAL = 3000000L;  // todo configurable
H
Haojun Liao 已提交
64 65
  assert(pRequest != NULL);

dengyihao's avatar
dengyihao 已提交
66
  STscObj            *pTscObj = pRequest->pTscObj;
D
dapan1121 已提交
67
  SAppClusterSummary *pActivity = &pTscObj->pAppInfo->summary;
H
Haojun Liao 已提交
68

L
Liu Jicong 已提交
69
  int32_t currentInst = atomic_sub_fetch_64((int64_t *)&pActivity->currentRequests, 1);
H
Haojun Liao 已提交
70 71
  int32_t num = atomic_sub_fetch_32(&pTscObj->numOfReqs, 1);

D
dapan1121 已提交
72 73
  int64_t nowUs = taosGetTimestampUs();
  int64_t duration = nowUs - pRequest->metric.start;
dengyihao's avatar
dengyihao 已提交
74 75
  tscDebug("0x%" PRIx64 " free Request from connObj: 0x%" PRIx64 ", reqId:0x%" PRIx64 " elapsed:%" PRIu64
           " ms, current:%d, app current:%d",
D
dapan1121 已提交
76
           pRequest->self, pTscObj->id, pRequest->requestId, duration / 1000, num, currentInst);
D
dapan1121 已提交
77 78

  if (QUERY_NODE_VNODE_MODIF_STMT == pRequest->stmtType) {
X
Xiaoyu Wang 已提交
79 80 81 82
    tscPerf("insert duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
            "us, exec:%" PRId64 "us",
            duration, pRequest->metric.syntaxEnd - pRequest->metric.syntaxStart,
            pRequest->metric.ctgEnd - pRequest->metric.ctgStart, pRequest->metric.semanticEnd - pRequest->metric.ctgEnd,
D
dapan1121 已提交
83
            pRequest->metric.execEnd - pRequest->metric.semanticEnd);
D
dapan1121 已提交
84 85
    atomic_add_fetch_64((int64_t *)&pActivity->insertElapsedTime, duration);
  } else if (QUERY_NODE_SELECT_STMT == pRequest->stmtType) {
X
Xiaoyu Wang 已提交
86 87 88 89
    tscPerf("select duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
            "us, planner:%" PRId64 "us, exec:%" PRId64 "us",
            duration, pRequest->metric.syntaxEnd - pRequest->metric.syntaxStart,
            pRequest->metric.ctgEnd - pRequest->metric.ctgStart, pRequest->metric.semanticEnd - pRequest->metric.ctgEnd,
90
            pRequest->metric.planEnd - pRequest->metric.semanticEnd,
X
Xiaoyu Wang 已提交
91
            pRequest->metric.resultReady - pRequest->metric.planEnd);
dengyihao's avatar
dengyihao 已提交
92
    atomic_add_fetch_64((int64_t *)&pActivity->queryElapsedTime, duration);
D
dapan1121 已提交
93
  }
dengyihao's avatar
dengyihao 已提交
94

D
dapan1121 已提交
95 96 97
  if (duration >= SLOW_QUERY_INTERVAL) {
    atomic_add_fetch_64((int64_t *)&pActivity->numOfSlowQueries, 1);
  }
dengyihao's avatar
dengyihao 已提交
98

D
dapan1121 已提交
99
  releaseTscObj(pTscObj->id);
H
Haojun Liao 已提交
100 101
}

102
// todo close the transporter properly
D
dapan1121 已提交
103 104
void closeTransporter(SAppInstInfo *pAppInfo) {
  if (pAppInfo == NULL || pAppInfo->pTransporter == NULL) {
H
Haojun Liao 已提交
105 106 107
    return;
  }

D
dapan1121 已提交
108 109
  tscDebug("free transporter:%p in app inst %p", pAppInfo->pTransporter, pAppInfo);
  rpcClose(pAppInfo->pTransporter);
H
Haojun Liao 已提交
110 111
}

dengyihao's avatar
dengyihao 已提交
112
static bool clientRpcRfp(int32_t code, tmsg_t msgType) {
D
dapan1121 已提交
113
  if (NEED_REDIRECT_ERROR(code)) {
dengyihao's avatar
dengyihao 已提交
114
    if (msgType == TDMT_SCH_QUERY || msgType == TDMT_SCH_MERGE_QUERY || msgType == TDMT_SCH_FETCH ||
D
dapan1121 已提交
115
        msgType == TDMT_SCH_MERGE_FETCH || msgType == TDMT_SCH_QUERY_HEARTBEAT || msgType == TDMT_SCH_DROP_TASK) {
dengyihao's avatar
dengyihao 已提交
116 117
      return false;
    }
D
dapan1121 已提交
118 119 120 121 122 123
    return true;
  } else {
    return false;
  }
}

dengyihao's avatar
dengyihao 已提交
124 125
// start timer for particular msgType
static bool clientRpcTfp(int32_t code, tmsg_t msgType) {
dengyihao's avatar
dengyihao 已提交
126 127 128
  if (msgType == TDMT_VND_SUBMIT || msgType == TDMT_VND_CREATE_TABLE) {
    return true;
  }
dengyihao's avatar
dengyihao 已提交
129 130 131
  return false;
}

H
Haojun Liao 已提交
132
// TODO refactor
dengyihao's avatar
dengyihao 已提交
133
void *openTransporter(const char *user, const char *auth, int32_t numOfThread) {
H
Haojun Liao 已提交
134 135 136 137 138 139
  SRpcInit rpcInit;
  memset(&rpcInit, 0, sizeof(rpcInit));
  rpcInit.localPort = 0;
  rpcInit.label = "TSC";
  rpcInit.numOfThreads = numOfThread;
  rpcInit.cfp = processMsgFromServer;
D
dapan1121 已提交
140
  rpcInit.rfp = clientRpcRfp;
dengyihao's avatar
dengyihao 已提交
141
  // rpcInit.tfp = clientRpcTfp;
S
Shengliang Guan 已提交
142
  rpcInit.sessions = 1024;
H
Haojun Liao 已提交
143 144
  rpcInit.connType = TAOS_CONN_CLIENT;
  rpcInit.user = (char *)user;
S
Shengliang Guan 已提交
145
  rpcInit.idleTime = tsShellActivityTimer * 1000;
dengyihao's avatar
dengyihao 已提交
146
  void *pDnodeConn = rpcOpen(&rpcInit);
H
Haojun Liao 已提交
147 148 149 150 151 152 153 154
  if (pDnodeConn == NULL) {
    tscError("failed to init connection to server");
    return NULL;
  }

  return pDnodeConn;
}

D
dapan1121 已提交
155
void destroyAllRequests(SHashObj *pRequests) {
L
Liu Jicong 已提交
156
  void *pIter = taosHashIterate(pRequests, NULL);
D
dapan1121 已提交
157 158 159
  while (pIter != NULL) {
    int64_t *rid = pIter;

D
dapan1121 已提交
160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179
    SRequestObj *pRequest = acquireRequest(*rid);
    if (pRequest) {
      destroyRequest(pRequest);
      releaseRequest(*rid);
    }

    pIter = taosHashIterate(pRequests, pIter);
  }
}

void stopAllRequests(SHashObj *pRequests) {
  void *pIter = taosHashIterate(pRequests, NULL);
  while (pIter != NULL) {
    int64_t *rid = pIter;

    SRequestObj *pRequest = acquireRequest(*rid);
    if (pRequest) {
      taos_stop_query(pRequest);
      releaseRequest(*rid);
    }
L
Liu Jicong 已提交
180

D
dapan1121 已提交
181 182 183 184
    pIter = taosHashIterate(pRequests, pIter);
  }
}

dengyihao's avatar
dengyihao 已提交
185
void destroyAppInst(SAppInstInfo *pAppInfo) {
D
dapan1121 已提交
186
  tscDebug("destroy app inst mgr %p", pAppInfo);
D
dapan1121 已提交
187 188

  taosThreadMutexLock(&appInfo.mutex);
dengyihao's avatar
dengyihao 已提交
189

D
dapan1121 已提交
190 191
  hbRemoveAppHbMrg(&pAppInfo->pAppHbMgr);
  taosHashRemove(appInfo.pInstMap, pAppInfo->instKey, strlen(pAppInfo->instKey));
D
dapan1121 已提交
192 193 194

  taosThreadMutexUnlock(&appInfo.mutex);

D
dapan1121 已提交
195 196
  taosMemoryFreeClear(pAppInfo->instKey);
  closeTransporter(pAppInfo);
dengyihao's avatar
dengyihao 已提交
197

D
dapan1121 已提交
198 199
  taosThreadMutexLock(&pAppInfo->qnodeMutex);
  taosArrayDestroy(pAppInfo->pQnodeList);
dengyihao's avatar
dengyihao 已提交
200
  taosThreadMutexUnlock(&pAppInfo->qnodeMutex);
D
dapan1121 已提交
201 202 203 204

  taosMemoryFree(pAppInfo);
}

H
Haojun Liao 已提交
205
void destroyTscObj(void *pObj) {
D
dapan1121 已提交
206 207 208
  if (NULL == pObj) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
209

H
Haojun Liao 已提交
210
  STscObj *pTscObj = pObj;
dengyihao's avatar
dengyihao 已提交
211
  int64_t  tscId = pTscObj->id;
D
dapan1121 已提交
212
  tscTrace("begin to destroy tscObj %" PRIx64 " p:%p", tscId, pTscObj);
H
Haojun Liao 已提交
213

D
dapan1121 已提交
214
  SClientHbKey connKey = {.tscRid = pTscObj->id, .connType = pTscObj->connType};
215
  hbDeregisterConn(pTscObj->pAppInfo->pAppHbMgr, connKey);
216

D
dapan1121 已提交
217
  destroyAllRequests(pTscObj->pRequests);
218
  taosHashCleanup(pTscObj->pRequests);
dengyihao's avatar
dengyihao 已提交
219

D
dapan1121 已提交
220
  schedulerStopQueryHb(pTscObj->pAppInfo->pTransporter);
dengyihao's avatar
dengyihao 已提交
221
  tscDebug("connObj 0x%" PRIx64 " p:%p destroyed, remain inst totalConn:%" PRId64, pTscObj->id, pTscObj,
D
dapan1121 已提交
222 223
           pTscObj->pAppInfo->numOfConns);

224
  int64_t connNum = atomic_sub_fetch_64(&pTscObj->pAppInfo->numOfConns, 1);
D
dapan1121 已提交
225
  if (0 == connNum) {
D
dapan1121 已提交
226
    destroyAppInst(pTscObj->pAppInfo);
D
dapan1121 已提交
227
  }
wafwerar's avatar
wafwerar 已提交
228
  taosThreadMutexDestroy(&pTscObj->mutex);
D
dapan1121 已提交
229
  taosMemoryFree(pTscObj);
D
dapan1121 已提交
230

D
dapan1121 已提交
231
  tscTrace("end to destroy tscObj %" PRIx64 " p:%p", tscId, pTscObj);
H
Haojun Liao 已提交
232 233
}

234
void *createTscObj(const char *user, const char *auth, const char *db, int32_t connType, SAppInstInfo *pAppInfo) {
wafwerar's avatar
wafwerar 已提交
235
  STscObj *pObj = (STscObj *)taosMemoryCalloc(1, sizeof(STscObj));
H
Haojun Liao 已提交
236 237 238 239 240
  if (NULL == pObj) {
    terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
    return NULL;
  }

D
dapan1121 已提交
241 242 243 244 245 246
  pObj->pRequests = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
  if (NULL == pObj->pRequests) {
    taosMemoryFree(pObj);
    terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
    return NULL;
  }
L
Liu Jicong 已提交
247

248
  pObj->connType = connType;
H
Haojun Liao 已提交
249 250 251 252
  pObj->pAppInfo = pAppInfo;
  tstrncpy(pObj->user, user, sizeof(pObj->user));
  memcpy(pObj->pass, auth, TSDB_PASSWORD_LEN);

253 254 255 256
  if (db != NULL) {
    tstrncpy(pObj->db, db, tListLen(pObj->db));
  }

wafwerar's avatar
wafwerar 已提交
257
  taosThreadMutexInit(&pObj->mutex, NULL);
D
dapan1121 已提交
258
  pObj->id = taosAddRef(clientConnRefPool, pObj);
259
  pObj->schemalessType = 1;
H
Haojun Liao 已提交
260

D
dapan1121 已提交
261 262
  atomic_add_fetch_64(&pObj->pAppInfo->numOfConns, 1);

D
dapan1121 已提交
263
  tscDebug("connObj created, 0x%" PRIx64 ",p:%p", pObj->id, pObj);
H
Haojun Liao 已提交
264 265 266
  return pObj;
}

L
Liu Jicong 已提交
267
STscObj *acquireTscObj(int64_t rid) { return (STscObj *)taosAcquireRef(clientConnRefPool, rid); }
D
dapan1121 已提交
268

L
Liu Jicong 已提交
269
int32_t releaseTscObj(int64_t rid) { return taosReleaseRef(clientConnRefPool, rid); }
D
dapan1121 已提交
270

271
void *createRequest(uint64_t connId, int32_t type) {
wafwerar's avatar
wafwerar 已提交
272
  SRequestObj *pRequest = (SRequestObj *)taosMemoryCalloc(1, sizeof(SRequestObj));
H
Haojun Liao 已提交
273 274 275 276 277
  if (NULL == pRequest) {
    terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
    return NULL;
  }

dengyihao's avatar
dengyihao 已提交
278
  STscObj *pTscObj = acquireTscObj(connId);
279
  if (pTscObj == NULL) {
280
    taosMemoryFree(pRequest);
281 282 283 284
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
  }

L
Liu Jicong 已提交
285
  pRequest->resType = RES_TYPE__QUERY;
dengyihao's avatar
dengyihao 已提交
286
  pRequest->requestId = generateRequestId();
D
dapan1121 已提交
287
  pRequest->metric.start = taosGetTimestampUs();
H
Haojun Liao 已提交
288

289
  pRequest->body.resInfo.convertUcs4 = true;  // convert ucs4 by default
dengyihao's avatar
dengyihao 已提交
290
  pRequest->type = type;
291 292 293 294

  pRequest->pDb = getDbOfConnection(pTscObj);
  pRequest->pTscObj = pTscObj;

wafwerar's avatar
wafwerar 已提交
295
  pRequest->msgBuf = taosMemoryCalloc(1, ERROR_MSG_BUF_DEFAULT_SIZE);
D
stmt  
dapan1121 已提交
296
  pRequest->msgBufLen = ERROR_MSG_BUF_DEFAULT_SIZE;
H
Haojun Liao 已提交
297 298
  tsem_init(&pRequest->body.rspSem, 0, 0);

299
  if (registerRequest(pRequest, pTscObj)) {
D
dapan1121 已提交
300 301 302
    doDestroyRequest(pRequest);
    return NULL;
  }
L
Liu Jicong 已提交
303

H
Haojun Liao 已提交
304 305 306
  return pRequest;
}

L
Liu Jicong 已提交
307
void doFreeReqResultInfo(SReqResultInfo *pResInfo) {
wafwerar's avatar
wafwerar 已提交
308 309 310 311 312
  taosMemoryFreeClear(pResInfo->pRspMsg);
  taosMemoryFreeClear(pResInfo->length);
  taosMemoryFreeClear(pResInfo->row);
  taosMemoryFreeClear(pResInfo->pCol);
  taosMemoryFreeClear(pResInfo->fields);
313
  taosMemoryFreeClear(pResInfo->userFields);
wmmhello's avatar
wmmhello 已提交
314
  taosMemoryFreeClear(pResInfo->convertJson);
315 316 317

  if (pResInfo->convertBuf != NULL) {
    for (int32_t i = 0; i < pResInfo->numOfCols; ++i) {
wafwerar's avatar
wafwerar 已提交
318
      taosMemoryFreeClear(pResInfo->convertBuf[i]);
319
    }
wafwerar's avatar
wafwerar 已提交
320
    taosMemoryFreeClear(pResInfo->convertBuf);
321
  }
H
Haojun Liao 已提交
322 323
}

D
dapan1121 已提交
324 325 326 327 328 329
SRequestObj *acquireRequest(int64_t rid) { return (SRequestObj *)taosAcquireRef(clientReqRefPool, rid); }

int32_t releaseRequest(int64_t rid) { return taosReleaseRef(clientReqRefPool, rid); }

int32_t removeRequest(int64_t rid) { return taosRemoveRef(clientReqRefPool, rid); }

D
dapan1121 已提交
330
void doDestroyRequest(void *p) {
D
dapan1121 已提交
331 332 333
  if (NULL == p) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
334

dengyihao's avatar
dengyihao 已提交
335
  SRequestObj *pRequest = (SRequestObj *)p;
dengyihao's avatar
dengyihao 已提交
336

D
dapan1121 已提交
337
  int64_t reqId = pRequest->self;
D
dapan1121 已提交
338
  tscTrace("begin to destroy request %" PRIx64 " p:%p", reqId, pRequest);
dengyihao's avatar
dengyihao 已提交
339

D
dapan1121 已提交
340
  taosHashRemove(pRequest->pTscObj->pRequests, &pRequest->self, sizeof(pRequest->self));
L
Liu Jicong 已提交
341

D
dapan1121 已提交
342
  schedulerFreeJob(&pRequest->body.queryJob, 0);
D
dapan1121 已提交
343

wafwerar's avatar
wafwerar 已提交
344 345
  taosMemoryFreeClear(pRequest->msgBuf);
  taosMemoryFreeClear(pRequest->pDb);
H
Haojun Liao 已提交
346

H
Haojun Liao 已提交
347
  doFreeReqResultInfo(&pRequest->body.resInfo);
H
Haojun Liao 已提交
348

X
Xiaoyu Wang 已提交
349 350
  taosArrayDestroy(pRequest->tableList);
  taosArrayDestroy(pRequest->dbList);
D
dapan1121 已提交
351
  taosArrayDestroy(pRequest->targetTableList);
X
Xiaoyu Wang 已提交
352

D
dapan1121 已提交
353
  destroyQueryExecRes(&pRequest->body.resInfo.execRes);
D
dapan1121 已提交
354

D
dapan1121 已提交
355 356 357
  if (pRequest->self) {
    deregisterRequest(pRequest);
  }
D
dapan1121 已提交
358

H
Haojun Liao 已提交
359 360 361 362
  if (pRequest->syncQuery) {
    taosMemoryFree(pRequest->body.param);
  }

D
dapan1121 已提交
363
  taosMemoryFreeClear(pRequest->sqlstr);
364
  taosMemoryFree(pRequest);
D
dapan1121 已提交
365
  tscTrace("end to destroy request %" PRIx64 " p:%p", reqId, pRequest);
H
Haojun Liao 已提交
366 367
}

dengyihao's avatar
dengyihao 已提交
368
void destroyRequest(SRequestObj *pRequest) {
H
Haojun Liao 已提交
369 370 371 372
  if (pRequest == NULL) {
    return;
  }

D
dapan1121 已提交
373
  taos_stop_query(pRequest);
D
dapan1121 已提交
374
  removeRequest(pRequest->self);
H
Haojun Liao 已提交
375 376 377 378 379 380 381
}

void taos_init_imp(void) {
  // In the APIs of other program language, taos_cleanup is not available yet.
  // So, to make sure taos_cleanup will be invoked to clean up the allocated resource to suppress the valgrind warning.
  atexit(taos_cleanup);
  errno = TSDB_CODE_SUCCESS;
wafwerar's avatar
wafwerar 已提交
382
  taosSeedRand(taosGetTimestampSec());
H
Haojun Liao 已提交
383 384

  deltaToUtcInitOnce();
S
Shengliang Guan 已提交
385

wafwerar's avatar
wafwerar 已提交
386
  if (taosCreateLog("taoslog", 10, configDir, NULL, NULL, NULL, NULL, 1) != 0) {
S
Shengliang Guan 已提交
387 388 389 390
    tscInitRes = -1;
    return;
  }

wafwerar's avatar
wafwerar 已提交
391
  if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 1) != 0) {
S
Shengliang Guan 已提交
392 393 394 395
    tscInitRes = -1;
    return;
  }

396
  initQueryModuleMsgHandle();
H
Haojun Liao 已提交
397

D
dapan1121 已提交
398
  taosConvInit();
dengyihao's avatar
dengyihao 已提交
399

S
Shengliang Guan 已提交
400
  rpcInit();
H
Haojun Liao 已提交
401

D
dapan1121 已提交
402
  SCatalogCfg cfg = {.maxDBCacheNum = 100, .maxTblCacheNum = 100};
403 404
  catalogInit(&cfg);

D
dapan1121 已提交
405
  schedulerInit();
S
Shengliang Guan 已提交
406
  tscDebug("starting to initialize TAOS driver");
H
Haojun Liao 已提交
407

wafwerar's avatar
wafwerar 已提交
408
#ifndef WINDOWS
H
Haojun Liao 已提交
409
  taosSetCoreDump(true);
wafwerar's avatar
wafwerar 已提交
410
#endif
H
Haojun Liao 已提交
411 412

  initTaskQueue();
413
  fmFuncMgtInit();
H
Haojun Liao 已提交
414

415
  clientConnRefPool = taosOpenRef(200, destroyTscObj);
dengyihao's avatar
dengyihao 已提交
416
  clientReqRefPool = taosOpenRef(40960, doDestroyRequest);
H
Haojun Liao 已提交
417

dengyihao's avatar
dengyihao 已提交
418
  // transDestroyBuffer(&conn->readBuf);
H
Haojun Liao 已提交
419
  taosGetAppName(appInfo.appName, NULL);
wafwerar's avatar
wafwerar 已提交
420
  taosThreadMutexInit(&appInfo.mutex, NULL);
H
Haojun Liao 已提交
421

dengyihao's avatar
dengyihao 已提交
422
  appInfo.pid = taosGetPId();
H
Haojun Liao 已提交
423
  appInfo.startTime = taosGetTimestampMs();
dengyihao's avatar
dengyihao 已提交
424
  appInfo.pInstMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
H
Haojun Liao 已提交
425 426 427 428
  tscDebug("client is initialized successfully");
}

int taos_init() {
wafwerar's avatar
wafwerar 已提交
429
  taosThreadOnce(&tscinit, taos_init_imp);
H
Haojun Liao 已提交
430 431 432 433
  return tscInitRes;
}

int taos_options_imp(TSDB_OPTION option, const char *str) {
S
Shengliang Guan 已提交
434 435 436 437 438 439 440 441
  if (option != TSDB_OPTION_CONFIGDIR) {
    taos_init();  // initialize global config
  } else {
    tstrncpy(configDir, str, PATH_MAX);
    tscInfo("set cfg:%s to %s", configDir, str);
    return 0;
  }

dengyihao's avatar
dengyihao 已提交
442
  SConfig     *pCfg = taosGetCfg();
S
Shengliang Guan 已提交
443
  SConfigItem *pItem = NULL;
H
Haojun Liao 已提交
444 445 446

  switch (option) {
    case TSDB_OPTION_CONFIGDIR:
S
Shengliang Guan 已提交
447
      pItem = cfgGetItem(pCfg, "configDir");
H
Haojun Liao 已提交
448 449
      break;
    case TSDB_OPTION_SHELL_ACTIVITY_TIMER:
S
Shengliang Guan 已提交
450
      pItem = cfgGetItem(pCfg, "shellActivityTimer");
H
Haojun Liao 已提交
451
      break;
S
Shengliang Guan 已提交
452 453
    case TSDB_OPTION_LOCALE:
      pItem = cfgGetItem(pCfg, "locale");
H
Haojun Liao 已提交
454
      break;
S
Shengliang Guan 已提交
455 456
    case TSDB_OPTION_CHARSET:
      pItem = cfgGetItem(pCfg, "charset");
H
Haojun Liao 已提交
457 458
      break;
    case TSDB_OPTION_TIMEZONE:
S
Shengliang Guan 已提交
459
      pItem = cfgGetItem(pCfg, "timezone");
H
Haojun Liao 已提交
460 461
      break;
    default:
S
Shengliang Guan 已提交
462 463 464 465 466 467 468 469 470 471 472 473 474
      break;
  }

  if (pItem == NULL) {
    tscError("Invalid option %d", option);
    return -1;
  }

  int code = cfgSetItem(pCfg, pItem->name, str, CFG_STYPE_TAOS_OPTIONS);
  if (code != 0) {
    tscError("failed to set cfg:%s to %s since %s", pItem->name, str, terrstr());
  } else {
    tscInfo("set cfg:%s to %s", pItem->name, str);
H
Haojun Liao 已提交
475
  }
S
Shengliang Guan 已提交
476 477

  return code;
H
Haojun Liao 已提交
478 479
}

480 481 482 483 484
/**
 * The request id is an unsigned integer format of 64bit.
 *+------------+-----+-----------+---------------+
 *| uid|localIp| PId | timestamp | serial number |
 *+------------+-----+-----------+---------------+
485
 *| 12bit      |12bit|24bit      |16bit          |
486 487 488 489
 *+------------+-----+-----------+---------------+
 * @return
 */
uint64_t generateRequestId() {
H
Haojun Liao 已提交
490
  static uint64_t hashId = 0;
dengyihao's avatar
dengyihao 已提交
491
  static int32_t  requestSerialId = 0;
H
Haojun Liao 已提交
492 493 494 495 496 497 498 499 500 501 502

  if (hashId == 0) {
    char    uid[64] = {0};
    int32_t code = taosGetSystemUUID(uid, tListLen(uid));
    if (code != TSDB_CODE_SUCCESS) {
      tscError("Failed to get the system uid to generated request id, reason:%s. use ip address instead",
               tstrerror(TAOS_SYSTEM_ERROR(errno)));

    } else {
      hashId = MurmurHash3_32(uid, strlen(uid));
    }
503 504
  }

D
dapan1121 已提交
505
  uint64_t id = 0;
L
Liu Jicong 已提交
506

D
dapan1121 已提交
507 508 509 510
  while (true) {
    int64_t  ts = taosGetTimestampMs();
    uint64_t pid = taosGetPId();
    int32_t  val = atomic_add_fetch_32(&requestSerialId, 1);
511

D
dapan1121 已提交
512 513 514 515 516
    id = ((hashId & 0x0FFF) << 52) | ((pid & 0x0FFF) << 40) | ((ts & 0xFFFFFF) << 16) | (val & 0xFFFF);
    if (id) {
      break;
    }
  }
517 518 519
  return id;
}

H
Haojun Liao 已提交
520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577
#if 0
#include "cJSON.h"
static setConfRet taos_set_config_imp(const char *config){
  setConfRet ret = {SET_CONF_RET_SUCC, {0}};
  static bool setConfFlag = false;
  if (setConfFlag) {
    ret.retCode = SET_CONF_RET_ERR_ONLY_ONCE;
    strcpy(ret.retMsg, "configuration can only set once");
    return ret;
  }
  taosInitGlobalCfg();
  cJSON *root = cJSON_Parse(config);
  if (root == NULL){
    ret.retCode = SET_CONF_RET_ERR_JSON_PARSE;
    strcpy(ret.retMsg, "parse json error");
    return ret;
  }

  int size = cJSON_GetArraySize(root);
  if(!cJSON_IsObject(root) || size == 0) {
    ret.retCode = SET_CONF_RET_ERR_JSON_INVALID;
    strcpy(ret.retMsg, "json content is invalid, must be not empty object");
    return ret;
  }

  if(size >= 1000) {
    ret.retCode = SET_CONF_RET_ERR_TOO_LONG;
    strcpy(ret.retMsg, "json object size is too long");
    return ret;
  }

  for(int i = 0; i < size; i++){
    cJSON *item = cJSON_GetArrayItem(root, i);
    if(!item) {
      ret.retCode = SET_CONF_RET_ERR_INNER;
      strcpy(ret.retMsg, "inner error");
      return ret;
    }
    if(!taosReadConfigOption(item->string, item->valuestring, NULL, NULL, TAOS_CFG_CSTATUS_OPTION, TSDB_CFG_CTYPE_B_CLIENT)){
      ret.retCode = SET_CONF_RET_ERR_PART;
      if (strlen(ret.retMsg) == 0){
        snprintf(ret.retMsg, RET_MSG_LENGTH, "part error|%s", item->string);
      }else{
        int tmp = RET_MSG_LENGTH - 1 - (int)strlen(ret.retMsg);
        size_t leftSize = tmp >= 0 ? tmp : 0;
        strncat(ret.retMsg, "|",  leftSize);
        tmp = RET_MSG_LENGTH - 1 - (int)strlen(ret.retMsg);
        leftSize = tmp >= 0 ? tmp : 0;
        strncat(ret.retMsg, item->string, leftSize);
      }
    }
  }
  cJSON_Delete(root);
  setConfFlag = true;
  return ret;
}

setConfRet taos_set_config(const char *config){
wafwerar's avatar
wafwerar 已提交
578
  taosThreadMutexLock(&setConfMutex);
H
Haojun Liao 已提交
579
  setConfRet ret = taos_set_config_imp(config);
wafwerar's avatar
wafwerar 已提交
580
  taosThreadMutexUnlock(&setConfMutex);
H
Haojun Liao 已提交
581 582
  return ret;
}
583
#endif