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 "tmsg.h"
H
Haojun Liao 已提交
27 28
#include "tref.h"
#include "trpc.h"
dengyihao's avatar
dengyihao 已提交
29
#include "tsched.h"
H
Haojun Liao 已提交
30
#include "ttime.h"
D
dapan1121 已提交
31
#include "thttp.h"
H
Haojun Liao 已提交
32 33

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

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

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

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

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

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

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

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

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

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

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

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

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

wmmhello's avatar
wmmhello 已提交
85
  tscPerf("insert duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
X
Xiaoyu Wang 已提交
86
          "us, exec:%" PRId64 "us, stmtType:%d",
wmmhello's avatar
wmmhello 已提交
87
          duration, pRequest->metric.syntaxEnd - pRequest->metric.syntaxStart,
X
Xiaoyu Wang 已提交
88 89 90
          pRequest->metric.ctgEnd - pRequest->metric.ctgStart, pRequest->metric.semanticEnd - pRequest->metric.ctgEnd,
          pRequest->metric.execEnd - pRequest->metric.semanticEnd, pRequest->stmtType);

X
Xiaoyu Wang 已提交
91
  if (QUERY_NODE_VNODE_MODIFY_STMT == pRequest->stmtType) {
X
Xiaoyu Wang 已提交
92 93 94 95 96 97
    //        tscPerf("insert duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
    //                "us, exec:%" PRId64 "us",
    //                duration, pRequest->metric.syntaxEnd - pRequest->metric.syntaxStart,
    //                pRequest->metric.ctgEnd - pRequest->metric.ctgStart, pRequest->metric.semanticEnd -
    //                pRequest->metric.ctgEnd, pRequest->metric.execEnd - pRequest->metric.semanticEnd);
    //    atomic_add_fetch_64((int64_t *)&pActivity->insertElapsedTime, duration);
D
dapan1121 已提交
98
  } else if (QUERY_NODE_SELECT_STMT == pRequest->stmtType) {
dengyihao's avatar
dengyihao 已提交
99 100 101 102 103 104
    //    tscPerf("select duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
    //            "us, planner:%" PRId64 "us, exec:%" PRId64 "us, reqId:0x%" PRIx64,
    //            duration, pRequest->metric.syntaxEnd - pRequest->metric.syntaxStart,
    //            pRequest->metric.ctgEnd - pRequest->metric.ctgStart, pRequest->metric.semanticEnd -
    //            pRequest->metric.ctgEnd, pRequest->metric.planEnd - pRequest->metric.semanticEnd,
    //            pRequest->metric.resultReady - pRequest->metric.planEnd, pRequest->requestId);
105

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

D
dapan1121 已提交
109 110 111
  if (duration >= SLOW_QUERY_INTERVAL) {
    atomic_add_fetch_64((int64_t *)&pActivity->numOfSlowQueries, 1);
  }
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 152 153
  SRpcInit rpcInit;
  memset(&rpcInit, 0, sizeof(rpcInit));
  rpcInit.localPort = 0;
  rpcInit.label = "TSC";
  rpcInit.numOfThreads = numOfThread;
  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
  void *pDnodeConn = rpcOpen(&rpcInit);
H
Haojun Liao 已提交
168 169 170 171 172 173 174 175
  if (pDnodeConn == NULL) {
    tscError("failed to init connection to server");
    return NULL;
  }

  return pDnodeConn;
}

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

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

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

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

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

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

  taosThreadMutexUnlock(&appInfo.mutex);

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

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

  taosMemoryFree(pAppInfo);
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

H
Haojun Liao 已提交
367
  doFreeReqResultInfo(&pRequest->body.resInfo);
H
Haojun Liao 已提交
368

X
Xiaoyu Wang 已提交
369 370
  taosArrayDestroy(pRequest->tableList);
  taosArrayDestroy(pRequest->dbList);
D
dapan1121 已提交
371
  taosArrayDestroy(pRequest->targetTableList);
X
Xiaoyu Wang 已提交
372
  qDestroyQuery(pRequest->pQuery);
373
  nodesDestroyAllocator(pRequest->allocatorRefId);
X
Xiaoyu Wang 已提交
374

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

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

H
Haojun Liao 已提交
381 382 383 384
  if (pRequest->syncQuery) {
    taosMemoryFree(pRequest->body.param);
  }

D
dapan1121 已提交
385
  taosMemoryFreeClear(pRequest->sqlstr);
386
  taosMemoryFree(pRequest);
D
dapan1121 已提交
387
  tscTrace("end to destroy request %" PRIx64 " p:%p", reqId, pRequest);
H
Haojun Liao 已提交
388 389
}

dengyihao's avatar
dengyihao 已提交
390
void destroyRequest(SRequestObj *pRequest) {
H
Haojun Liao 已提交
391 392 393 394
  if (pRequest == NULL) {
    return;
  }

D
dapan1121 已提交
395
  taos_stop_query(pRequest);
D
dapan1121 已提交
396
  removeRequest(pRequest->self);
H
Haojun Liao 已提交
397 398
}

D
dapan1121 已提交
399 400 401 402 403
void taosClientCrash(int signum, void *sigInfo, void *context) {
  taosIgnSignal(SIGTERM);
  taosIgnSignal(SIGHUP);
  taosIgnSignal(SIGINT);
  taosIgnSignal(SIGBREAK);
D
dapan1121 已提交
404 405

#if !defined(WINDOWS)
D
dapan1121 已提交
406
  taosIgnSignal(SIGBUS);
D
dapan1121 已提交
407
#endif  
D
dapan1121 已提交
408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429
  taosIgnSignal(SIGABRT);
  taosIgnSignal(SIGFPE);
  taosIgnSignal(SIGSEGV);

  char *pMsg = NULL;
  const char *flags = "UTL FATAL ";
  ELogLevel   level = DEBUG_FATAL;
  int32_t     dflag = 255;
  int64_t     msgLen= -1;

  if (tsEnableCrashReport) {
    if (taosGenCrashJsonMsg(signum, &pMsg, lastClusterId, appInfo.startTime)) {
      taosPrintLog(flags, level, dflag, "failed to generate crash json msg");
      goto _return;
    } else {
      msgLen = strlen(pMsg);  
    }
  }
  
_return:

  taosLogCrashInfo("taos", pMsg, msgLen, signum, sigInfo);
D
dapan 已提交
430

D
dapan1121 已提交
431
#ifdef _TD_DARWIN_64
D
dapan 已提交
432
  exit(signum);
D
dapan1121 已提交
433 434 435
#elif defined(WINDOWS)
  exit(signum);
#endif
D
dapan1121 已提交
436 437
}

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

D
dapan1121 已提交
440 441 442 443 444 445 446 447 448 449 450
static void *tscCrashReportThreadFp(void *param) {
  setThreadName("client-crashReport");
  char filepath[PATH_MAX] = {0};
  snprintf(filepath, sizeof(filepath), "%s%s.taosCrashLog", tsLogDir, TD_DIRSEP);
  char *pMsg = NULL;
  int64_t msgLen = 0;
  TdFilePtr pFile = NULL;
  bool truncateFile = false;
  int32_t sleepTime = 200;
  int32_t reportPeriodNum = 3600 * 1000 / sleepTime;
  int32_t loopTimes = reportPeriodNum;
D
dapan 已提交
451 452 453 454 455 456

#ifdef WINDOWS
  if (taosCheckCurrentInDll()) {
    atexit(crashReportThreadFuncUnexpectedStopped);
  }
#endif
D
dapan1121 已提交
457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523
  
  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;
    }
    
    if (pFile) {
      taosReleaseCrashLogFile(pFile, truncateFile);
      truncateFile = false;
    }
    
    taosMsleep(sleepTime);
    loopTimes = 0;
  }

  clientStop = -1;
  return NULL;
}

int32_t tscCrashReportInit() {
  if (!tsEnableCrashReport) {
    return 0;
  }
  
  TdThreadAttr thAttr;
  taosThreadAttrInit(&thAttr);
  taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
  TdThread            crashReportThread;
  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 已提交
524 525 526 527 528
  if (atomic_val_compare_exchange_32(&clientStop, 0, 1)) {
    tscDebug("hb thread already stopped");
    return;
  }

D
dapan1121 已提交
529 530 531 532 533 534
  while (atomic_load_32(&clientStop) > 0) {
    taosMsleep(100);
  }
}

static void tscSetSignalHandle() {
D
dapan1121 已提交
535
#if !defined(WINDOWS)
D
dapan1121 已提交
536
  taosSetSignal(SIGBUS, taosClientCrash);
D
dapan1121 已提交
537
#endif
D
dapan1121 已提交
538 539 540 541 542
  taosSetSignal(SIGABRT, taosClientCrash);
  taosSetSignal(SIGFPE, taosClientCrash);
  taosSetSignal(SIGSEGV, taosClientCrash);
}

H
Haojun Liao 已提交
543 544 545 546 547
void taos_init_imp(void) {
  // 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 已提交
548
  taosSeedRand(taosGetTimestampSec());
H
Haojun Liao 已提交
549

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

H
Haojun Liao 已提交
554
  deltaToUtcInitOnce();
S
Shengliang Guan 已提交
555

wafwerar's avatar
wafwerar 已提交
556
  if (taosCreateLog("taoslog", 10, configDir, NULL, NULL, NULL, NULL, 1) != 0) {
A
Alex Duan 已提交
557 558
    // ignore create log failed, only print
    printf(" WARING: Create taoslog failed. configDir=%s\n", configDir);
S
Shengliang Guan 已提交
559 560
  }

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

D
dapan1121 已提交
566 567
  tscSetSignalHandle();

568
  initQueryModuleMsgHandle();
H
Haojun Liao 已提交
569

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

S
Shengliang Guan 已提交
575
  rpcInit();
H
Haojun Liao 已提交
576

D
dapan1121 已提交
577
  SCatalogCfg cfg = {.maxDBCacheNum = 100, .maxTblCacheNum = 100};
578 579
  catalogInit(&cfg);

D
dapan1121 已提交
580
  schedulerInit();
S
Shengliang Guan 已提交
581
  tscDebug("starting to initialize TAOS driver");
H
Haojun Liao 已提交
582

wafwerar's avatar
wafwerar 已提交
583
#ifndef WINDOWS
H
Haojun Liao 已提交
584
  taosSetCoreDump(true);
wafwerar's avatar
wafwerar 已提交
585
#endif
H
Haojun Liao 已提交
586 587

  initTaskQueue();
588
  fmFuncMgtInit();
589
  nodesInitAllocatorSet();
H
Haojun Liao 已提交
590

591
  clientConnRefPool = taosOpenRef(200, destroyTscObj);
dengyihao's avatar
dengyihao 已提交
592
  clientReqRefPool = taosOpenRef(40960, doDestroyRequest);
H
Haojun Liao 已提交
593

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

D
dapan1121 已提交
598 599
  tscCrashReportInit();
  
H
Haojun Liao 已提交
600 601 602 603
  tscDebug("client is initialized successfully");
}

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

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

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

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