clientEnv.c 21.9 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"

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
  if (duration >= SLOW_QUERY_INTERVAL) {
    atomic_add_fetch_64((int64_t *)&pActivity->numOfSlowQueries, 1);
D
dapan1121 已提交
110
    tscWarnL("slow query: %s, duration:%" PRId64, pRequest->sqlstr, duration);
D
dapan1121 已提交
111
  }
dengyihao's avatar
dengyihao 已提交
112

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

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

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

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

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

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

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

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

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

  return pDnodeConn;
}

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

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

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

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

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

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

  taosThreadMutexUnlock(&appInfo.mutex);

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

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

  taosMemoryFree(pAppInfo);
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

D
dapan1121 已提交
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 458
  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 已提交
459

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

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

  clientStop = -1;
  return NULL;
}

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

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

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

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

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

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

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

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

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

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

549 550 551 552 553 554 555
  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 已提交
556
    // ignore create log failed, only print
557
    printf(" WARING: Create %s failed:%s. configDir=%s\n", logDirName, strerror(errno), configDir);
S
Shengliang Guan 已提交
558 559
  }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

H
Haojun Liao 已提交
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 753
#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 已提交
754
  taosThreadMutexLock(&setConfMutex);
H
Haojun Liao 已提交
755
  setConfRet ret = taos_set_config_imp(config);
wafwerar's avatar
wafwerar 已提交
756
  taosThreadMutexUnlock(&setConfMutex);
H
Haojun Liao 已提交
757 758
  return ret;
}
759
#endif