clientEnv.c 22.2 KB
Newer Older
H
Haojun Liao 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

16
#include "catalog.h"
H
Haojun Liao 已提交
17 18
#include "clientInt.h"
#include "clientLog.h"
dengyihao's avatar
dengyihao 已提交
19 20
#include "functionMgt.h"
#include "os.h"
21
#include "query.h"
X
Xiaoyu Wang 已提交
22
#include "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;
45
int32_t  clientStop = -1;
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) {
X
Xiaoyu Wang 已提交
72
  if (pRequest == NULL) {
73 74 75
    tscError("pRequest == NULL");
    return;
  }
H
Haojun Liao 已提交
76

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

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

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
      atomic_add_fetch_64((int64_t *)&pActivity->insertElapsedTime, duration);
98
      reqType = SLOW_LOG_TYPE_INSERT;
99 100
    } 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 已提交
101 102 103
               "us, planCost:%" PRId64 "us, exec:%" PRId64 "us",
               duration, pRequest->metric.parseCostUs, pRequest->metric.ctgCostUs, pRequest->metric.analyseCostUs,
               pRequest->metric.planCostUs, pRequest->metric.execCostUs);
104 105

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

110
  if (duration >= (tsSlowLogThreshold * 1000000UL)) {
D
dapan1121 已提交
111
    atomic_add_fetch_64((int64_t *)&pActivity->numOfSlowQueries, 1);
112
    if (tsSlowLogScope & reqType) {
H
Haojun Liao 已提交
113
      taosPrintSlowLog("PID:%d, Conn:%u, QID:0x%" PRIx64 ", Start:%" PRId64 ", Duration:%" PRId64 "us, SQL:%s",
114 115
        taosGetPId(), pTscObj->connId, pRequest->requestId, pRequest->metric.start, duration, pRequest->sqlstr);
    }
D
dapan1121 已提交
116
  }
dengyihao's avatar
dengyihao 已提交
117

D
dapan1121 已提交
118
  releaseTscObj(pTscObj->id);
H
Haojun Liao 已提交
119 120
}

121
// todo close the transporter properly
D
dapan1121 已提交
122 123
void closeTransporter(SAppInstInfo *pAppInfo) {
  if (pAppInfo == NULL || pAppInfo->pTransporter == NULL) {
H
Haojun Liao 已提交
124 125 126
    return;
  }

D
dapan1121 已提交
127 128
  tscDebug("free transporter:%p in app inst %p", pAppInfo->pTransporter, pAppInfo);
  rpcClose(pAppInfo->pTransporter);
H
Haojun Liao 已提交
129 130
}

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

dengyihao's avatar
dengyihao 已提交
143 144
// start timer for particular msgType
static bool clientRpcTfp(int32_t code, tmsg_t msgType) {
dengyihao's avatar
dengyihao 已提交
145 146 147
  if (msgType == TDMT_VND_SUBMIT || msgType == TDMT_VND_CREATE_TABLE) {
    return true;
  }
dengyihao's avatar
dengyihao 已提交
148 149 150
  return false;
}

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

dengyihao's avatar
dengyihao 已提交
167 168 169 170
  rpcInit.retryMinInterval = tsRedirectPeriod;
  rpcInit.retryStepFactor = tsRedirectFactor;
  rpcInit.retryMaxInterval = tsRedirectMaxPeriod;
  rpcInit.retryMaxTimouet = tsMaxRetryWaitTime;
dengyihao's avatar
dengyihao 已提交
171

dengyihao's avatar
dengyihao 已提交
172
  int32_t connLimitNum = tsNumOfRpcSessions / (tsNumOfRpcThreads * 3);
dengyihao's avatar
dengyihao 已提交
173
  connLimitNum = TMAX(connLimitNum, 10);
dengyihao's avatar
dengyihao 已提交
174
  connLimitNum = TMIN(connLimitNum, 1000);
dengyihao's avatar
dengyihao 已提交
175
  rpcInit.connLimitNum = connLimitNum;
dengyihao's avatar
dengyihao 已提交
176
  rpcInit.timeToGetConn = tsTimeToGetAvailableConn;
dengyihao's avatar
dengyihao 已提交
177

dengyihao's avatar
dengyihao 已提交
178
  void *pDnodeConn = rpcOpen(&rpcInit);
H
Haojun Liao 已提交
179 180 181 182 183 184 185 186
  if (pDnodeConn == NULL) {
    tscError("failed to init connection to server");
    return NULL;
  }

  return pDnodeConn;
}

D
dapan1121 已提交
187
void destroyAllRequests(SHashObj *pRequests) {
L
Liu Jicong 已提交
188
  void *pIter = taosHashIterate(pRequests, NULL);
D
dapan1121 已提交
189 190 191
  while (pIter != NULL) {
    int64_t *rid = pIter;

D
dapan1121 已提交
192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211
    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 已提交
212

D
dapan1121 已提交
213 214 215 216
    pIter = taosHashIterate(pRequests, pIter);
  }
}

dengyihao's avatar
dengyihao 已提交
217
void destroyAppInst(SAppInstInfo *pAppInfo) {
D
dapan1121 已提交
218
  tscDebug("destroy app inst mgr %p", pAppInfo);
D
dapan1121 已提交
219 220

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

D
dapan1121 已提交
222 223
  hbRemoveAppHbMrg(&pAppInfo->pAppHbMgr);
  taosHashRemove(appInfo.pInstMap, pAppInfo->instKey, strlen(pAppInfo->instKey));
D
dapan1121 已提交
224 225 226

  taosThreadMutexUnlock(&appInfo.mutex);

D
dapan1121 已提交
227 228
  taosMemoryFreeClear(pAppInfo->instKey);
  closeTransporter(pAppInfo);
dengyihao's avatar
dengyihao 已提交
229

D
dapan1121 已提交
230 231
  taosThreadMutexLock(&pAppInfo->qnodeMutex);
  taosArrayDestroy(pAppInfo->pQnodeList);
dengyihao's avatar
dengyihao 已提交
232
  taosThreadMutexUnlock(&pAppInfo->qnodeMutex);
D
dapan1121 已提交
233 234 235 236

  taosMemoryFree(pAppInfo);
}

H
Haojun Liao 已提交
237
void destroyTscObj(void *pObj) {
D
dapan1121 已提交
238 239 240
  if (NULL == pObj) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
241

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

D
dapan1121 已提交
246
  SClientHbKey connKey = {.tscRid = pTscObj->id, .connType = pTscObj->connType};
K
kailixu 已提交
247
  hbDeregisterConn(pTscObj, connKey);
248

D
dapan1121 已提交
249
  destroyAllRequests(pTscObj->pRequests);
250
  taosHashCleanup(pTscObj->pRequests);
dengyihao's avatar
dengyihao 已提交
251

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

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

wafwerar's avatar
wafwerar 已提交
259
  taosThreadMutexDestroy(&pTscObj->mutex);
D
dapan1121 已提交
260
  taosMemoryFree(pTscObj);
D
dapan1121 已提交
261

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

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

D
dapan1121 已提交
272 273 274
  pObj->pRequests = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
  if (NULL == pObj->pRequests) {
    taosMemoryFree(pObj);
S
Shengliang Guan 已提交
275
    terrno = TSDB_CODE_OUT_OF_MEMORY;
D
dapan1121 已提交
276 277
    return NULL;
  }
L
Liu Jicong 已提交
278

279
  pObj->connType = connType;
H
Haojun Liao 已提交
280 281 282 283
  pObj->pAppInfo = pAppInfo;
  tstrncpy(pObj->user, user, sizeof(pObj->user));
  memcpy(pObj->pass, auth, TSDB_PASSWORD_LEN);

284 285 286 287
  if (db != NULL) {
    tstrncpy(pObj->db, db, tListLen(pObj->db));
  }

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

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

D
dapan1121 已提交
293
  tscDebug("connObj created, 0x%" PRIx64 ",p:%p", pObj->id, pObj);
H
Haojun Liao 已提交
294 295 296
  return pObj;
}

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

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

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

dengyihao's avatar
dengyihao 已提交
308
  STscObj *pTscObj = acquireTscObj(connId);
309
  if (pTscObj == NULL) {
310
    taosMemoryFree(pRequest);
311 312 313 314
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
  }

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

319
  pRequest->body.resInfo.convertUcs4 = true;  // convert ucs4 by default
dengyihao's avatar
dengyihao 已提交
320
  pRequest->type = type;
321
  pRequest->allocatorRefId = -1;
322 323 324 325

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

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

330
  if (registerRequest(pRequest, pTscObj)) {
D
dapan1121 已提交
331 332 333
    doDestroyRequest(pRequest);
    return NULL;
  }
L
Liu Jicong 已提交
334

H
Haojun Liao 已提交
335 336 337
  return pRequest;
}

L
Liu Jicong 已提交
338
void doFreeReqResultInfo(SReqResultInfo *pResInfo) {
wafwerar's avatar
wafwerar 已提交
339 340 341 342 343
  taosMemoryFreeClear(pResInfo->pRspMsg);
  taosMemoryFreeClear(pResInfo->length);
  taosMemoryFreeClear(pResInfo->row);
  taosMemoryFreeClear(pResInfo->pCol);
  taosMemoryFreeClear(pResInfo->fields);
344
  taosMemoryFreeClear(pResInfo->userFields);
wmmhello's avatar
wmmhello 已提交
345
  taosMemoryFreeClear(pResInfo->convertJson);
346 347 348

  if (pResInfo->convertBuf != NULL) {
    for (int32_t i = 0; i < pResInfo->numOfCols; ++i) {
wafwerar's avatar
wafwerar 已提交
349
      taosMemoryFreeClear(pResInfo->convertBuf[i]);
350
    }
wafwerar's avatar
wafwerar 已提交
351
    taosMemoryFreeClear(pResInfo->convertBuf);
352
  }
H
Haojun Liao 已提交
353 354
}

D
dapan1121 已提交
355 356 357 358 359 360
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 已提交
361
void doDestroyRequest(void *p) {
D
dapan1121 已提交
362 363 364
  if (NULL == p) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
365

dengyihao's avatar
dengyihao 已提交
366
  SRequestObj *pRequest = (SRequestObj *)p;
dengyihao's avatar
dengyihao 已提交
367

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

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

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

wafwerar's avatar
wafwerar 已提交
375 376
  taosMemoryFreeClear(pRequest->msgBuf);
  taosMemoryFreeClear(pRequest->pDb);
H
Haojun Liao 已提交
377

H
Haojun Liao 已提交
378
  doFreeReqResultInfo(&pRequest->body.resInfo);
379
  tsem_destroy(&pRequest->body.rspSem);
H
Haojun Liao 已提交
380

X
Xiaoyu Wang 已提交
381 382
  taosArrayDestroy(pRequest->tableList);
  taosArrayDestroy(pRequest->dbList);
D
dapan1121 已提交
383
  taosArrayDestroy(pRequest->targetTableList);
X
Xiaoyu Wang 已提交
384

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

D
dapan1121 已提交
387 388 389
  if (pRequest->self) {
    deregisterRequest(pRequest);
  }
D
dapan1121 已提交
390

H
Haojun Liao 已提交
391
  if (pRequest->syncQuery) {
dengyihao's avatar
dengyihao 已提交
392 393 394
    if (pRequest->body.param) {
      tsem_destroy(&((SSyncQueryParam *)pRequest->body.param)->sem);
    }
H
Haojun Liao 已提交
395 396 397
    taosMemoryFree(pRequest->body.param);
  }

398 399 400
  qDestroyQuery(pRequest->pQuery);
  nodesDestroyAllocator(pRequest->allocatorRefId);

D
dapan1121 已提交
401
  taosMemoryFreeClear(pRequest->sqlstr);
402
  taosMemoryFree(pRequest);
D
dapan1121 已提交
403
  tscTrace("end to destroy request %" PRIx64 " p:%p", reqId, pRequest);
H
Haojun Liao 已提交
404 405
}

dengyihao's avatar
dengyihao 已提交
406
void destroyRequest(SRequestObj *pRequest) {
H
Haojun Liao 已提交
407 408 409 410
  if (pRequest == NULL) {
    return;
  }

D
dapan1121 已提交
411
  taos_stop_query(pRequest);
D
dapan1121 已提交
412
  removeRequest(pRequest->self);
H
Haojun Liao 已提交
413 414
}

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

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

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

435 436 437 438
  if (-1 != atomic_val_compare_exchange_32(&clientStop, -1, 0)) {
    return NULL;
  }

D
dapan1121 已提交
439
  while (1) {
440
    if (clientStop > 0) break;
D
dapan1121 已提交
441 442 443 444 445 446 447 448 449 450 451
    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);
D
dapan1121 已提交
452
          pFile = NULL;
D
dapan1121 已提交
453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468
          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 已提交
469

D
dapan1121 已提交
470 471
    if (pFile) {
      taosReleaseCrashLogFile(pFile, truncateFile);
D
dapan1121 已提交
472
      pFile = NULL;
D
dapan1121 已提交
473 474
      truncateFile = false;
    }
dengyihao's avatar
dengyihao 已提交
475

D
dapan1121 已提交
476 477 478 479
    taosMsleep(sleepTime);
    loopTimes = 0;
  }

480
  clientStop = -2;
D
dapan1121 已提交
481 482 483 484 485 486 487
  return NULL;
}

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

D
dapan1121 已提交
489 490 491
  TdThreadAttr thAttr;
  taosThreadAttrInit(&thAttr);
  taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
dengyihao's avatar
dengyihao 已提交
492
  TdThread crashReportThread;
D
dapan1121 已提交
493 494 495 496 497 498 499 500 501 502 503 504 505 506
  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 已提交
507 508 509 510 511
  if (atomic_val_compare_exchange_32(&clientStop, 0, 1)) {
    tscDebug("hb thread already stopped");
    return;
  }

D
dapan1121 已提交
512 513 514 515 516
  while (atomic_load_32(&clientStop) > 0) {
    taosMsleep(100);
  }
}

D
dapan1121 已提交
517
void tscWriteCrashInfo(int signum, void *sigInfo, void *context) {
dengyihao's avatar
dengyihao 已提交
518
  char       *pMsg = NULL;
D
dapan1121 已提交
519 520 521
  const char *flags = "UTL FATAL ";
  ELogLevel   level = DEBUG_FATAL;
  int32_t     dflag = 255;
dengyihao's avatar
dengyihao 已提交
522
  int64_t     msgLen = -1;
D
dapan1121 已提交
523 524 525 526 527

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

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

H
Haojun Liao 已提交
535
void taos_init_imp(void) {
D
dapan1121 已提交
536 537 538 539 540 541
#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 已提交
542
      tsAsyncLog = false;
D
dapan1121 已提交
543 544 545 546 547
      printf("memory dbg enabled\n");
    }
  }
#endif

H
Haojun Liao 已提交
548 549 550 551
  // 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 已提交
552
  taosSeedRand(taosGetTimestampSec());
H
Haojun Liao 已提交
553

D
dapan1121 已提交
554 555 556 557
  appInfo.pid = taosGetPId();
  appInfo.startTime = taosGetTimestampMs();
  appInfo.pInstMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);

H
Haojun Liao 已提交
558
  deltaToUtcInitOnce();
S
Shengliang Guan 已提交
559

560 561 562 563 564 565 566
  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 已提交
567
    // ignore create log failed, only print
568
    printf(" WARING: Create %s failed:%s. configDir=%s\n", logDirName, strerror(errno), configDir);
S
Shengliang Guan 已提交
569 570
  }

wafwerar's avatar
wafwerar 已提交
571
  if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 1) != 0) {
S
Shengliang Guan 已提交
572 573 574 575
    tscInitRes = -1;
    return;
  }

576
  initQueryModuleMsgHandle();
H
Haojun Liao 已提交
577

S
Shengliang Guan 已提交
578
  if (taosConvInit() != 0) {
579 580
    tscError("failed to init conv");
    return;
S
Shengliang Guan 已提交
581
  }
dengyihao's avatar
dengyihao 已提交
582

S
Shengliang Guan 已提交
583
  rpcInit();
H
Haojun Liao 已提交
584

D
dapan1121 已提交
585
  SCatalogCfg cfg = {.maxDBCacheNum = 100, .maxTblCacheNum = 100};
586 587
  catalogInit(&cfg);

D
dapan1121 已提交
588
  schedulerInit();
S
Shengliang Guan 已提交
589
  tscDebug("starting to initialize TAOS driver");
H
Haojun Liao 已提交
590

wafwerar's avatar
wafwerar 已提交
591
#ifndef WINDOWS
H
Haojun Liao 已提交
592
  taosSetCoreDump(true);
wafwerar's avatar
wafwerar 已提交
593
#endif
H
Haojun Liao 已提交
594 595

  initTaskQueue();
596
  fmFuncMgtInit();
597
  nodesInitAllocatorSet();
H
Haojun Liao 已提交
598

599
  clientConnRefPool = taosOpenRef(200, destroyTscObj);
dengyihao's avatar
dengyihao 已提交
600
  clientReqRefPool = taosOpenRef(40960, doDestroyRequest);
H
Haojun Liao 已提交
601

dengyihao's avatar
dengyihao 已提交
602
  // transDestroyBuffer(&conn->readBuf);
H
Haojun Liao 已提交
603
  taosGetAppName(appInfo.appName, NULL);
wafwerar's avatar
wafwerar 已提交
604
  taosThreadMutexInit(&appInfo.mutex, NULL);
H
Haojun Liao 已提交
605

D
dapan1121 已提交
606
  tscCrashReportInit();
dengyihao's avatar
dengyihao 已提交
607

H
Haojun Liao 已提交
608 609 610 611
  tscDebug("client is initialized successfully");
}

int taos_init() {
wafwerar's avatar
wafwerar 已提交
612
  taosThreadOnce(&tscinit, taos_init_imp);
H
Haojun Liao 已提交
613 614 615 616
  return tscInitRes;
}

int taos_options_imp(TSDB_OPTION option, const char *str) {
H
Haojun Liao 已提交
617
  if (option == TSDB_OPTION_CONFIGDIR) {
S
Shengliang Guan 已提交
618 619 620
    tstrncpy(configDir, str, PATH_MAX);
    tscInfo("set cfg:%s to %s", configDir, str);
    return 0;
H
Haojun Liao 已提交
621 622
  } else {
    taos_init();  // initialize global config
S
Shengliang Guan 已提交
623 624
  }

dengyihao's avatar
dengyihao 已提交
625
  SConfig     *pCfg = taosGetCfg();
S
Shengliang Guan 已提交
626
  SConfigItem *pItem = NULL;
H
Haojun Liao 已提交
627 628 629

  switch (option) {
    case TSDB_OPTION_SHELL_ACTIVITY_TIMER:
S
Shengliang Guan 已提交
630
      pItem = cfgGetItem(pCfg, "shellActivityTimer");
H
Haojun Liao 已提交
631
      break;
S
Shengliang Guan 已提交
632 633
    case TSDB_OPTION_LOCALE:
      pItem = cfgGetItem(pCfg, "locale");
H
Haojun Liao 已提交
634
      break;
S
Shengliang Guan 已提交
635 636
    case TSDB_OPTION_CHARSET:
      pItem = cfgGetItem(pCfg, "charset");
H
Haojun Liao 已提交
637 638
      break;
    case TSDB_OPTION_TIMEZONE:
S
Shengliang Guan 已提交
639
      pItem = cfgGetItem(pCfg, "timezone");
H
Haojun Liao 已提交
640
      break;
641 642 643
    case TSDB_OPTION_USE_ADAPTER:
      pItem = cfgGetItem(pCfg, "useAdapter");
      break;
H
Haojun Liao 已提交
644
    default:
S
Shengliang Guan 已提交
645 646 647 648 649 650 651 652 653 654 655 656 657
      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 已提交
658
    if (TSDB_OPTION_SHELL_ACTIVITY_TIMER == option || TSDB_OPTION_USE_ADAPTER == option) {
D
dapan1121 已提交
659
      code = taosApplyLocalCfg(pCfg, pItem->name);
X
Xiaoyu Wang 已提交
660
    }
H
Haojun Liao 已提交
661
  }
S
Shengliang Guan 已提交
662 663

  return code;
H
Haojun Liao 已提交
664 665
}

666 667 668 669 670
/**
 * The request id is an unsigned integer format of 64bit.
 *+------------+-----+-----------+---------------+
 *| uid|localIp| PId | timestamp | serial number |
 *+------------+-----+-----------+---------------+
671
 *| 12bit      |12bit|24bit      |16bit          |
672 673 674 675
 *+------------+-----+-----------+---------------+
 * @return
 */
uint64_t generateRequestId() {
H
Haojun Liao 已提交
676
  static uint64_t hashId = 0;
dengyihao's avatar
dengyihao 已提交
677
  static uint32_t requestSerialId = 0;
H
Haojun Liao 已提交
678 679 680 681 682 683 684 685 686 687 688

  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));
    }
689 690
  }

D
dapan1121 已提交
691
  uint64_t id = 0;
L
Liu Jicong 已提交
692

D
dapan1121 已提交
693 694 695
  while (true) {
    int64_t  ts = taosGetTimestampMs();
    uint64_t pid = taosGetPId();
dengyihao's avatar
dengyihao 已提交
696 697
    uint32_t val = atomic_add_fetch_32(&requestSerialId, 1);
    if (val >= 0xFFFF) atomic_store_32(&requestSerialId, 0);
698

D
dapan1121 已提交
699 700 701 702 703
    id = ((hashId & 0x0FFF) << 52) | ((pid & 0x0FFF) << 40) | ((ts & 0xFFFFFF) << 16) | (val & 0xFFFF);
    if (id) {
      break;
    }
  }
704 705 706
  return id;
}

H
Haojun Liao 已提交
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 754 755 756 757 758 759 760 761 762 763 764
#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 已提交
765
  taosThreadMutexLock(&setConfMutex);
H
Haojun Liao 已提交
766
  setConfRet ret = taos_set_config_imp(config);
wafwerar's avatar
wafwerar 已提交
767
  taosThreadMutexUnlock(&setConfMutex);
H
Haojun Liao 已提交
768 769
  return ret;
}
770
#endif