clientInt.h 8.2 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22
/*
 * 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/>.
 */

#ifndef TDENGINE_CLIENTINT_H
#define TDENGINE_CLIENTINT_H

#ifdef __cplusplus
extern "C" {
#endif

dengyihao's avatar
dengyihao 已提交
23
#include "parser.h"
24
#include "planner.h"
dengyihao's avatar
dengyihao 已提交
25 26
#include "query.h"
#include "taos.h"
L
Liu Jicong 已提交
27
#include "tcommon.h"
H
Haojun Liao 已提交
28
#include "tdatablock.h"
29
#include "tdef.h"
30 31
#include "thash.h"
#include "tlist.h"
dengyihao's avatar
dengyihao 已提交
32
#include "tmsg.h"
33
#include "tmsgtype.h"
34
#include "trpc.h"
35

S
Shengliang Guan 已提交
36
#include "tconfig.h"
S
Shengliang Guan 已提交
37

38 39
#define CHECK_CODE_GOTO(expr, label) \
  do {                               \
D
dapan1121 已提交
40
    code = expr;                     \
41 42 43 44
    if (TSDB_CODE_SUCCESS != code) { \
      goto label;                    \
    }                                \
  } while (0)
45

L
Liu Jicong 已提交
46 47 48 49
#define HEARTBEAT_INTERVAL 1500  // ms

typedef struct SAppInstInfo SAppInstInfo;

L
Liu Jicong 已提交
50
typedef struct {
dengyihao's avatar
dengyihao 已提交
51 52
  void*         param;
  SClientHbReq* req;
D
dapan1121 已提交
53 54
} SHbConnInfo;

L
Liu Jicong 已提交
55
typedef struct {
dengyihao's avatar
dengyihao 已提交
56
  char* key;
L
Liu Jicong 已提交
57 58 59 60 61 62 63 64 65 66
  // statistics
  int32_t reportCnt;
  int32_t connKeyCnt;
  int64_t reportBytes;  // not implemented
  int64_t startTime;
  // ctl
  SRWLatch lock;  // lock is used in serialization
  // connection
  SAppInstInfo* pAppInstInfo;
  // info
dengyihao's avatar
dengyihao 已提交
67 68
  SHashObj* activeInfo;  // hash<SClientHbKey, SClientHbReq>
  SHashObj* connInfo;    // hash<SClientHbKey, SHbConnInfo>
L
Liu Jicong 已提交
69 70
} SAppHbMgr;

L
Liu Jicong 已提交
71
typedef int32_t (*FHbRspHandle)(SAppHbMgr* pAppHbMgr, SClientHbRsp* pRsp);
D
dapan1121 已提交
72

dengyihao's avatar
dengyihao 已提交
73
typedef int32_t (*FHbReqHandle)(SClientHbKey* connKey, void* param, SClientHbReq* req);
D
dapan1121 已提交
74

L
Liu Jicong 已提交
75
typedef struct {
L
Liu Jicong 已提交
76 77 78
  int8_t inited;
  // ctl
  int8_t          threadStop;
wafwerar's avatar
wafwerar 已提交
79 80
  TdThread       thread;
  TdThreadMutex lock;       // used when app init and cleanup
L
Liu Jicong 已提交
81
  SArray*         appHbMgrs;  // SArray<SAppHbMgr*> one for each cluster
D
dapan1121 已提交
82 83
  FHbReqHandle    reqHandle[HEARTBEAT_TYPE_MAX];
  FHbRspHandle    rspHandle[HEARTBEAT_TYPE_MAX];
L
Liu Jicong 已提交
84 85
} SClientHbMgr;

86
typedef struct SQueryExecMetric {
dengyihao's avatar
dengyihao 已提交
87 88 89 90
  int64_t start;   // start timestamp
  int64_t parsed;  // start to parse
  int64_t send;    // start to send to server
  int64_t rsp;     // receive response from server
91 92
} SQueryExecMetric;

93
typedef struct SInstanceSummary {
dengyihao's avatar
dengyihao 已提交
94 95 96 97 98 99 100 101 102 103
  uint64_t numOfInsertsReq;
  uint64_t numOfInsertRows;
  uint64_t insertElapsedTime;
  uint64_t insertBytes;  // submit to tsdb since launched.

  uint64_t fetchBytes;
  uint64_t queryElapsedTime;
  uint64_t numOfSlowQueries;
  uint64_t totalRequests;
  uint64_t currentRequests;  // the number of SRequestObj
104
} SInstanceSummary;
105 106

typedef struct SHeartBeatInfo {
dengyihao's avatar
dengyihao 已提交
107
  void* pTimer;  // timer, used to send request msg to mnode
108 109
} SHeartBeatInfo;

L
Liu Jicong 已提交
110
struct SAppInstInfo {
L
Liu Jicong 已提交
111 112 113 114 115 116 117
  int64_t          numOfConns;
  SCorEpSet        mgmtEp;
  SInstanceSummary summary;
  SList*           pConnList;  // STscObj linked list
  int64_t          clusterId;
  void*            pTransporter;
  SAppHbMgr*       pAppHbMgr;
L
Liu Jicong 已提交
118
};
119 120

typedef struct SAppInfo {
H
Haojun Liao 已提交
121 122
  int64_t         startTime;
  char            appName[TSDB_APP_NAME_LEN];
dengyihao's avatar
dengyihao 已提交
123
  char*           ep;
H
Haojun Liao 已提交
124 125
  int32_t         pid;
  int32_t         numOfThreads;
dengyihao's avatar
dengyihao 已提交
126
  SHashObj*       pInstMap;
wafwerar's avatar
wafwerar 已提交
127
  TdThreadMutex mutex;
128 129 130
} SAppInfo;

typedef struct STscObj {
dengyihao's avatar
dengyihao 已提交
131 132 133 134 135 136 137 138
  char            user[TSDB_USER_LEN];
  char            pass[TSDB_PASSWORD_LEN];
  char            db[TSDB_DB_FNAME_LEN];
  char            ver[128];
  int32_t         acctId;
  uint32_t        connId;
  int32_t         connType;
  uint64_t        id;         // ref ID returned by taosAddRef
H
Haojun Liao 已提交
139
  TdThreadMutex   mutex;      // used to protect the operation on db
dengyihao's avatar
dengyihao 已提交
140 141
  int32_t         numOfReqs;  // number of sqlObj bound to this connection
  SAppInstInfo*   pAppInfo;
142 143
} STscObj;

144 145 146 147 148 149 150 151
typedef struct SResultColumn {
  union {
    char*       nullbitmap;  // bitmap, one bit for each item in the list
    int32_t*    offset;
  };
  char*         pData;
} SResultColumn;

152
typedef struct SReqResultInfo {
153 154
  const char*    pRspMsg;
  const char*    pData;
H
Haojun Liao 已提交
155
  TAOS_FIELD*    fields;     // todo, column names are not needed.
156
  TAOS_FIELD*    userFields; // the fields info that return to user
157 158
  uint32_t       numOfCols;
  int32_t*       length;
159
  char**         convertBuf;
160 161 162 163 164 165
  TAOS_ROW       row;
  SResultColumn* pCol;
  uint32_t       numOfRows;
  uint64_t       totalRows;
  uint32_t       current;
  bool           completed;
H
Haojun Liao 已提交
166
  int32_t        precision;
167
  int32_t        payloadLen;
168 169
} SReqResultInfo;

170
typedef struct SShowReqInfo {
dengyihao's avatar
dengyihao 已提交
171 172 173 174
  int64_t execId;  // showId/queryId
  int32_t vgId;
  SArray* pArray;        // SArray<SVgroupInfo>
  int32_t currentIndex;  // current accessed vgroup index.
175 176
} SShowReqInfo;

177
typedef struct SRequestSendRecvBody {
178 179 180 181 182
  tsem_t             rspSem;  // not used now
  void*              fp;
  SShowReqInfo       showInfo;  // todo this attribute will be removed after the query framework being completed.
  SDataBuf           requestMsg;
  int64_t            queryJob;  // query job, created according to sql query DAG.
X
Xiaoyu Wang 已提交
183
  struct SQueryPlan* pDag;       // the query dag, generated according to the sql statement.
184
  SReqResultInfo     resInfo;
185
} SRequestSendRecvBody;
186

dengyihao's avatar
dengyihao 已提交
187
#define ERROR_MSG_BUF_DEFAULT_SIZE 512
188

189
typedef struct SRequestObj {
dengyihao's avatar
dengyihao 已提交
190 191 192
  uint64_t             requestId;
  int32_t              type;  // request type
  STscObj*             pTscObj;
X
Xiaoyu Wang 已提交
193
  char*                pDb;
dengyihao's avatar
dengyihao 已提交
194 195 196 197 198 199
  char*                sqlstr;  // sql string
  int32_t              sqlLen;
  int64_t              self;
  char*                msgBuf;
  void*                pInfo;  // sql parse info, generated by parser module
  int32_t              code;
D
dapan1121 已提交
200 201
  SArray*              dbList;
  SArray*              tableList;
dengyihao's avatar
dengyihao 已提交
202
  SQueryExecMetric     metric;
H
Haojun Liao 已提交
203
  SRequestSendRecvBody body;
204 205
} SRequestObj;

206
extern SAppInfo appInfo;
H
Haojun Liao 已提交
207
extern int32_t  clientReqRefPool;
208
extern int32_t  clientConnRefPool;
209

H
Haojun Liao 已提交
210
extern int (*handleRequestRspFp[TDMT_MAX])(void*, const SDataBuf* pMsg, int32_t code);
dengyihao's avatar
dengyihao 已提交
211
int           genericRspCallback(void* param, const SDataBuf* pMsg, int32_t code);
H
Haojun Liao 已提交
212
SMsgSendInfo* buildMsgInfoImpl(SRequestObj* pReqObj);
213

dengyihao's avatar
dengyihao 已提交
214
int taos_init();
215

dengyihao's avatar
dengyihao 已提交
216 217
void* createTscObj(const char* user, const char* auth, const char* db, SAppInstInfo* pAppInfo);
void  destroyTscObj(void* pObj);
218

219 220
uint64_t generateRequestId();

dengyihao's avatar
dengyihao 已提交
221
void* createRequest(STscObj* pObj, __taos_async_fn_t fp, void* param, int32_t type);
222 223
void  destroyRequest(SRequestObj* pRequest);

dengyihao's avatar
dengyihao 已提交
224
char* getDbOfConnection(STscObj* pObj);
225
void  setConnectionDB(STscObj* pTscObj, const char* db);
H
Haojun Liao 已提交
226
void  resetConnectDB(STscObj* pTscObj);
227 228

void taos_init_imp(void);
dengyihao's avatar
dengyihao 已提交
229
int  taos_options_imp(TSDB_OPTION option, const char* str);
230

dengyihao's avatar
dengyihao 已提交
231
void* openTransporter(const char* user, const char* auth, int32_t numOfThreads);
232

dengyihao's avatar
dengyihao 已提交
233
bool persistConnForSpecificMsg(void* parenct, tmsg_t msgType);
234
void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet);
235

236 237
void initMsgHandleFp();

dengyihao's avatar
dengyihao 已提交
238 239
TAOS* taos_connect_internal(const char* ip, const char* user, const char* pass, const char* auth, const char* db,
                            uint16_t port);
240

241
void* doFetchRow(SRequestObj* pRequest, bool setupOneRowPtr);
242

243
int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32_t numOfCols, int32_t numOfRows);
244

dengyihao's avatar
dengyihao 已提交
245
int32_t buildRequest(STscObj* pTscObj, const char* sql, int sqlLen, SRequestObj** pRequest);
246

X
Xiaoyu Wang 已提交
247
int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery);
248
int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArray* pNodeList);
249

dengyihao's avatar
dengyihao 已提交
250
// --- heartbeat
L
Liu Jicong 已提交
251 252 253 254 255 256
// global, called by mgmt
int  hbMgrInit();
void hbMgrCleanUp();
int  hbHandleRsp(SClientHbBatchRsp* hbRsp);

// cluster level
dengyihao's avatar
dengyihao 已提交
257 258
SAppHbMgr* appHbMgrInit(SAppInstInfo* pAppInstInfo, char* key);
void       appHbMgrCleanup(void);
L
Liu Jicong 已提交
259 260

// conn level
D
dapan1121 已提交
261
int  hbRegisterConn(SAppHbMgr* pAppHbMgr, int32_t connId, int64_t clusterId, int32_t hbType);
L
Liu Jicong 已提交
262 263 264 265 266 267 268
void hbDeregisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey);

int hbAddConnInfo(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, void* key, void* value, int32_t keyLen, int32_t valueLen);

// --- mq
void hbMgrInitMqHbRspHandle();

269 270 271 272 273
#ifdef __cplusplus
}
#endif

#endif  // TDENGINE_CLIENTINT_H