提交 e52fd48d 编写于 作者: H Hongze Cheng

Merge branch 'develop' into feature/2.0tsdb

...@@ -264,6 +264,7 @@ bool hasMoreVnodesToTry(SSqlObj *pSql); ...@@ -264,6 +264,7 @@ bool hasMoreVnodesToTry(SSqlObj *pSql);
void tscTryQueryNextVnode(SSqlObj *pSql, __async_cb_func_t fp); void tscTryQueryNextVnode(SSqlObj *pSql, __async_cb_func_t fp);
void tscAsyncQuerySingleRowForNextVnode(void *param, TAOS_RES *tres, int numOfRows); void tscAsyncQuerySingleRowForNextVnode(void *param, TAOS_RES *tres, int numOfRows);
void tscTryQueryNextClause(SSqlObj* pSql, void (*queryFp)()); void tscTryQueryNextClause(SSqlObj* pSql, void (*queryFp)());
int tscSetMgmtIpListFromCfg(const char *first, const char *second);
void* malloc_throw(size_t size); void* malloc_throw(size_t size);
void* calloc_throw(size_t nmemb, size_t size); void* calloc_throw(size_t nmemb, size_t size);
......
...@@ -68,7 +68,7 @@ void tscPrintMgmtIp() { ...@@ -68,7 +68,7 @@ void tscPrintMgmtIp() {
} }
} }
void tscSetMgmtIpListFromCluster(SRpcIpSet *pIpList) { void tscSetMgmtIpList(SRpcIpSet *pIpList) {
tscMgmtIpSet.numOfIps = pIpList->numOfIps; tscMgmtIpSet.numOfIps = pIpList->numOfIps;
tscMgmtIpSet.inUse = pIpList->inUse; tscMgmtIpSet.inUse = pIpList->inUse;
for (int32_t i = 0; i < tscMgmtIpSet.numOfIps; ++i) { for (int32_t i = 0; i < tscMgmtIpSet.numOfIps; ++i) {
...@@ -76,16 +76,6 @@ void tscSetMgmtIpListFromCluster(SRpcIpSet *pIpList) { ...@@ -76,16 +76,6 @@ void tscSetMgmtIpListFromCluster(SRpcIpSet *pIpList) {
} }
} }
void tscSetMgmtIpListFromEdge() {
if (tscMgmtIpSet.numOfIps != 1) {
tscMgmtIpSet.numOfIps = 1;
tscMgmtIpSet.inUse = 0;
taosGetFqdnPortFromEp(tsFirst, tscMgmtIpSet.fqdn[0], &tscMgmtIpSet.port[0]);
tscTrace("edge mgmt IP list:");
tscPrintMgmtIp();
}
}
void tscUpdateIpSet(void *ahandle, SRpcIpSet *pIpSet) { void tscUpdateIpSet(void *ahandle, SRpcIpSet *pIpSet) {
tscMgmtIpSet = *pIpSet; tscMgmtIpSet = *pIpSet;
tscTrace("mgmt IP list is changed for ufp is called, numOfIps:%d inUse:%d", tscMgmtIpSet.numOfIps, tscMgmtIpSet.inUse); tscTrace("mgmt IP list is changed for ufp is called, numOfIps:%d inUse:%d", tscMgmtIpSet.numOfIps, tscMgmtIpSet.inUse);
...@@ -94,18 +84,6 @@ void tscUpdateIpSet(void *ahandle, SRpcIpSet *pIpSet) { ...@@ -94,18 +84,6 @@ void tscUpdateIpSet(void *ahandle, SRpcIpSet *pIpSet) {
} }
} }
void tscSetMgmtIpList(SRpcIpSet *pIpList) {
/*
* The iplist returned by the cluster edition is the current management nodes
* and the iplist returned by the edge edition is empty
*/
if (pIpList->numOfIps != 0) {
tscSetMgmtIpListFromCluster(pIpList);
} else {
tscSetMgmtIpListFromEdge();
}
}
/* /*
* For each management node, try twice at least in case of poor network situation. * For each management node, try twice at least in case of poor network situation.
* If the client start to connect to a non-management node from the client, and the first retry may fail due to * If the client start to connect to a non-management node from the client, and the first retry may fail due to
...@@ -133,7 +111,8 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) { ...@@ -133,7 +111,8 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) {
if (code == 0) { if (code == 0) {
SCMHeartBeatRsp *pRsp = (SCMHeartBeatRsp *)pRes->pRsp; SCMHeartBeatRsp *pRsp = (SCMHeartBeatRsp *)pRes->pRsp;
SRpcIpSet * pIpList = &pRsp->ipList; SRpcIpSet * pIpList = &pRsp->ipList;
tscSetMgmtIpList(pIpList); if (pIpList->numOfIps > 0)
tscSetMgmtIpList(pIpList);
if (pRsp->killConnection) { if (pRsp->killConnection) {
tscKillConnection(pObj); tscKillConnection(pObj);
...@@ -2225,7 +2204,8 @@ int tscProcessConnectRsp(SSqlObj *pSql) { ...@@ -2225,7 +2204,8 @@ int tscProcessConnectRsp(SSqlObj *pSql) {
assert(len <= tListLen(pObj->db)); assert(len <= tListLen(pObj->db));
strncpy(pObj->db, temp, tListLen(pObj->db)); strncpy(pObj->db, temp, tListLen(pObj->db));
tscSetMgmtIpList(&pConnect->ipList); if (pConnect->ipList.numOfIps > 0)
tscSetMgmtIpList(&pConnect->ipList);
strcpy(pObj->sversion, pConnect->serverVersion); strcpy(pObj->sversion, pConnect->serverVersion);
pObj->writeAuth = pConnect->writeAuth; pObj->writeAuth = pConnect->writeAuth;
......
...@@ -65,32 +65,18 @@ STscObj *taosConnectImpl(const char *ip, const char *user, const char *pass, con ...@@ -65,32 +65,18 @@ STscObj *taosConnectImpl(const char *ip, const char *user, const char *pass, con
terrno = TSDB_CODE_INVALID_PASS; terrno = TSDB_CODE_INVALID_PASS;
return NULL; return NULL;
} }
if (ip) {
if (tscSetMgmtIpListFromCfg(ip, NULL) < 0) return NULL;
if (port) tscMgmtIpSet.port[0] = port;
}
void *pDnodeConn = NULL; void *pDnodeConn = NULL;
if (tscInitRpc(user, pass, &pDnodeConn) != 0) { if (tscInitRpc(user, pass, &pDnodeConn) != 0) {
terrno = TSDB_CODE_NETWORK_UNAVAIL; terrno = TSDB_CODE_NETWORK_UNAVAIL;
return NULL; return NULL;
} }
tscMgmtIpSet.numOfIps = 0;
if (ip && ip[0]) {
tscMgmtIpSet.inUse = 0;
tscMgmtIpSet.numOfIps = 1;
strcpy(tscMgmtIpSet.fqdn[0], ip);
tscMgmtIpSet.port[0] = port? port: tsDnodeShellPort;
} else {
if (tsFirst[0] != 0) {
taosGetFqdnPortFromEp(tsFirst, tscMgmtIpSet.fqdn[tscMgmtIpSet.numOfIps], &tscMgmtIpSet.port[tscMgmtIpSet.numOfIps]);
tscMgmtIpSet.numOfIps++;
}
if (tsSecond[0] != 0) {
taosGetFqdnPortFromEp(tsSecond, tscMgmtIpSet.fqdn[tscMgmtIpSet.numOfIps], &tscMgmtIpSet.port[tscMgmtIpSet.numOfIps]);
tscMgmtIpSet.numOfIps++;
}
}
STscObj *pObj = (STscObj *)calloc(1, sizeof(STscObj)); STscObj *pObj = (STscObj *)calloc(1, sizeof(STscObj));
if (NULL == pObj) { if (NULL == pObj) {
terrno = TSDB_CODE_CLI_OUT_OF_MEMORY; terrno = TSDB_CODE_CLI_OUT_OF_MEMORY;
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "tutil.h" #include "tutil.h"
#include "tsched.h" #include "tsched.h"
#include "tscLog.h" #include "tscLog.h"
#include "tscUtil.h"
#include "tsclient.h" #include "tsclient.h"
#include "tglobal.h" #include "tglobal.h"
#include "tconfig.h" #include "tconfig.h"
...@@ -114,14 +115,10 @@ void taos_init_imp() { ...@@ -114,14 +115,10 @@ void taos_init_imp() {
taosInitNote(tsNumOfLogLines / 10, 1, (char*)"tsc_note"); taosInitNote(tsNumOfLogLines / 10, 1, (char*)"tsc_note");
} }
tscMgmtIpSet.inUse = 0; if (tscSetMgmtIpListFromCfg(tsFirst, tsSecond) < 0) {
tscMgmtIpSet.numOfIps = 1; tscError("failed to init mgmt IP list");
taosGetFqdnPortFromEp(tsFirst, tscMgmtIpSet.fqdn[0], &tscMgmtIpSet.port[0]); return;
}
if (tsSecond[0] && strcmp(tsSecond, tsFirst) != 0) {
tscMgmtIpSet.numOfIps = 2;
taosGetFqdnPortFromEp(tsSecond, tscMgmtIpSet.fqdn[1], &tscMgmtIpSet.port[1]);
}
tscInitMsgsFp(); tscInitMsgsFp();
int queueSize = tsMaxVnodeConnections + tsMaxMeterConnections + tsMaxMgmtConnections + tsMaxMgmtConnections; int queueSize = tsMaxVnodeConnections + tsMaxMeterConnections + tsMaxMgmtConnections + tsMaxMgmtConnections;
......
...@@ -2163,3 +2163,33 @@ char* strdup_throw(const char* str) { ...@@ -2163,3 +2163,33 @@ char* strdup_throw(const char* str) {
} }
return p; return p;
} }
int tscSetMgmtIpListFromCfg(const char *first, const char *second) {
tscMgmtIpSet.numOfIps = 0;
tscMgmtIpSet.inUse = 0;
if (first && first[0] != 0) {
if (strlen(first) >= TSDB_FQDN_LEN) {
terrno = TSDB_CODE_INVALID_FQDN;
return -1;
}
taosGetFqdnPortFromEp(first, tscMgmtIpSet.fqdn[tscMgmtIpSet.numOfIps], &tscMgmtIpSet.port[tscMgmtIpSet.numOfIps]);
tscMgmtIpSet.numOfIps++;
}
if (second && second[0] != 0) {
if (strlen(second) >= TSDB_FQDN_LEN) {
terrno = TSDB_CODE_INVALID_FQDN;
return -1;
}
taosGetFqdnPortFromEp(second, tscMgmtIpSet.fqdn[tscMgmtIpSet.numOfIps], &tscMgmtIpSet.port[tscMgmtIpSet.numOfIps]);
tscMgmtIpSet.numOfIps++;
}
if ( tscMgmtIpSet.numOfIps == 0) {
terrno = TSDB_CODE_INVALID_FQDN;
return -1;
}
return 0;
}
...@@ -175,7 +175,7 @@ void taosInitGlobalCfg(); ...@@ -175,7 +175,7 @@ void taosInitGlobalCfg();
bool taosCheckGlobalCfg(); bool taosCheckGlobalCfg();
void taosSetAllDebugFlag(); void taosSetAllDebugFlag();
bool taosCfgDynamicOptions(char *msg); bool taosCfgDynamicOptions(char *msg);
int taosGetFqdnPortFromEp(char *ep, char *fqdn, uint16_t *port); int taosGetFqdnPortFromEp(const char *ep, char *fqdn, uint16_t *port);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
...@@ -61,7 +61,7 @@ int32_t tscEmbedded = 0; ...@@ -61,7 +61,7 @@ int32_t tscEmbedded = 0;
*/ */
int64_t tsMsPerDay[] = {86400000L, 86400000000L}; int64_t tsMsPerDay[] = {86400000L, 86400000000L};
char tsFirst[TSDB_FQDN_LEN] = {0}; char tsFirst[TSDB_FQDN_LEN] = {0};
char tsSecond[TSDB_FQDN_LEN] = {0}; char tsSecond[TSDB_FQDN_LEN] = {0};
char tsArbitrator[TSDB_FQDN_LEN] = {0}; char tsArbitrator[TSDB_FQDN_LEN] = {0};
char tsLocalEp[TSDB_FQDN_LEN] = {0}; // Local End Point, hostname:port char tsLocalEp[TSDB_FQDN_LEN] = {0}; // Local End Point, hostname:port
...@@ -1252,7 +1252,7 @@ bool taosCheckGlobalCfg() { ...@@ -1252,7 +1252,7 @@ bool taosCheckGlobalCfg() {
return true; return true;
} }
int taosGetFqdnPortFromEp(char *ep, char *fqdn, uint16_t *port) { int taosGetFqdnPortFromEp(const char *ep, char *fqdn, uint16_t *port) {
*port = 0; *port = 0;
strcpy(fqdn, ep); strcpy(fqdn, ep);
......
...@@ -218,7 +218,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size); ...@@ -218,7 +218,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
#define TSDB_LOCALE_LEN 64 #define TSDB_LOCALE_LEN 64
#define TSDB_TIMEZONE_LEN 64 #define TSDB_TIMEZONE_LEN 64
#define TSDB_FQDN_LEN 72 #define TSDB_FQDN_LEN 256
#define TSDB_IPv4ADDR_LEN 16 #define TSDB_IPv4ADDR_LEN 16
#define TSDB_FILENAME_LEN 128 #define TSDB_FILENAME_LEN 128
#define TSDB_METER_VNODE_BITS 20 #define TSDB_METER_VNODE_BITS 20
...@@ -234,7 +234,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size); ...@@ -234,7 +234,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
#define TSDB_DEFAULT_PKT_SIZE 65480 //same as RPC_MAX_UDP_SIZE #define TSDB_DEFAULT_PKT_SIZE 65480 //same as RPC_MAX_UDP_SIZE
#define TSDB_PAYLOAD_SIZE (TSDB_DEFAULT_PKT_SIZE - 100) #define TSDB_PAYLOAD_SIZE (TSDB_DEFAULT_PKT_SIZE - 100)
#define TSDB_DEFAULT_PAYLOAD_SIZE 1024 // default payload size #define TSDB_DEFAULT_PAYLOAD_SIZE 2048 // default payload size
#define TSDB_EXTRA_PAYLOAD_SIZE 128 // extra bytes for auth #define TSDB_EXTRA_PAYLOAD_SIZE 128 // extra bytes for auth
#define TSDB_CQ_SQL_SIZE 1024 #define TSDB_CQ_SQL_SIZE 1024
#define TSDB_MAX_VNODES 256 #define TSDB_MAX_VNODES 256
......
...@@ -170,6 +170,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_QHANDLE, 0, 459, "invalid handle" ...@@ -170,6 +170,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_QHANDLE, 0, 459, "invalid handle"
TAOS_DEFINE_ERROR(TSDB_CODE_QUERY_CANCELLED, 0, 460, "query cancelled") TAOS_DEFINE_ERROR(TSDB_CODE_QUERY_CANCELLED, 0, 460, "query cancelled")
TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_IE, 0, 461, "invalid ie") TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_IE, 0, 461, "invalid ie")
TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_VALUE, 0, 462, "invalid value") TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_VALUE, 0, 462, "invalid value")
TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_FQDN, 0, 463, "invalid FQDN")
// others // others
TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_FILE_FORMAT, 0, 500, "invalid file format") TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_FILE_FORMAT, 0, 500, "invalid file format")
......
...@@ -187,13 +187,13 @@ typedef struct SMsgHead { ...@@ -187,13 +187,13 @@ typedef struct SMsgHead {
// Submit message for one table // Submit message for one table
typedef struct SSubmitBlk { typedef struct SSubmitBlk {
int64_t uid; // table unique id uint64_t uid; // table unique id
int32_t tid; // table id int32_t tid; // table id
int32_t padding; // TODO just for padding here int32_t padding; // TODO just for padding here
int32_t sversion; // data schema version int32_t sversion; // data schema version
int32_t len; // data part length, not including the SSubmitBlk head int32_t len; // data part length, not including the SSubmitBlk head
int16_t numOfRows; // total number of rows in current submit block int16_t numOfRows; // total number of rows in current submit block
char data[]; char data[];
} SSubmitBlk; } SSubmitBlk;
// Submit message for this TSDB // Submit message for this TSDB
...@@ -327,9 +327,9 @@ typedef struct { ...@@ -327,9 +327,9 @@ typedef struct {
} SMDDropTableMsg; } SMDDropTableMsg;
typedef struct { typedef struct {
int32_t contLen; int32_t contLen;
int32_t vgId; int32_t vgId;
int64_t uid; uint64_t uid;
char tableId[TSDB_TABLE_ID_LEN + 1]; char tableId[TSDB_TABLE_ID_LEN + 1];
} SMDDropSTableMsg; } SMDDropSTableMsg;
...@@ -404,9 +404,9 @@ typedef struct SColumnInfo { ...@@ -404,9 +404,9 @@ typedef struct SColumnInfo {
} SColumnInfo; } SColumnInfo;
typedef struct STableIdInfo { typedef struct STableIdInfo {
int64_t uid; uint64_t uid;
int32_t tid; int32_t tid;
TSKEY key; // last accessed ts, for subscription TSKEY key; // last accessed ts, for subscription
} STableIdInfo; } STableIdInfo;
typedef struct STimeWindow { typedef struct STimeWindow {
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册