clientEnv.c 21.5 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 "qworker.h"
X
Xiaoyu Wang 已提交
23
#include "scheduler.h"
H
Haojun Liao 已提交
24 25
#include "tcache.h"
#include "tglobal.h"
dengyihao's avatar
dengyihao 已提交
26
#include "thttp.h"
dengyihao's avatar
dengyihao 已提交
27
#include "tmsg.h"
H
Haojun Liao 已提交
28 29
#include "tref.h"
#include "trpc.h"
dengyihao's avatar
dengyihao 已提交
30
#include "tsched.h"
H
Haojun Liao 已提交
31 32 33
#include "ttime.h"

#define TSC_VAR_NOT_RELEASE 1
34
#define TSC_VAR_RELEASED    0
H
Haojun Liao 已提交
35

D
dapan1121 已提交
36
STscDbg  tscDbg = {0};
dengyihao's avatar
dengyihao 已提交
37
SAppInfo appInfo;
D
dapan1121 已提交
38
int64_t  lastClusterId = 0;
L
Liu Jicong 已提交
39
int32_t  clientReqRefPool = -1;
dengyihao's avatar
dengyihao 已提交
40
int32_t  clientConnRefPool = -1;
D
dapan1121 已提交
41
int32_t  clientStop = 0;
H
Haojun Liao 已提交
42

dengyihao's avatar
dengyihao 已提交
43 44
int32_t timestampDeltaLimit = 900;  // s

L
Liu Jicong 已提交
45 46
static TdThreadOnce tscinit = PTHREAD_ONCE_INIT;
volatile int32_t    tscInitRes = 0;
H
Haojun Liao 已提交
47

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

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

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

L
Liu Jicong 已提交
57 58
    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 已提交
59 60
    tscDebug("0x%" PRIx64 " new Request from connObj:0x%" PRIx64
             ", current:%d, app current:%d, total:%d, reqId:0x%" PRIx64,
D
dapan1121 已提交
61
             pRequest->self, pRequest->pTscObj->id, num, currentInst, total, pRequest->requestId);
H
Haojun Liao 已提交
62
  }
D
dapan1121 已提交
63 64

  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
65 66
}

dengyihao's avatar
dengyihao 已提交
67
static void deregisterRequest(SRequestObj *pRequest) {
dengyihao's avatar
dengyihao 已提交
68
  const static int64_t SLOW_QUERY_INTERVAL = 3000000L;  // todo configurable
X
Xiaoyu Wang 已提交
69
  if (pRequest == NULL) {
70 71 72
    tscError("pRequest == NULL");
    return;
  }
H
Haojun Liao 已提交
73

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

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

80
  int64_t duration = taosGetTimestampUs() - pRequest->metric.start;
dengyihao's avatar
dengyihao 已提交
81 82
  tscDebug("0x%" PRIx64 " free Request from connObj: 0x%" PRIx64 ", reqId:0x%" PRIx64
           " elapsed:%.2f ms, "
83 84
           "current:%d, app current:%d",
           pRequest->self, pTscObj->id, pRequest->requestId, duration / 1000.0, num, currentInst);
D
dapan1121 已提交
85

86
  if (pRequest->pQuery && pRequest->pQuery->pRoot) {
X
Xiaoyu Wang 已提交
87 88
    if (QUERY_NODE_VNODE_MODIFY_STMT == pRequest->pQuery->pRoot->type &&
        (0 == ((SVnodeModifyOpStmt *)pRequest->pQuery->pRoot)->sqlNodeType)) {
89
      tscDebug("insert duration %" PRId64 "us: parseCost:%" PRId64 "us, ctgCost:%" PRId64 "us, analyseCost:%" PRId64
dengyihao's avatar
dengyihao 已提交
90 91 92
               "us, planCost:%" PRId64 "us, exec:%" PRId64 "us",
               duration, pRequest->metric.parseCostUs, pRequest->metric.ctgCostUs, pRequest->metric.analyseCostUs,
               pRequest->metric.planCostUs, pRequest->metric.execCostUs);
93 94 95
      atomic_add_fetch_64((int64_t *)&pActivity->insertElapsedTime, duration);
    } else if (QUERY_NODE_SELECT_STMT == pRequest->stmtType) {
      tscDebug("query duration %" PRId64 "us: parseCost:%" PRId64 "us, ctgCost:%" PRId64 "us, analyseCost:%" PRId64
dengyihao's avatar
dengyihao 已提交
96 97 98
               "us, planCost:%" PRId64 "us, exec:%" PRId64 "us",
               duration, pRequest->metric.parseCostUs, pRequest->metric.ctgCostUs, pRequest->metric.analyseCostUs,
               pRequest->metric.planCostUs, pRequest->metric.execCostUs);
99 100 101

      atomic_add_fetch_64((int64_t *)&pActivity->queryElapsedTime, duration);
    }
D
dapan1121 已提交
102
  }
dengyihao's avatar
dengyihao 已提交
103

D
dapan1121 已提交
104 105 106
  if (duration >= SLOW_QUERY_INTERVAL) {
    atomic_add_fetch_64((int64_t *)&pActivity->numOfSlowQueries, 1);
  }
dengyihao's avatar
dengyihao 已提交
107

D
dapan1121 已提交
108
  releaseTscObj(pTscObj->id);
H
Haojun Liao 已提交
109 110
}

111
// todo close the transporter properly
D
dapan1121 已提交
112 113
void closeTransporter(SAppInstInfo *pAppInfo) {
  if (pAppInfo == NULL || pAppInfo->pTransporter == NULL) {
H
Haojun Liao 已提交
114 115 116
    return;
  }

D
dapan1121 已提交
117 118
  tscDebug("free transporter:%p in app inst %p", pAppInfo->pTransporter, pAppInfo);
  rpcClose(pAppInfo->pTransporter);
H
Haojun Liao 已提交
119 120
}

dengyihao's avatar
dengyihao 已提交
121
static bool clientRpcRfp(int32_t code, tmsg_t msgType) {
D
dapan1121 已提交
122
  if (NEED_REDIRECT_ERROR(code)) {
dengyihao's avatar
dengyihao 已提交
123
    if (msgType == TDMT_SCH_QUERY || msgType == TDMT_SCH_MERGE_QUERY || msgType == TDMT_SCH_FETCH ||
D
dapan1121 已提交
124
        msgType == TDMT_SCH_MERGE_FETCH || msgType == TDMT_SCH_QUERY_HEARTBEAT || msgType == TDMT_SCH_DROP_TASK) {
dengyihao's avatar
dengyihao 已提交
125 126
      return false;
    }
D
dapan1121 已提交
127 128 129 130 131 132
    return true;
  } else {
    return false;
  }
}

dengyihao's avatar
dengyihao 已提交
133 134
// start timer for particular msgType
static bool clientRpcTfp(int32_t code, tmsg_t msgType) {
dengyihao's avatar
dengyihao 已提交
135 136 137
  if (msgType == TDMT_VND_SUBMIT || msgType == TDMT_VND_CREATE_TABLE) {
    return true;
  }
dengyihao's avatar
dengyihao 已提交
138 139 140
  return false;
}

H
Haojun Liao 已提交
141
// TODO refactor
dengyihao's avatar
dengyihao 已提交
142
void *openTransporter(const char *user, const char *auth, int32_t numOfThread) {
H
Haojun Liao 已提交
143 144 145 146
  SRpcInit rpcInit;
  memset(&rpcInit, 0, sizeof(rpcInit));
  rpcInit.localPort = 0;
  rpcInit.label = "TSC";
dengyihao's avatar
dengyihao 已提交
147
  rpcInit.numOfThreads = tsNumOfRpcThreads;
H
Haojun Liao 已提交
148
  rpcInit.cfp = processMsgFromServer;
D
dapan1121 已提交
149
  rpcInit.rfp = clientRpcRfp;
S
Shengliang Guan 已提交
150
  rpcInit.sessions = 1024;
H
Haojun Liao 已提交
151 152
  rpcInit.connType = TAOS_CONN_CLIENT;
  rpcInit.user = (char *)user;
S
Shengliang Guan 已提交
153
  rpcInit.idleTime = tsShellActivityTimer * 1000;
dengyihao's avatar
dengyihao 已提交
154
  rpcInit.compressSize = tsCompressMsgSize;
dengyihao's avatar
dengyihao 已提交
155
  rpcInit.dfp = destroyAhandle;
dengyihao's avatar
dengyihao 已提交
156

dengyihao's avatar
dengyihao 已提交
157 158 159 160
  rpcInit.retryMinInterval = tsRedirectPeriod;
  rpcInit.retryStepFactor = tsRedirectFactor;
  rpcInit.retryMaxInterval = tsRedirectMaxPeriod;
  rpcInit.retryMaxTimouet = tsMaxRetryWaitTime;
dengyihao's avatar
dengyihao 已提交
161

dengyihao's avatar
dengyihao 已提交
162
  int32_t connLimitNum = tsNumOfRpcSessions / (tsNumOfRpcThreads * 3);
dengyihao's avatar
dengyihao 已提交
163
  connLimitNum = TMAX(connLimitNum, 10);
dengyihao's avatar
dengyihao 已提交
164
  connLimitNum = TMIN(connLimitNum, 1000);
dengyihao's avatar
dengyihao 已提交
165
  rpcInit.connLimitNum = connLimitNum;
dengyihao's avatar
dengyihao 已提交
166
  rpcInit.timeToGetConn = tsTimeToGetAvailableConn;
dengyihao's avatar
dengyihao 已提交
167

dengyihao's avatar
dengyihao 已提交
168
  void *pDnodeConn = rpcOpen(&rpcInit);
H
Haojun Liao 已提交
169 170 171 172 173 174 175 176
  if (pDnodeConn == NULL) {
    tscError("failed to init connection to server");
    return NULL;
  }

  return pDnodeConn;
}

D
dapan1121 已提交
177
void destroyAllRequests(SHashObj *pRequests) {
L
Liu Jicong 已提交
178
  void *pIter = taosHashIterate(pRequests, NULL);
D
dapan1121 已提交
179 180 181
  while (pIter != NULL) {
    int64_t *rid = pIter;

D
dapan1121 已提交
182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201
    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 已提交
202

D
dapan1121 已提交
203 204 205 206
    pIter = taosHashIterate(pRequests, pIter);
  }
}

dengyihao's avatar
dengyihao 已提交
207
void destroyAppInst(SAppInstInfo *pAppInfo) {
D
dapan1121 已提交
208
  tscDebug("destroy app inst mgr %p", pAppInfo);
D
dapan1121 已提交
209 210

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

D
dapan1121 已提交
212 213
  hbRemoveAppHbMrg(&pAppInfo->pAppHbMgr);
  taosHashRemove(appInfo.pInstMap, pAppInfo->instKey, strlen(pAppInfo->instKey));
D
dapan1121 已提交
214 215 216

  taosThreadMutexUnlock(&appInfo.mutex);

D
dapan1121 已提交
217 218
  taosMemoryFreeClear(pAppInfo->instKey);
  closeTransporter(pAppInfo);
dengyihao's avatar
dengyihao 已提交
219

D
dapan1121 已提交
220 221
  taosThreadMutexLock(&pAppInfo->qnodeMutex);
  taosArrayDestroy(pAppInfo->pQnodeList);
dengyihao's avatar
dengyihao 已提交
222
  taosThreadMutexUnlock(&pAppInfo->qnodeMutex);
D
dapan1121 已提交
223 224 225 226

  taosMemoryFree(pAppInfo);
}

H
Haojun Liao 已提交
227
void destroyTscObj(void *pObj) {
D
dapan1121 已提交
228 229 230
  if (NULL == pObj) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
231

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

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

D
dapan1121 已提交
239
  destroyAllRequests(pTscObj->pRequests);
240
  taosHashCleanup(pTscObj->pRequests);
dengyihao's avatar
dengyihao 已提交
241

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

H
Haojun Liao 已提交
246
  // In any cases, we should not free app inst here. Or an race condition rises.
dengyihao's avatar
dengyihao 已提交
247
  /*int64_t connNum = */ atomic_sub_fetch_64(&pTscObj->pAppInfo->numOfConns, 1);
248

wafwerar's avatar
wafwerar 已提交
249
  taosThreadMutexDestroy(&pTscObj->mutex);
D
dapan1121 已提交
250
  taosMemoryFree(pTscObj);
D
dapan1121 已提交
251

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

255
void *createTscObj(const char *user, const char *auth, const char *db, int32_t connType, SAppInstInfo *pAppInfo) {
wafwerar's avatar
wafwerar 已提交
256
  STscObj *pObj = (STscObj *)taosMemoryCalloc(1, sizeof(STscObj));
H
Haojun Liao 已提交
257
  if (NULL == pObj) {
S
Shengliang Guan 已提交
258
    terrno = TSDB_CODE_OUT_OF_MEMORY;
H
Haojun Liao 已提交
259 260 261
    return NULL;
  }

D
dapan1121 已提交
262 263 264
  pObj->pRequests = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
  if (NULL == pObj->pRequests) {
    taosMemoryFree(pObj);
S
Shengliang Guan 已提交
265
    terrno = TSDB_CODE_OUT_OF_MEMORY;
D
dapan1121 已提交
266 267
    return NULL;
  }
L
Liu Jicong 已提交
268

269
  pObj->connType = connType;
H
Haojun Liao 已提交
270 271 272 273
  pObj->pAppInfo = pAppInfo;
  tstrncpy(pObj->user, user, sizeof(pObj->user));
  memcpy(pObj->pass, auth, TSDB_PASSWORD_LEN);

274 275 276 277
  if (db != NULL) {
    tstrncpy(pObj->db, db, tListLen(pObj->db));
  }

wafwerar's avatar
wafwerar 已提交
278
  taosThreadMutexInit(&pObj->mutex, NULL);
D
dapan1121 已提交
279
  pObj->id = taosAddRef(clientConnRefPool, pObj);
H
Haojun Liao 已提交
280

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

D
dapan1121 已提交
283
  tscDebug("connObj created, 0x%" PRIx64 ",p:%p", pObj->id, pObj);
H
Haojun Liao 已提交
284 285 286
  return pObj;
}

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

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

dengyihao's avatar
dengyihao 已提交
291
void *createRequest(uint64_t connId, int32_t type, int64_t reqid) {
wafwerar's avatar
wafwerar 已提交
292
  SRequestObj *pRequest = (SRequestObj *)taosMemoryCalloc(1, sizeof(SRequestObj));
H
Haojun Liao 已提交
293
  if (NULL == pRequest) {
S
Shengliang Guan 已提交
294
    terrno = TSDB_CODE_OUT_OF_MEMORY;
H
Haojun Liao 已提交
295 296 297
    return NULL;
  }

dengyihao's avatar
dengyihao 已提交
298
  STscObj *pTscObj = acquireTscObj(connId);
299
  if (pTscObj == NULL) {
300
    taosMemoryFree(pRequest);
301 302 303 304
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
  }

L
Liu Jicong 已提交
305
  pRequest->resType = RES_TYPE__QUERY;
dengyihao's avatar
dengyihao 已提交
306
  pRequest->requestId = reqid == 0 ? generateRequestId() : reqid;
D
dapan1121 已提交
307
  pRequest->metric.start = taosGetTimestampUs();
H
Haojun Liao 已提交
308

309
  pRequest->body.resInfo.convertUcs4 = true;  // convert ucs4 by default
dengyihao's avatar
dengyihao 已提交
310
  pRequest->type = type;
311
  pRequest->allocatorRefId = -1;
312 313 314 315

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

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

320
  if (registerRequest(pRequest, pTscObj)) {
D
dapan1121 已提交
321 322 323
    doDestroyRequest(pRequest);
    return NULL;
  }
L
Liu Jicong 已提交
324

H
Haojun Liao 已提交
325 326 327
  return pRequest;
}

L
Liu Jicong 已提交
328
void doFreeReqResultInfo(SReqResultInfo *pResInfo) {
wafwerar's avatar
wafwerar 已提交
329 330 331 332 333
  taosMemoryFreeClear(pResInfo->pRspMsg);
  taosMemoryFreeClear(pResInfo->length);
  taosMemoryFreeClear(pResInfo->row);
  taosMemoryFreeClear(pResInfo->pCol);
  taosMemoryFreeClear(pResInfo->fields);
334
  taosMemoryFreeClear(pResInfo->userFields);
wmmhello's avatar
wmmhello 已提交
335
  taosMemoryFreeClear(pResInfo->convertJson);
336 337 338

  if (pResInfo->convertBuf != NULL) {
    for (int32_t i = 0; i < pResInfo->numOfCols; ++i) {
wafwerar's avatar
wafwerar 已提交
339
      taosMemoryFreeClear(pResInfo->convertBuf[i]);
340
    }
wafwerar's avatar
wafwerar 已提交
341
    taosMemoryFreeClear(pResInfo->convertBuf);
342
  }
H
Haojun Liao 已提交
343 344
}

D
dapan1121 已提交
345 346 347 348 349 350
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 已提交
351
void doDestroyRequest(void *p) {
D
dapan1121 已提交
352 353 354
  if (NULL == p) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
355

dengyihao's avatar
dengyihao 已提交
356
  SRequestObj *pRequest = (SRequestObj *)p;
dengyihao's avatar
dengyihao 已提交
357

D
dapan1121 已提交
358
  uint64_t reqId = pRequest->requestId;
D
dapan1121 已提交
359
  tscTrace("begin to destroy request %" PRIx64 " p:%p", reqId, pRequest);
dengyihao's avatar
dengyihao 已提交
360

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

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

wafwerar's avatar
wafwerar 已提交
365 366
  taosMemoryFreeClear(pRequest->msgBuf);
  taosMemoryFreeClear(pRequest->pDb);
H
Haojun Liao 已提交
367

H
Haojun Liao 已提交
368
  doFreeReqResultInfo(&pRequest->body.resInfo);
369
  tsem_destroy(&pRequest->body.rspSem);
H
Haojun Liao 已提交
370

X
Xiaoyu Wang 已提交
371 372
  taosArrayDestroy(pRequest->tableList);
  taosArrayDestroy(pRequest->dbList);
D
dapan1121 已提交
373
  taosArrayDestroy(pRequest->targetTableList);
X
Xiaoyu Wang 已提交
374

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

D
dapan1121 已提交
377 378 379
  if (pRequest->self) {
    deregisterRequest(pRequest);
  }
D
dapan1121 已提交
380

H
Haojun Liao 已提交
381
  if (pRequest->syncQuery) {
dengyihao's avatar
dengyihao 已提交
382 383 384
    if (pRequest->body.param) {
      tsem_destroy(&((SSyncQueryParam *)pRequest->body.param)->sem);
    }
H
Haojun Liao 已提交
385 386 387
    taosMemoryFree(pRequest->body.param);
  }

388 389 390
  qDestroyQuery(pRequest->pQuery);
  nodesDestroyAllocator(pRequest->allocatorRefId);

D
dapan1121 已提交
391
  taosMemoryFreeClear(pRequest->sqlstr);
392
  taosMemoryFree(pRequest);
D
dapan1121 已提交
393
  tscTrace("end to destroy request %" PRIx64 " p:%p", reqId, pRequest);
H
Haojun Liao 已提交
394 395
}

dengyihao's avatar
dengyihao 已提交
396
void destroyRequest(SRequestObj *pRequest) {
H
Haojun Liao 已提交
397 398 399 400
  if (pRequest == NULL) {
    return;
  }

D
dapan1121 已提交
401
  taos_stop_query(pRequest);
D
dapan1121 已提交
402
  removeRequest(pRequest->self);
H
Haojun Liao 已提交
403 404
}

D
dapan 已提交
405 406
void crashReportThreadFuncUnexpectedStopped(void) { atomic_store_32(&clientStop, -1); }

D
dapan1121 已提交
407 408 409 410
static void *tscCrashReportThreadFp(void *param) {
  setThreadName("client-crashReport");
  char filepath[PATH_MAX] = {0};
  snprintf(filepath, sizeof(filepath), "%s%s.taosCrashLog", tsLogDir, TD_DIRSEP);
dengyihao's avatar
dengyihao 已提交
411 412
  char     *pMsg = NULL;
  int64_t   msgLen = 0;
D
dapan1121 已提交
413
  TdFilePtr pFile = NULL;
dengyihao's avatar
dengyihao 已提交
414 415 416 417
  bool      truncateFile = false;
  int32_t   sleepTime = 200;
  int32_t   reportPeriodNum = 3600 * 1000 / sleepTime;
  int32_t   loopTimes = reportPeriodNum;
D
dapan 已提交
418 419 420 421 422 423

#ifdef WINDOWS
  if (taosCheckCurrentInDll()) {
    atexit(crashReportThreadFuncUnexpectedStopped);
  }
#endif
dengyihao's avatar
dengyihao 已提交
424

D
dapan1121 已提交
425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453
  while (1) {
    if (clientStop) break;
    if (loopTimes++ < reportPeriodNum) {
      taosMsleep(sleepTime);
      continue;
    }

    taosReadCrashInfo(filepath, &pMsg, &msgLen, &pFile);
    if (pMsg && msgLen > 0) {
      if (taosSendHttpReport(tsTelemServer, tsClientCrashReportUri, tsTelemPort, pMsg, msgLen, HTTP_FLAT) != 0) {
        tscError("failed to send crash report");
        if (pFile) {
          taosReleaseCrashLogFile(pFile, false);
          continue;
        }
      } else {
        tscInfo("succeed to send crash report");
        truncateFile = true;
      }
    } else {
      tscDebug("no crash info");
    }

    taosMemoryFree(pMsg);

    if (pMsg && msgLen > 0) {
      pMsg = NULL;
      continue;
    }
dengyihao's avatar
dengyihao 已提交
454

D
dapan1121 已提交
455 456 457 458
    if (pFile) {
      taosReleaseCrashLogFile(pFile, truncateFile);
      truncateFile = false;
    }
dengyihao's avatar
dengyihao 已提交
459

D
dapan1121 已提交
460 461 462 463 464 465 466 467 468 469 470 471
    taosMsleep(sleepTime);
    loopTimes = 0;
  }

  clientStop = -1;
  return NULL;
}

int32_t tscCrashReportInit() {
  if (!tsEnableCrashReport) {
    return 0;
  }
dengyihao's avatar
dengyihao 已提交
472

D
dapan1121 已提交
473 474 475
  TdThreadAttr thAttr;
  taosThreadAttrInit(&thAttr);
  taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
dengyihao's avatar
dengyihao 已提交
476
  TdThread crashReportThread;
D
dapan1121 已提交
477 478 479 480 481 482 483 484 485 486 487 488 489 490
  if (taosThreadCreate(&crashReportThread, &thAttr, tscCrashReportThreadFp, NULL) != 0) {
    tscError("failed to create crashReport thread since %s", strerror(errno));
    return -1;
  }

  taosThreadAttrDestroy(&thAttr);
  return 0;
}

void tscStopCrashReport() {
  if (!tsEnableCrashReport) {
    return;
  }

D
dapan 已提交
491 492 493 494 495
  if (atomic_val_compare_exchange_32(&clientStop, 0, 1)) {
    tscDebug("hb thread already stopped");
    return;
  }

D
dapan1121 已提交
496 497 498 499 500
  while (atomic_load_32(&clientStop) > 0) {
    taosMsleep(100);
  }
}

D
dapan1121 已提交
501
void tscWriteCrashInfo(int signum, void *sigInfo, void *context) {
dengyihao's avatar
dengyihao 已提交
502
  char       *pMsg = NULL;
D
dapan1121 已提交
503 504 505
  const char *flags = "UTL FATAL ";
  ELogLevel   level = DEBUG_FATAL;
  int32_t     dflag = 255;
dengyihao's avatar
dengyihao 已提交
506
  int64_t     msgLen = -1;
D
dapan1121 已提交
507 508 509 510 511

  if (tsEnableCrashReport) {
    if (taosGenCrashJsonMsg(signum, &pMsg, lastClusterId, appInfo.startTime)) {
      taosPrintLog(flags, level, dflag, "failed to generate crash json msg");
    } else {
dengyihao's avatar
dengyihao 已提交
512
      msgLen = strlen(pMsg);
D
dapan1121 已提交
513 514 515 516
    }
  }

  taosLogCrashInfo("taos", pMsg, msgLen, signum, sigInfo);
D
dapan1121 已提交
517 518
}

H
Haojun Liao 已提交
519
void taos_init_imp(void) {
D
dapan1121 已提交
520 521 522 523 524 525
#if defined(LINUX)
  if (tscDbg.memEnable) {
    int32_t code = taosMemoryDbgInit();
    if (code) {
      printf("failed to init memory dbg, error:%s\n", tstrerror(code));
    } else {
dengyihao's avatar
dengyihao 已提交
526
      tsAsyncLog = false;
D
dapan1121 已提交
527 528 529 530 531
      printf("memory dbg enabled\n");
    }
  }
#endif

H
Haojun Liao 已提交
532 533 534 535
  // 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 已提交
536
  taosSeedRand(taosGetTimestampSec());
H
Haojun Liao 已提交
537

D
dapan1121 已提交
538 539 540 541
  appInfo.pid = taosGetPId();
  appInfo.startTime = taosGetTimestampMs();
  appInfo.pInstMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);

H
Haojun Liao 已提交
542
  deltaToUtcInitOnce();
S
Shengliang Guan 已提交
543

wafwerar's avatar
wafwerar 已提交
544
  if (taosCreateLog("taoslog", 10, configDir, NULL, NULL, NULL, NULL, 1) != 0) {
A
Alex Duan 已提交
545
    // ignore create log failed, only print
D
dapan1121 已提交
546
    printf(" WARING: Create taoslog failed:%s. configDir=%s\n", strerror(errno), configDir);
S
Shengliang Guan 已提交
547 548
  }

wafwerar's avatar
wafwerar 已提交
549
  if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 1) != 0) {
S
Shengliang Guan 已提交
550 551 552 553
    tscInitRes = -1;
    return;
  }

554
  initQueryModuleMsgHandle();
H
Haojun Liao 已提交
555

S
Shengliang Guan 已提交
556
  if (taosConvInit() != 0) {
557 558
    tscError("failed to init conv");
    return;
S
Shengliang Guan 已提交
559
  }
dengyihao's avatar
dengyihao 已提交
560

S
Shengliang Guan 已提交
561
  rpcInit();
H
Haojun Liao 已提交
562

D
dapan1121 已提交
563
  SCatalogCfg cfg = {.maxDBCacheNum = 100, .maxTblCacheNum = 100};
564 565
  catalogInit(&cfg);

D
dapan1121 已提交
566
  schedulerInit();
S
Shengliang Guan 已提交
567
  tscDebug("starting to initialize TAOS driver");
H
Haojun Liao 已提交
568

wafwerar's avatar
wafwerar 已提交
569
#ifndef WINDOWS
H
Haojun Liao 已提交
570
  taosSetCoreDump(true);
wafwerar's avatar
wafwerar 已提交
571
#endif
H
Haojun Liao 已提交
572 573

  initTaskQueue();
574
  fmFuncMgtInit();
575
  nodesInitAllocatorSet();
H
Haojun Liao 已提交
576

577
  clientConnRefPool = taosOpenRef(200, destroyTscObj);
dengyihao's avatar
dengyihao 已提交
578
  clientReqRefPool = taosOpenRef(40960, doDestroyRequest);
H
Haojun Liao 已提交
579

dengyihao's avatar
dengyihao 已提交
580
  // transDestroyBuffer(&conn->readBuf);
H
Haojun Liao 已提交
581
  taosGetAppName(appInfo.appName, NULL);
wafwerar's avatar
wafwerar 已提交
582
  taosThreadMutexInit(&appInfo.mutex, NULL);
H
Haojun Liao 已提交
583

D
dapan1121 已提交
584
  tscCrashReportInit();
dengyihao's avatar
dengyihao 已提交
585

H
Haojun Liao 已提交
586 587 588 589
  tscDebug("client is initialized successfully");
}

int taos_init() {
wafwerar's avatar
wafwerar 已提交
590
  taosThreadOnce(&tscinit, taos_init_imp);
H
Haojun Liao 已提交
591 592 593 594
  return tscInitRes;
}

int taos_options_imp(TSDB_OPTION option, const char *str) {
H
Haojun Liao 已提交
595
  if (option == TSDB_OPTION_CONFIGDIR) {
S
Shengliang Guan 已提交
596 597 598
    tstrncpy(configDir, str, PATH_MAX);
    tscInfo("set cfg:%s to %s", configDir, str);
    return 0;
H
Haojun Liao 已提交
599 600
  } else {
    taos_init();  // initialize global config
S
Shengliang Guan 已提交
601 602
  }

dengyihao's avatar
dengyihao 已提交
603
  SConfig     *pCfg = taosGetCfg();
S
Shengliang Guan 已提交
604
  SConfigItem *pItem = NULL;
H
Haojun Liao 已提交
605 606 607

  switch (option) {
    case TSDB_OPTION_SHELL_ACTIVITY_TIMER:
S
Shengliang Guan 已提交
608
      pItem = cfgGetItem(pCfg, "shellActivityTimer");
H
Haojun Liao 已提交
609
      break;
S
Shengliang Guan 已提交
610 611
    case TSDB_OPTION_LOCALE:
      pItem = cfgGetItem(pCfg, "locale");
H
Haojun Liao 已提交
612
      break;
S
Shengliang Guan 已提交
613 614
    case TSDB_OPTION_CHARSET:
      pItem = cfgGetItem(pCfg, "charset");
H
Haojun Liao 已提交
615 616
      break;
    case TSDB_OPTION_TIMEZONE:
S
Shengliang Guan 已提交
617
      pItem = cfgGetItem(pCfg, "timezone");
H
Haojun Liao 已提交
618
      break;
619 620 621
    case TSDB_OPTION_USE_ADAPTER:
      pItem = cfgGetItem(pCfg, "useAdapter");
      break;
H
Haojun Liao 已提交
622
    default:
S
Shengliang Guan 已提交
623 624 625 626 627 628 629 630 631 632 633 634 635
      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);
X
Xiaoyu Wang 已提交
636 637 638
    if (TSDB_OPTION_SHELL_ACTIVITY_TIMER == option || TSDB_OPTION_USE_ADAPTER == option) {
      code = taosSetCfg(pCfg, pItem->name);
    }
H
Haojun Liao 已提交
639
  }
S
Shengliang Guan 已提交
640 641

  return code;
H
Haojun Liao 已提交
642 643
}

644 645 646 647 648
/**
 * The request id is an unsigned integer format of 64bit.
 *+------------+-----+-----------+---------------+
 *| uid|localIp| PId | timestamp | serial number |
 *+------------+-----+-----------+---------------+
649
 *| 12bit      |12bit|24bit      |16bit          |
650 651 652 653
 *+------------+-----+-----------+---------------+
 * @return
 */
uint64_t generateRequestId() {
H
Haojun Liao 已提交
654
  static uint64_t hashId = 0;
dengyihao's avatar
dengyihao 已提交
655
  static uint32_t requestSerialId = 0;
H
Haojun Liao 已提交
656 657 658 659 660 661 662 663 664 665 666

  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));
    }
667 668
  }

D
dapan1121 已提交
669
  uint64_t id = 0;
L
Liu Jicong 已提交
670

D
dapan1121 已提交
671 672 673
  while (true) {
    int64_t  ts = taosGetTimestampMs();
    uint64_t pid = taosGetPId();
dengyihao's avatar
dengyihao 已提交
674 675
    uint32_t val = atomic_add_fetch_32(&requestSerialId, 1);
    if (val >= 0xFFFF) atomic_store_32(&requestSerialId, 0);
676

D
dapan1121 已提交
677 678 679 680 681
    id = ((hashId & 0x0FFF) << 52) | ((pid & 0x0FFF) << 40) | ((ts & 0xFFFFFF) << 16) | (val & 0xFFFF);
    if (id) {
      break;
    }
  }
682 683 684
  return id;
}

H
Haojun Liao 已提交
685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742
#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 已提交
743
  taosThreadMutexLock(&setConfMutex);
H
Haojun Liao 已提交
744
  setConfRet ret = taos_set_config_imp(config);
wafwerar's avatar
wafwerar 已提交
745
  taosThreadMutexUnlock(&setConfMutex);
H
Haojun Liao 已提交
746 747
  return ret;
}
748
#endif