clientEnv.c 21.6 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"
X
Xiaoyu Wang 已提交
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 33
#include "ttime.h"

#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
X
Xiaoyu Wang 已提交
68
  if (pRequest == NULL) {
69 70 71
    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);
368
  tsem_destroy(&pRequest->body.rspSem);
H
Haojun Liao 已提交
369

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

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

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

H
Haojun Liao 已提交
382
  if (pRequest->syncQuery) {
X
Xiaoyu Wang 已提交
383 384 385
    if (pRequest->body.param) {
      tsem_destroy(&((SSyncQueryParam *)pRequest->body.param)->sem);
    }
H
Haojun Liao 已提交
386 387 388
    taosMemoryFree(pRequest->body.param);
  }

D
dapan1121 已提交
389
  taosMemoryFreeClear(pRequest->sqlstr);
390
  taosMemoryFree(pRequest);
D
dapan1121 已提交
391
  tscTrace("end to destroy request %" PRIx64 " p:%p", reqId, pRequest);
H
Haojun Liao 已提交
392 393
}

dengyihao's avatar
dengyihao 已提交
394
void destroyRequest(SRequestObj *pRequest) {
H
Haojun Liao 已提交
395 396 397 398
  if (pRequest == NULL) {
    return;
  }

D
dapan1121 已提交
399
  taos_stop_query(pRequest);
D
dapan1121 已提交
400
  removeRequest(pRequest->self);
H
Haojun Liao 已提交
401 402
}

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

D
dapan1121 已提交
405 406 407 408
static void *tscCrashReportThreadFp(void *param) {
  setThreadName("client-crashReport");
  char filepath[PATH_MAX] = {0};
  snprintf(filepath, sizeof(filepath), "%s%s.taosCrashLog", tsLogDir, TD_DIRSEP);
X
Xiaoyu Wang 已提交
409 410
  char     *pMsg = NULL;
  int64_t   msgLen = 0;
D
dapan1121 已提交
411
  TdFilePtr pFile = NULL;
X
Xiaoyu Wang 已提交
412 413 414 415
  bool      truncateFile = false;
  int32_t   sleepTime = 200;
  int32_t   reportPeriodNum = 3600 * 1000 / sleepTime;
  int32_t   loopTimes = reportPeriodNum;
D
dapan 已提交
416 417 418 419 420 421

#ifdef WINDOWS
  if (taosCheckCurrentInDll()) {
    atexit(crashReportThreadFuncUnexpectedStopped);
  }
#endif
X
Xiaoyu Wang 已提交
422

D
dapan1121 已提交
423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451
  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;
    }
X
Xiaoyu Wang 已提交
452

D
dapan1121 已提交
453 454 455 456
    if (pFile) {
      taosReleaseCrashLogFile(pFile, truncateFile);
      truncateFile = false;
    }
X
Xiaoyu Wang 已提交
457

D
dapan1121 已提交
458 459 460 461 462 463 464 465 466 467 468 469
    taosMsleep(sleepTime);
    loopTimes = 0;
  }

  clientStop = -1;
  return NULL;
}

int32_t tscCrashReportInit() {
  if (!tsEnableCrashReport) {
    return 0;
  }
X
Xiaoyu Wang 已提交
470

D
dapan1121 已提交
471 472 473
  TdThreadAttr thAttr;
  taosThreadAttrInit(&thAttr);
  taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
X
Xiaoyu Wang 已提交
474
  TdThread crashReportThread;
D
dapan1121 已提交
475 476 477 478 479 480 481 482 483 484 485 486 487 488
  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 已提交
489 490 491 492 493
  if (atomic_val_compare_exchange_32(&clientStop, 0, 1)) {
    tscDebug("hb thread already stopped");
    return;
  }

D
dapan1121 已提交
494 495 496 497 498
  while (atomic_load_32(&clientStop) > 0) {
    taosMsleep(100);
  }
}

D
dapan1121 已提交
499
void tscWriteCrashInfo(int signum, void *sigInfo, void *context) {
X
Xiaoyu Wang 已提交
500
  char       *pMsg = NULL;
D
dapan1121 已提交
501 502 503
  const char *flags = "UTL FATAL ";
  ELogLevel   level = DEBUG_FATAL;
  int32_t     dflag = 255;
X
Xiaoyu Wang 已提交
504
  int64_t     msgLen = -1;
D
dapan1121 已提交
505 506 507 508 509

  if (tsEnableCrashReport) {
    if (taosGenCrashJsonMsg(signum, &pMsg, lastClusterId, appInfo.startTime)) {
      taosPrintLog(flags, level, dflag, "failed to generate crash json msg");
    } else {
X
Xiaoyu Wang 已提交
510
      msgLen = strlen(pMsg);
D
dapan1121 已提交
511 512 513 514
    }
  }

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

H
Haojun Liao 已提交
517 518 519 520 521
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 已提交
522
  taosSeedRand(taosGetTimestampSec());
H
Haojun Liao 已提交
523

D
dapan1121 已提交
524 525 526 527
  appInfo.pid = taosGetPId();
  appInfo.startTime = taosGetTimestampMs();
  appInfo.pInstMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);

H
Haojun Liao 已提交
528
  deltaToUtcInitOnce();
S
Shengliang Guan 已提交
529

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

wafwerar's avatar
wafwerar 已提交
535
  if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 1) != 0) {
S
Shengliang Guan 已提交
536 537 538 539
    tscInitRes = -1;
    return;
  }

540
  initQueryModuleMsgHandle();
H
Haojun Liao 已提交
541

S
Shengliang Guan 已提交
542
  if (taosConvInit() != 0) {
543 544
    tscError("failed to init conv");
    return;
S
Shengliang Guan 已提交
545
  }
dengyihao's avatar
dengyihao 已提交
546

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

D
dapan1121 已提交
549
  SCatalogCfg cfg = {.maxDBCacheNum = 100, .maxTblCacheNum = 100};
550 551
  catalogInit(&cfg);

D
dapan1121 已提交
552
  schedulerInit();
S
Shengliang Guan 已提交
553
  tscDebug("starting to initialize TAOS driver");
H
Haojun Liao 已提交
554

wafwerar's avatar
wafwerar 已提交
555
#ifndef WINDOWS
H
Haojun Liao 已提交
556
  taosSetCoreDump(true);
wafwerar's avatar
wafwerar 已提交
557
#endif
H
Haojun Liao 已提交
558 559

  initTaskQueue();
560
  fmFuncMgtInit();
561
  nodesInitAllocatorSet();
H
Haojun Liao 已提交
562

563
  clientConnRefPool = taosOpenRef(200, destroyTscObj);
dengyihao's avatar
dengyihao 已提交
564
  clientReqRefPool = taosOpenRef(40960, doDestroyRequest);
H
Haojun Liao 已提交
565

dengyihao's avatar
dengyihao 已提交
566
  // transDestroyBuffer(&conn->readBuf);
H
Haojun Liao 已提交
567
  taosGetAppName(appInfo.appName, NULL);
wafwerar's avatar
wafwerar 已提交
568
  taosThreadMutexInit(&appInfo.mutex, NULL);
H
Haojun Liao 已提交
569

D
dapan1121 已提交
570
  tscCrashReportInit();
X
Xiaoyu Wang 已提交
571

H
Haojun Liao 已提交
572 573 574 575
  tscDebug("client is initialized successfully");
}

int taos_init() {
wafwerar's avatar
wafwerar 已提交
576
  taosThreadOnce(&tscinit, taos_init_imp);
H
Haojun Liao 已提交
577 578 579 580
  return tscInitRes;
}

int taos_options_imp(TSDB_OPTION option, const char *str) {
H
Haojun Liao 已提交
581
  if (option == TSDB_OPTION_CONFIGDIR) {
S
Shengliang Guan 已提交
582 583 584
    tstrncpy(configDir, str, PATH_MAX);
    tscInfo("set cfg:%s to %s", configDir, str);
    return 0;
H
Haojun Liao 已提交
585 586
  } else {
    taos_init();  // initialize global config
S
Shengliang Guan 已提交
587 588
  }

dengyihao's avatar
dengyihao 已提交
589
  SConfig     *pCfg = taosGetCfg();
S
Shengliang Guan 已提交
590
  SConfigItem *pItem = NULL;
H
Haojun Liao 已提交
591 592 593

  switch (option) {
    case TSDB_OPTION_SHELL_ACTIVITY_TIMER:
S
Shengliang Guan 已提交
594
      pItem = cfgGetItem(pCfg, "shellActivityTimer");
H
Haojun Liao 已提交
595
      break;
S
Shengliang Guan 已提交
596 597
    case TSDB_OPTION_LOCALE:
      pItem = cfgGetItem(pCfg, "locale");
H
Haojun Liao 已提交
598
      break;
S
Shengliang Guan 已提交
599 600
    case TSDB_OPTION_CHARSET:
      pItem = cfgGetItem(pCfg, "charset");
H
Haojun Liao 已提交
601 602
      break;
    case TSDB_OPTION_TIMEZONE:
S
Shengliang Guan 已提交
603
      pItem = cfgGetItem(pCfg, "timezone");
H
Haojun Liao 已提交
604
      break;
605 606 607
    case TSDB_OPTION_USE_ADAPTER:
      pItem = cfgGetItem(pCfg, "useAdapter");
      break;
H
Haojun Liao 已提交
608
    default:
S
Shengliang Guan 已提交
609 610 611 612 613 614 615 616 617 618 619 620 621
      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 已提交
622 623 624
    if (TSDB_OPTION_SHELL_ACTIVITY_TIMER == option || TSDB_OPTION_USE_ADAPTER == option) {
      code = taosSetCfg(pCfg, pItem->name);
    }
H
Haojun Liao 已提交
625
  }
S
Shengliang Guan 已提交
626 627

  return code;
H
Haojun Liao 已提交
628 629
}

630 631 632 633 634
/**
 * The request id is an unsigned integer format of 64bit.
 *+------------+-----+-----------+---------------+
 *| uid|localIp| PId | timestamp | serial number |
 *+------------+-----+-----------+---------------+
635
 *| 12bit      |12bit|24bit      |16bit          |
636 637 638 639
 *+------------+-----+-----------+---------------+
 * @return
 */
uint64_t generateRequestId() {
H
Haojun Liao 已提交
640
  static uint64_t hashId = 0;
dengyihao's avatar
dengyihao 已提交
641
  static uint32_t requestSerialId = 0;
H
Haojun Liao 已提交
642 643 644 645 646 647 648 649 650 651 652

  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));
    }
653 654
  }

D
dapan1121 已提交
655
  uint64_t id = 0;
L
Liu Jicong 已提交
656

D
dapan1121 已提交
657 658 659
  while (true) {
    int64_t  ts = taosGetTimestampMs();
    uint64_t pid = taosGetPId();
dengyihao's avatar
dengyihao 已提交
660 661
    uint32_t val = atomic_add_fetch_32(&requestSerialId, 1);
    if (val >= 0xFFFF) atomic_store_32(&requestSerialId, 0);
662

D
dapan1121 已提交
663 664 665 666 667
    id = ((hashId & 0x0FFF) << 52) | ((pid & 0x0FFF) << 40) | ((ts & 0xFFFFFF) << 16) | (val & 0xFFFF);
    if (id) {
      break;
    }
  }
668 669 670
  return id;
}

H
Haojun Liao 已提交
671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728
#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 已提交
729
  taosThreadMutexLock(&setConfMutex);
H
Haojun Liao 已提交
730
  setConfRet ret = taos_set_config_imp(config);
wafwerar's avatar
wafwerar 已提交
731
  taosThreadMutexUnlock(&setConfMutex);
H
Haojun Liao 已提交
732 733
  return ret;
}
734
#endif