clientEnv.c 21.8 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
#include "ttime.h"

sangshuduo's avatar
sangshuduo 已提交
33 34 35 36
#if defined(CUS_NAME) || defined(CUS_PROMPT) || defined(CUS_EMAIL)
#include "cus_name.h"
#endif

H
Haojun Liao 已提交
37
#define TSC_VAR_NOT_RELEASE 1
38
#define TSC_VAR_RELEASED    0
H
Haojun Liao 已提交
39

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

dengyihao's avatar
dengyihao 已提交
47 48
int32_t timestampDeltaLimit = 900;  // s

L
Liu Jicong 已提交
49 50
static TdThreadOnce tscinit = PTHREAD_ONCE_INIT;
volatile int32_t    tscInitRes = 0;
H
Haojun Liao 已提交
51

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

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

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

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

  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
69 70
}

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

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

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

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

90
  if (pRequest->pQuery && pRequest->pQuery->pRoot) {
X
Xiaoyu Wang 已提交
91 92
    if (QUERY_NODE_VNODE_MODIFY_STMT == pRequest->pQuery->pRoot->type &&
        (0 == ((SVnodeModifyOpStmt *)pRequest->pQuery->pRoot)->sqlNodeType)) {
93
      tscDebug("insert duration %" PRId64 "us: parseCost:%" PRId64 "us, ctgCost:%" PRId64 "us, analyseCost:%" PRId64
dengyihao's avatar
dengyihao 已提交
94 95 96
               "us, planCost:%" PRId64 "us, exec:%" PRId64 "us",
               duration, pRequest->metric.parseCostUs, pRequest->metric.ctgCostUs, pRequest->metric.analyseCostUs,
               pRequest->metric.planCostUs, pRequest->metric.execCostUs);
97 98 99
      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 已提交
100 101 102
               "us, planCost:%" PRId64 "us, exec:%" PRId64 "us",
               duration, pRequest->metric.parseCostUs, pRequest->metric.ctgCostUs, pRequest->metric.analyseCostUs,
               pRequest->metric.planCostUs, pRequest->metric.execCostUs);
103 104 105

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

D
dapan1121 已提交
108 109 110
  if (duration >= SLOW_QUERY_INTERVAL) {
    atomic_add_fetch_64((int64_t *)&pActivity->numOfSlowQueries, 1);
  }
dengyihao's avatar
dengyihao 已提交
111

D
dapan1121 已提交
112
  releaseTscObj(pTscObj->id);
H
Haojun Liao 已提交
113 114
}

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

D
dapan1121 已提交
121 122
  tscDebug("free transporter:%p in app inst %p", pAppInfo->pTransporter, pAppInfo);
  rpcClose(pAppInfo->pTransporter);
H
Haojun Liao 已提交
123 124
}

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

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

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

dengyihao's avatar
dengyihao 已提交
161 162 163 164
  rpcInit.retryMinInterval = tsRedirectPeriod;
  rpcInit.retryStepFactor = tsRedirectFactor;
  rpcInit.retryMaxInterval = tsRedirectMaxPeriod;
  rpcInit.retryMaxTimouet = tsMaxRetryWaitTime;
dengyihao's avatar
dengyihao 已提交
165

dengyihao's avatar
dengyihao 已提交
166
  int32_t connLimitNum = tsNumOfRpcSessions / (tsNumOfRpcThreads * 3);
dengyihao's avatar
dengyihao 已提交
167
  connLimitNum = TMAX(connLimitNum, 10);
dengyihao's avatar
dengyihao 已提交
168
  connLimitNum = TMIN(connLimitNum, 1000);
dengyihao's avatar
dengyihao 已提交
169
  rpcInit.connLimitNum = connLimitNum;
dengyihao's avatar
dengyihao 已提交
170
  rpcInit.timeToGetConn = tsTimeToGetAvailableConn;
dengyihao's avatar
dengyihao 已提交
171

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

  return pDnodeConn;
}

D
dapan1121 已提交
181
void destroyAllRequests(SHashObj *pRequests) {
L
Liu Jicong 已提交
182
  void *pIter = taosHashIterate(pRequests, NULL);
D
dapan1121 已提交
183 184 185
  while (pIter != NULL) {
    int64_t *rid = pIter;

D
dapan1121 已提交
186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205
    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 已提交
206

D
dapan1121 已提交
207 208 209 210
    pIter = taosHashIterate(pRequests, pIter);
  }
}

dengyihao's avatar
dengyihao 已提交
211
void destroyAppInst(SAppInstInfo *pAppInfo) {
D
dapan1121 已提交
212
  tscDebug("destroy app inst mgr %p", pAppInfo);
D
dapan1121 已提交
213 214

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

D
dapan1121 已提交
216 217
  hbRemoveAppHbMrg(&pAppInfo->pAppHbMgr);
  taosHashRemove(appInfo.pInstMap, pAppInfo->instKey, strlen(pAppInfo->instKey));
D
dapan1121 已提交
218 219 220

  taosThreadMutexUnlock(&appInfo.mutex);

D
dapan1121 已提交
221 222
  taosMemoryFreeClear(pAppInfo->instKey);
  closeTransporter(pAppInfo);
dengyihao's avatar
dengyihao 已提交
223

D
dapan1121 已提交
224 225
  taosThreadMutexLock(&pAppInfo->qnodeMutex);
  taosArrayDestroy(pAppInfo->pQnodeList);
dengyihao's avatar
dengyihao 已提交
226
  taosThreadMutexUnlock(&pAppInfo->qnodeMutex);
D
dapan1121 已提交
227 228 229 230

  taosMemoryFree(pAppInfo);
}

H
Haojun Liao 已提交
231
void destroyTscObj(void *pObj) {
D
dapan1121 已提交
232 233 234
  if (NULL == pObj) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
235

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

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

D
dapan1121 已提交
243
  destroyAllRequests(pTscObj->pRequests);
244
  taosHashCleanup(pTscObj->pRequests);
dengyihao's avatar
dengyihao 已提交
245

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

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

wafwerar's avatar
wafwerar 已提交
253
  taosThreadMutexDestroy(&pTscObj->mutex);
D
dapan1121 已提交
254
  taosMemoryFree(pTscObj);
D
dapan1121 已提交
255

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

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

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

273
  pObj->connType = connType;
H
Haojun Liao 已提交
274 275 276 277
  pObj->pAppInfo = pAppInfo;
  tstrncpy(pObj->user, user, sizeof(pObj->user));
  memcpy(pObj->pass, auth, TSDB_PASSWORD_LEN);

278 279 280 281
  if (db != NULL) {
    tstrncpy(pObj->db, db, tListLen(pObj->db));
  }

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

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

D
dapan1121 已提交
287
  tscDebug("connObj created, 0x%" PRIx64 ",p:%p", pObj->id, pObj);
H
Haojun Liao 已提交
288 289 290
  return pObj;
}

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

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

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

dengyihao's avatar
dengyihao 已提交
302
  STscObj *pTscObj = acquireTscObj(connId);
303
  if (pTscObj == NULL) {
304
    taosMemoryFree(pRequest);
305 306 307 308
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
  }

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

313
  pRequest->body.resInfo.convertUcs4 = true;  // convert ucs4 by default
dengyihao's avatar
dengyihao 已提交
314
  pRequest->type = type;
315
  pRequest->allocatorRefId = -1;
316 317 318 319

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

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

324
  if (registerRequest(pRequest, pTscObj)) {
D
dapan1121 已提交
325 326 327
    doDestroyRequest(pRequest);
    return NULL;
  }
L
Liu Jicong 已提交
328

H
Haojun Liao 已提交
329 330 331
  return pRequest;
}

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

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

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

dengyihao's avatar
dengyihao 已提交
360
  SRequestObj *pRequest = (SRequestObj *)p;
dengyihao's avatar
dengyihao 已提交
361

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

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

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

wafwerar's avatar
wafwerar 已提交
369 370
  taosMemoryFreeClear(pRequest->msgBuf);
  taosMemoryFreeClear(pRequest->pDb);
H
Haojun Liao 已提交
371

H
Haojun Liao 已提交
372
  doFreeReqResultInfo(&pRequest->body.resInfo);
373
  tsem_destroy(&pRequest->body.rspSem);
H
Haojun Liao 已提交
374

X
Xiaoyu Wang 已提交
375 376
  taosArrayDestroy(pRequest->tableList);
  taosArrayDestroy(pRequest->dbList);
D
dapan1121 已提交
377
  taosArrayDestroy(pRequest->targetTableList);
X
Xiaoyu Wang 已提交
378

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

D
dapan1121 已提交
381 382 383
  if (pRequest->self) {
    deregisterRequest(pRequest);
  }
D
dapan1121 已提交
384

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

392 393 394
  qDestroyQuery(pRequest->pQuery);
  nodesDestroyAllocator(pRequest->allocatorRefId);

D
dapan1121 已提交
395
  taosMemoryFreeClear(pRequest->sqlstr);
396
  taosMemoryFree(pRequest);
D
dapan1121 已提交
397
  tscTrace("end to destroy request %" PRIx64 " p:%p", reqId, pRequest);
H
Haojun Liao 已提交
398 399
}

dengyihao's avatar
dengyihao 已提交
400
void destroyRequest(SRequestObj *pRequest) {
H
Haojun Liao 已提交
401 402 403 404
  if (pRequest == NULL) {
    return;
  }

D
dapan1121 已提交
405
  taos_stop_query(pRequest);
D
dapan1121 已提交
406
  removeRequest(pRequest->self);
H
Haojun Liao 已提交
407 408
}

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

D
dapan1121 已提交
411 412 413 414
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 已提交
415 416
  char     *pMsg = NULL;
  int64_t   msgLen = 0;
D
dapan1121 已提交
417
  TdFilePtr pFile = NULL;
dengyihao's avatar
dengyihao 已提交
418 419 420 421
  bool      truncateFile = false;
  int32_t   sleepTime = 200;
  int32_t   reportPeriodNum = 3600 * 1000 / sleepTime;
  int32_t   loopTimes = reportPeriodNum;
D
dapan 已提交
422 423 424 425 426 427

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

D
dapan1121 已提交
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 454 455 456 457
  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 已提交
458

D
dapan1121 已提交
459 460 461 462
    if (pFile) {
      taosReleaseCrashLogFile(pFile, truncateFile);
      truncateFile = false;
    }
dengyihao's avatar
dengyihao 已提交
463

D
dapan1121 已提交
464 465 466 467 468 469 470 471 472 473 474 475
    taosMsleep(sleepTime);
    loopTimes = 0;
  }

  clientStop = -1;
  return NULL;
}

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

D
dapan1121 已提交
477 478 479
  TdThreadAttr thAttr;
  taosThreadAttrInit(&thAttr);
  taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
dengyihao's avatar
dengyihao 已提交
480
  TdThread crashReportThread;
D
dapan1121 已提交
481 482 483 484 485 486 487 488 489 490 491 492 493 494
  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 已提交
495 496 497 498 499
  if (atomic_val_compare_exchange_32(&clientStop, 0, 1)) {
    tscDebug("hb thread already stopped");
    return;
  }

D
dapan1121 已提交
500 501 502 503 504
  while (atomic_load_32(&clientStop) > 0) {
    taosMsleep(100);
  }
}

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

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

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

H
Haojun Liao 已提交
523
void taos_init_imp(void) {
D
dapan1121 已提交
524 525 526 527 528 529
#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 已提交
530
      tsAsyncLog = false;
D
dapan1121 已提交
531 532 533 534 535
      printf("memory dbg enabled\n");
    }
  }
#endif

H
Haojun Liao 已提交
536 537 538 539
  // 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 已提交
540
  taosSeedRand(taosGetTimestampSec());
H
Haojun Liao 已提交
541

D
dapan1121 已提交
542 543 544 545
  appInfo.pid = taosGetPId();
  appInfo.startTime = taosGetTimestampMs();
  appInfo.pInstMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);

H
Haojun Liao 已提交
546
  deltaToUtcInitOnce();
S
Shengliang Guan 已提交
547

sangshuduo's avatar
sangshuduo 已提交
548 549 550 551 552 553 554
  char logDirName[64] = {0};
#ifdef CUS_PROMPT
  snprintf(logDirName, 64, "%slog", CUS_PROMPT);
#else
  snprintf(logDirName, 64, "taoslog");
#endif
  if (taosCreateLog(logDirName, 10, configDir, NULL, NULL, NULL, NULL, 1) != 0) {
A
Alex Duan 已提交
555
    // ignore create log failed, only print
sangshuduo's avatar
sangshuduo 已提交
556
    printf(" WARING: Create %s failed:%s. configDir=%s\n", logDirName, strerror(errno), configDir);
S
Shengliang Guan 已提交
557 558
  }

wafwerar's avatar
wafwerar 已提交
559
  if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 1) != 0) {
S
Shengliang Guan 已提交
560 561 562 563
    tscInitRes = -1;
    return;
  }

564
  initQueryModuleMsgHandle();
H
Haojun Liao 已提交
565

S
Shengliang Guan 已提交
566
  if (taosConvInit() != 0) {
567 568
    tscError("failed to init conv");
    return;
S
Shengliang Guan 已提交
569
  }
dengyihao's avatar
dengyihao 已提交
570

S
Shengliang Guan 已提交
571
  rpcInit();
H
Haojun Liao 已提交
572

D
dapan1121 已提交
573
  SCatalogCfg cfg = {.maxDBCacheNum = 100, .maxTblCacheNum = 100};
574 575
  catalogInit(&cfg);

D
dapan1121 已提交
576
  schedulerInit();
S
Shengliang Guan 已提交
577
  tscDebug("starting to initialize TAOS driver");
H
Haojun Liao 已提交
578

wafwerar's avatar
wafwerar 已提交
579
#ifndef WINDOWS
H
Haojun Liao 已提交
580
  taosSetCoreDump(true);
wafwerar's avatar
wafwerar 已提交
581
#endif
H
Haojun Liao 已提交
582 583

  initTaskQueue();
584
  fmFuncMgtInit();
585
  nodesInitAllocatorSet();
H
Haojun Liao 已提交
586

587
  clientConnRefPool = taosOpenRef(200, destroyTscObj);
dengyihao's avatar
dengyihao 已提交
588
  clientReqRefPool = taosOpenRef(40960, doDestroyRequest);
H
Haojun Liao 已提交
589

dengyihao's avatar
dengyihao 已提交
590
  // transDestroyBuffer(&conn->readBuf);
H
Haojun Liao 已提交
591
  taosGetAppName(appInfo.appName, NULL);
wafwerar's avatar
wafwerar 已提交
592
  taosThreadMutexInit(&appInfo.mutex, NULL);
H
Haojun Liao 已提交
593

D
dapan1121 已提交
594
  tscCrashReportInit();
dengyihao's avatar
dengyihao 已提交
595

H
Haojun Liao 已提交
596 597 598 599
  tscDebug("client is initialized successfully");
}

int taos_init() {
wafwerar's avatar
wafwerar 已提交
600
  taosThreadOnce(&tscinit, taos_init_imp);
H
Haojun Liao 已提交
601 602 603 604
  return tscInitRes;
}

int taos_options_imp(TSDB_OPTION option, const char *str) {
H
Haojun Liao 已提交
605
  if (option == TSDB_OPTION_CONFIGDIR) {
S
Shengliang Guan 已提交
606 607 608
    tstrncpy(configDir, str, PATH_MAX);
    tscInfo("set cfg:%s to %s", configDir, str);
    return 0;
H
Haojun Liao 已提交
609 610
  } else {
    taos_init();  // initialize global config
S
Shengliang Guan 已提交
611 612
  }

dengyihao's avatar
dengyihao 已提交
613
  SConfig     *pCfg = taosGetCfg();
S
Shengliang Guan 已提交
614
  SConfigItem *pItem = NULL;
H
Haojun Liao 已提交
615 616 617

  switch (option) {
    case TSDB_OPTION_SHELL_ACTIVITY_TIMER:
S
Shengliang Guan 已提交
618
      pItem = cfgGetItem(pCfg, "shellActivityTimer");
H
Haojun Liao 已提交
619
      break;
S
Shengliang Guan 已提交
620 621
    case TSDB_OPTION_LOCALE:
      pItem = cfgGetItem(pCfg, "locale");
H
Haojun Liao 已提交
622
      break;
S
Shengliang Guan 已提交
623 624
    case TSDB_OPTION_CHARSET:
      pItem = cfgGetItem(pCfg, "charset");
H
Haojun Liao 已提交
625 626
      break;
    case TSDB_OPTION_TIMEZONE:
S
Shengliang Guan 已提交
627
      pItem = cfgGetItem(pCfg, "timezone");
H
Haojun Liao 已提交
628
      break;
629 630 631
    case TSDB_OPTION_USE_ADAPTER:
      pItem = cfgGetItem(pCfg, "useAdapter");
      break;
H
Haojun Liao 已提交
632
    default:
S
Shengliang Guan 已提交
633 634 635 636 637 638 639 640 641 642 643 644 645
      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 已提交
646 647 648
    if (TSDB_OPTION_SHELL_ACTIVITY_TIMER == option || TSDB_OPTION_USE_ADAPTER == option) {
      code = taosSetCfg(pCfg, pItem->name);
    }
H
Haojun Liao 已提交
649
  }
S
Shengliang Guan 已提交
650 651

  return code;
H
Haojun Liao 已提交
652 653
}

654 655 656 657 658
/**
 * The request id is an unsigned integer format of 64bit.
 *+------------+-----+-----------+---------------+
 *| uid|localIp| PId | timestamp | serial number |
 *+------------+-----+-----------+---------------+
659
 *| 12bit      |12bit|24bit      |16bit          |
660 661 662 663
 *+------------+-----+-----------+---------------+
 * @return
 */
uint64_t generateRequestId() {
H
Haojun Liao 已提交
664
  static uint64_t hashId = 0;
dengyihao's avatar
dengyihao 已提交
665
  static uint32_t requestSerialId = 0;
H
Haojun Liao 已提交
666 667 668 669 670 671 672 673 674 675 676

  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));
    }
677 678
  }

D
dapan1121 已提交
679
  uint64_t id = 0;
L
Liu Jicong 已提交
680

D
dapan1121 已提交
681 682 683
  while (true) {
    int64_t  ts = taosGetTimestampMs();
    uint64_t pid = taosGetPId();
dengyihao's avatar
dengyihao 已提交
684 685
    uint32_t val = atomic_add_fetch_32(&requestSerialId, 1);
    if (val >= 0xFFFF) atomic_store_32(&requestSerialId, 0);
686

D
dapan1121 已提交
687 688 689 690 691
    id = ((hashId & 0x0FFF) << 52) | ((pid & 0x0FFF) << 40) | ((ts & 0xFFFFFF) << 16) | (val & 0xFFFF);
    if (id) {
      break;
    }
  }
692 693 694
  return id;
}

H
Haojun Liao 已提交
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 743 744 745 746 747 748 749 750 751 752
#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 已提交
753
  taosThreadMutexLock(&setConfMutex);
H
Haojun Liao 已提交
754
  setConfRet ret = taos_set_config_imp(config);
wafwerar's avatar
wafwerar 已提交
755
  taosThreadMutexUnlock(&setConfMutex);
H
Haojun Liao 已提交
756 757
  return ret;
}
758
#endif