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 113 114 115
    if (tsSlowLogScope & reqType) {
      taosPrintSlowLog("PID:%d, Conn:%u, QID:0x%" PRIx64 ", Start:%" PRId64 ", Duration:%" PRId64 "us, SQL:%s", 
        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->pAppInfo->pAppHbMgr, connKey, pTscObj->passInfo.fp);
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 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467
    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 已提交
468

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

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

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

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

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

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

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

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

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

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

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

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

H
Haojun Liao 已提交
556
  deltaToUtcInitOnce();
S
Shengliang Guan 已提交
557

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

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

574
  initQueryModuleMsgHandle();
H
Haojun Liao 已提交
575

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

S
Shengliang Guan 已提交
581
  rpcInit();
H
Haojun Liao 已提交
582

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

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

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

  initTaskQueue();
594
  fmFuncMgtInit();
595
  nodesInitAllocatorSet();
H
Haojun Liao 已提交
596

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

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

D
dapan1121 已提交
604
  tscCrashReportInit();
dengyihao's avatar
dengyihao 已提交
605

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

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

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

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

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

  return code;
H
Haojun Liao 已提交
662 663
}

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

  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));
    }
687 688
  }

D
dapan1121 已提交
689
  uint64_t id = 0;
L
Liu Jicong 已提交
690

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

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

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