Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
冷面1992
TDengine
提交
fdfa789c
TDengine
项目概览
冷面1992
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
fdfa789c
编写于
7月 18, 2020
作者:
S
Shengliang Guan
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
replace ipList and ipSet to epSet
上级
c48b5d7e
变更
33
隐藏空白更改
内联
并排
Showing
33 changed file
with
345 addition
and
345 deletion
+345
-345
src/client/inc/tscUtil.h
src/client/inc/tscUtil.h
+1
-1
src/client/inc/tsclient.h
src/client/inc/tsclient.h
+3
-3
src/client/src/tscSQLParser.c
src/client/src/tscSQLParser.c
+2
-2
src/client/src/tscServer.c
src/client/src/tscServer.c
+51
-51
src/client/src/tscSql.c
src/client/src/tscSql.c
+2
-2
src/client/src/tscSubquery.c
src/client/src/tscSubquery.c
+7
-7
src/client/src/tscSystem.c
src/client/src/tscSystem.c
+3
-3
src/client/src/tscUtil.c
src/client/src/tscUtil.c
+8
-8
src/dnode/inc/dnodeMgmt.h
src/dnode/inc/dnodeMgmt.h
+2
-2
src/dnode/src/dnodeMgmt.c
src/dnode/src/dnodeMgmt.c
+52
-52
src/dnode/src/dnodePeer.c
src/dnode/src/dnodePeer.c
+12
-12
src/dnode/src/dnodeShell.c
src/dnode/src/dnodeShell.c
+2
-2
src/inc/dnode.h
src/inc/dnode.h
+3
-3
src/inc/taosmsg.h
src/inc/taosmsg.h
+5
-5
src/inc/trpc.h
src/inc/trpc.h
+7
-7
src/kit/taosmigrate/taosmigrate.c
src/kit/taosmigrate/taosmigrate.c
+4
-4
src/kit/taosmigrate/taosmigrate.h
src/kit/taosmigrate/taosmigrate.h
+1
-1
src/kit/taosmigrate/taosmigrateDnodeCfg.c
src/kit/taosmigrate/taosmigrateDnodeCfg.c
+18
-18
src/mnode/inc/mnodeMnode.h
src/mnode/inc/mnodeMnode.h
+3
-3
src/mnode/inc/mnodeVgroup.h
src/mnode/inc/mnodeVgroup.h
+3
-3
src/mnode/src/mnodeDnode.c
src/mnode/src/mnodeDnode.c
+5
-5
src/mnode/src/mnodeMnode.c
src/mnode/src/mnodeMnode.c
+24
-24
src/mnode/src/mnodePeer.c
src/mnode/src/mnodePeer.c
+7
-7
src/mnode/src/mnodeRead.c
src/mnode/src/mnodeRead.c
+7
-7
src/mnode/src/mnodeSdb.c
src/mnode/src/mnodeSdb.c
+1
-1
src/mnode/src/mnodeShow.c
src/mnode/src/mnodeShow.c
+2
-2
src/mnode/src/mnodeTable.c
src/mnode/src/mnodeTable.c
+14
-14
src/mnode/src/mnodeVgroup.c
src/mnode/src/mnodeVgroup.c
+29
-29
src/mnode/src/mnodeWrite.c
src/mnode/src/mnodeWrite.c
+8
-8
src/rpc/src/rpcMain.c
src/rpc/src/rpcMain.c
+30
-30
src/rpc/test/rclient.c
src/rpc/test/rclient.c
+15
-15
src/rpc/test/rsclient.c
src/rpc/test/rsclient.c
+13
-13
src/rpc/test/rserver.c
src/rpc/test/rserver.c
+1
-1
未找到文件。
src/client/inc/tscUtil.h
浏览文件 @
fdfa789c
...
...
@@ -268,7 +268,7 @@ bool hasMoreClauseToTry(SSqlObj* pSql);
void
tscTryQueryNextVnode
(
SSqlObj
*
pSql
,
__async_cb_func_t
fp
);
void
tscAsyncQuerySingleRowForNextVnode
(
void
*
param
,
TAOS_RES
*
tres
,
int
numOfRows
);
void
tscTryQueryNextClause
(
SSqlObj
*
pSql
,
__async_cb_func_t
fp
);
int
tscSetMgmt
IpLis
tFromCfg
(
const
char
*
first
,
const
char
*
second
);
int
tscSetMgmt
EpSe
tFromCfg
(
const
char
*
first
,
const
char
*
second
);
void
*
malloc_throw
(
size_t
size
);
void
*
calloc_throw
(
size_t
nmemb
,
size_t
size
);
...
...
src/client/inc/tsclient.h
浏览文件 @
fdfa789c
...
...
@@ -306,7 +306,7 @@ typedef struct SSqlObj {
char
*
sqlstr
;
char
retry
;
char
maxRetry
;
SRpc
IpSet
ipLis
t
;
SRpc
EpSet
epSe
t
;
char
listed
;
tsem_t
rspSem
;
SSqlCmd
cmd
;
...
...
@@ -350,7 +350,7 @@ void tscInitMsgsFp();
int
tsParseSql
(
SSqlObj
*
pSql
,
bool
initial
);
void
tscProcessMsgFromServer
(
SRpcMsg
*
rpcMsg
,
SRpc
IpSet
*
pI
pSet
);
void
tscProcessMsgFromServer
(
SRpcMsg
*
rpcMsg
,
SRpc
EpSet
*
pE
pSet
);
int
tscProcessSql
(
SSqlObj
*
pSql
);
int
tscRenewTableMeta
(
SSqlObj
*
pSql
,
char
*
tableId
);
...
...
@@ -456,7 +456,7 @@ extern void * tscQhandle;
extern
int
tscKeepConn
[];
extern
int
tsInsertHeadSize
;
extern
int
tscNumOfThreads
;
extern
SRpc
IpSet
tscMgmtI
pSet
;
extern
SRpc
EpSet
tscMgmtE
pSet
;
extern
int
(
*
tscBuildMsg
[
TSDB_SQL_MAX
])(
SSqlObj
*
pSql
,
SSqlInfo
*
pInfo
);
...
...
src/client/src/tscSQLParser.c
浏览文件 @
fdfa789c
...
...
@@ -2355,9 +2355,9 @@ bool validateIpAddress(const char* ip, size_t size) {
strncpy
(
tmp
,
ip
,
size
);
in_addr_t
i
pAddr
=
inet_addr
(
tmp
);
in_addr_t
e
pAddr
=
inet_addr
(
tmp
);
return
i
pAddr
!=
INADDR_NONE
;
return
e
pAddr
!=
INADDR_NONE
;
}
int32_t
tscTansformSQLFuncForSTableQuery
(
SQueryInfo
*
pQueryInfo
)
{
...
...
src/client/src/tscServer.c
浏览文件 @
fdfa789c
...
...
@@ -29,8 +29,8 @@
#define TSC_MGMT_VNODE 999
SRpc
IpSet
tscMgmtI
pSet
;
SRpc
IpSet
tscDnodeI
pSet
;
SRpc
EpSet
tscMgmtE
pSet
;
SRpc
EpSet
tscDnodeE
pSet
;
int
(
*
tscBuildMsg
[
TSDB_SQL_MAX
])(
SSqlObj
*
pSql
,
SSqlInfo
*
pInfo
)
=
{
0
};
...
...
@@ -44,44 +44,44 @@ void tscSaveSubscriptionProgress(void* sub);
static
int32_t
minMsgSize
()
{
return
tsRpcHeadSize
+
100
;
}
static
void
tscSetDnode
IpLis
t
(
SSqlObj
*
pSql
,
SCMVgroupInfo
*
pVgroupInfo
)
{
SRpc
IpSet
*
pIpList
=
&
pSql
->
ipLis
t
;
p
IpLis
t
->
inUse
=
0
;
static
void
tscSetDnode
EpSe
t
(
SSqlObj
*
pSql
,
SCMVgroupInfo
*
pVgroupInfo
)
{
SRpc
EpSet
*
pEpSet
=
&
pSql
->
epSe
t
;
p
EpSe
t
->
inUse
=
0
;
if
(
pVgroupInfo
==
NULL
)
{
p
IpList
->
numOfI
ps
=
0
;
p
EpSet
->
numOfE
ps
=
0
;
return
;
}
p
IpList
->
numOfIps
=
pVgroupInfo
->
numOfI
ps
;
for
(
int32_t
i
=
0
;
i
<
pVgroupInfo
->
numOf
I
ps
;
++
i
)
{
strcpy
(
p
IpList
->
fqdn
[
i
],
pVgroupInfo
->
i
pAddr
[
i
].
fqdn
);
p
IpList
->
port
[
i
]
=
pVgroupInfo
->
i
pAddr
[
i
].
port
;
p
EpSet
->
numOfEps
=
pVgroupInfo
->
numOfE
ps
;
for
(
int32_t
i
=
0
;
i
<
pVgroupInfo
->
numOf
E
ps
;
++
i
)
{
strcpy
(
p
EpSet
->
fqdn
[
i
],
pVgroupInfo
->
e
pAddr
[
i
].
fqdn
);
p
EpSet
->
port
[
i
]
=
pVgroupInfo
->
e
pAddr
[
i
].
port
;
}
}
void
tscPrintMgmt
I
p
()
{
if
(
tscMgmt
IpSet
.
numOfI
ps
<=
0
)
{
tscError
(
"invalid mnode
IP list:%d"
,
tscMgmtIpSet
.
numOfI
ps
);
void
tscPrintMgmt
E
p
()
{
if
(
tscMgmt
EpSet
.
numOfE
ps
<=
0
)
{
tscError
(
"invalid mnode
EP list:%d"
,
tscMgmtEpSet
.
numOfE
ps
);
}
else
{
for
(
int
i
=
0
;
i
<
tscMgmt
IpSet
.
numOfI
ps
;
++
i
)
{
tscDebug
(
"mnode index:%d %s:%d"
,
i
,
tscMgmt
IpSet
.
fqdn
[
i
],
tscMgmtI
pSet
.
port
[
i
]);
for
(
int
i
=
0
;
i
<
tscMgmt
EpSet
.
numOfE
ps
;
++
i
)
{
tscDebug
(
"mnode index:%d %s:%d"
,
i
,
tscMgmt
EpSet
.
fqdn
[
i
],
tscMgmtE
pSet
.
port
[
i
]);
}
}
}
void
tscSetMgmt
IpList
(
SRpcIpSet
*
pIpLis
t
)
{
tscMgmt
IpSet
.
numOfIps
=
pIpList
->
numOfI
ps
;
tscMgmt
IpSet
.
inUse
=
pIpLis
t
->
inUse
;
for
(
int32_t
i
=
0
;
i
<
tscMgmt
IpSet
.
numOfI
ps
;
++
i
)
{
tscMgmt
IpSet
.
port
[
i
]
=
htons
(
pIpLis
t
->
port
[
i
]);
void
tscSetMgmt
EpSet
(
SRpcEpSet
*
pEpSe
t
)
{
tscMgmt
EpSet
.
numOfEps
=
pEpSet
->
numOfE
ps
;
tscMgmt
EpSet
.
inUse
=
pEpSe
t
->
inUse
;
for
(
int32_t
i
=
0
;
i
<
tscMgmt
EpSet
.
numOfE
ps
;
++
i
)
{
tscMgmt
EpSet
.
port
[
i
]
=
htons
(
pEpSe
t
->
port
[
i
]);
}
}
void
tscUpdate
IpSet
(
void
*
ahandle
,
SRpcIpSet
*
pI
pSet
)
{
tscMgmt
IpSet
=
*
pI
pSet
;
tscDebug
(
"mnode
IP list is changed for ufp is called, numOfIps:%d inUse:%d"
,
tscMgmtIpSet
.
numOfIps
,
tscMgmtI
pSet
.
inUse
);
for
(
int32_t
i
=
0
;
i
<
tscMgmt
IpSet
.
numOfI
ps
;
++
i
)
{
tscDebug
(
"index:%d fqdn:%s port:%d"
,
i
,
tscMgmt
IpSet
.
fqdn
[
i
],
tscMgmtI
pSet
.
port
[
i
]);
void
tscUpdate
EpSet
(
void
*
ahandle
,
SRpcEpSet
*
pE
pSet
)
{
tscMgmt
EpSet
=
*
pE
pSet
;
tscDebug
(
"mnode
EP list is changed for ufp is called, numOfEps:%d inUse:%d"
,
tscMgmtEpSet
.
numOfEps
,
tscMgmtE
pSet
.
inUse
);
for
(
int32_t
i
=
0
;
i
<
tscMgmt
EpSet
.
numOfE
ps
;
++
i
)
{
tscDebug
(
"index:%d fqdn:%s port:%d"
,
i
,
tscMgmt
EpSet
.
fqdn
[
i
],
tscMgmtE
pSet
.
port
[
i
]);
}
}
...
...
@@ -95,7 +95,7 @@ void tscUpdateIpSet(void *ahandle, SRpcIpSet *pIpSet) {
UNUSED_FUNC
static
int32_t
tscGetMgmtConnMaxRetryTimes
()
{
int32_t
factor
=
2
;
return
tscMgmt
IpSet
.
numOfI
ps
*
factor
;
return
tscMgmt
EpSet
.
numOfE
ps
*
factor
;
}
void
tscProcessHeartBeatRsp
(
void
*
param
,
TAOS_RES
*
tres
,
int
code
)
{
...
...
@@ -111,9 +111,9 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) {
if
(
code
==
0
)
{
SCMHeartBeatRsp
*
pRsp
=
(
SCMHeartBeatRsp
*
)
pRes
->
pRsp
;
SRpc
IpSet
*
pIpList
=
&
pRsp
->
ipLis
t
;
if
(
p
IpList
->
numOfI
ps
>
0
)
tscSetMgmt
IpList
(
pIpLis
t
);
SRpc
EpSet
*
pEpSet
=
&
pRsp
->
epSe
t
;
if
(
p
EpSet
->
numOfE
ps
>
0
)
tscSetMgmt
EpSet
(
pEpSe
t
);
pSql
->
pTscObj
->
connId
=
htonl
(
pRsp
->
connId
);
...
...
@@ -185,7 +185,7 @@ int tscSendMsgToServer(SSqlObj *pSql) {
// set the mgmt ip list
if
(
pSql
->
cmd
.
command
>=
TSDB_SQL_MGMT
)
{
pSql
->
ipList
=
tscMgmtI
pSet
;
pSql
->
epSet
=
tscMgmtE
pSet
;
}
memcpy
(
pMsg
,
pSql
->
cmd
.
payload
,
pSql
->
cmd
.
payloadLen
);
...
...
@@ -203,11 +203,11 @@ int tscSendMsgToServer(SSqlObj *pSql) {
// Otherwise, the pSql object may have been released already during the response function, which is
// processMsgFromServer function. In the meanwhile, the assignment of the rpc context to sql object will absolutely
// cause crash.
rpcSendRequest
(
pObj
->
pDnodeConn
,
&
pSql
->
ipLis
t
,
&
rpcMsg
);
rpcSendRequest
(
pObj
->
pDnodeConn
,
&
pSql
->
epSe
t
,
&
rpcMsg
);
return
TSDB_CODE_SUCCESS
;
}
void
tscProcessMsgFromServer
(
SRpcMsg
*
rpcMsg
,
SRpc
IpSet
*
pI
pSet
)
{
void
tscProcessMsgFromServer
(
SRpcMsg
*
rpcMsg
,
SRpc
EpSet
*
pE
pSet
)
{
SSqlObj
*
pSql
=
(
SSqlObj
*
)
rpcMsg
->
ahandle
;
if
(
pSql
==
NULL
||
pSql
->
signature
!=
pSql
)
{
tscError
(
"%p sql is already released"
,
pSql
);
...
...
@@ -237,9 +237,9 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet) {
}
if
(
pCmd
->
command
<
TSDB_SQL_MGMT
)
{
if
(
p
IpSet
)
pSql
->
ipList
=
*
pI
pSet
;
if
(
p
EpSet
)
pSql
->
epSet
=
*
pE
pSet
;
}
else
{
if
(
p
IpSet
)
tscMgmtIpSet
=
*
pI
pSet
;
if
(
p
EpSet
)
tscMgmtEpSet
=
*
pE
pSet
;
}
if
(
rpcMsg
->
pCont
==
NULL
)
{
...
...
@@ -421,7 +421,7 @@ int tscProcessSql(SSqlObj *pSql) {
return
pSql
->
res
.
code
;
}
}
else
if
(
pCmd
->
command
<
TSDB_SQL_LOCAL
)
{
pSql
->
ipList
=
tscMgmtI
pSet
;
pSql
->
epSet
=
tscMgmtE
pSet
;
}
else
{
// local handler
return
(
*
tscProcessMsgRsp
[
pCmd
->
command
])(
pSql
);
}
...
...
@@ -525,10 +525,10 @@ int tscBuildSubmitMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
// pSql->cmd.payloadLen is set during copying data into payload
pSql
->
cmd
.
msgType
=
TSDB_MSG_TYPE_SUBMIT
;
tscSetDnode
IpLis
t
(
pSql
,
&
pTableMeta
->
vgroupInfo
);
tscSetDnode
EpSe
t
(
pSql
,
&
pTableMeta
->
vgroupInfo
);
tscDebug
(
"%p build submit msg, vgId:%d numOfTables:%d numberOf
I
P:%d"
,
pSql
,
vgId
,
pSql
->
cmd
.
numOfTablesInSubmit
,
pSql
->
ipList
.
numOfI
ps
);
tscDebug
(
"%p build submit msg, vgId:%d numOfTables:%d numberOf
E
P:%d"
,
pSql
,
vgId
,
pSql
->
cmd
.
numOfTablesInSubmit
,
pSql
->
epSet
.
numOfE
ps
);
return
TSDB_CODE_SUCCESS
;
}
...
...
@@ -568,7 +568,7 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
pVgroupInfo
=
&
pTableMeta
->
vgroupInfo
;
}
tscSetDnode
IpLis
t
(
pSql
,
pVgroupInfo
);
tscSetDnode
EpSe
t
(
pSql
,
pVgroupInfo
);
if
(
pVgroupInfo
!=
NULL
)
{
pQueryMsg
->
head
.
vgId
=
htonl
(
pVgroupInfo
->
vgId
);
}
...
...
@@ -580,7 +580,7 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
pQueryMsg
->
numOfTables
=
htonl
(
1
);
// set the number of tables
pMsg
+=
sizeof
(
STableIdInfo
);
}
else
{
// it is a subquery of the super table query, this
I
P info is acquired from vgroupInfo
}
else
{
// it is a subquery of the super table query, this
E
P info is acquired from vgroupInfo
int32_t
index
=
pTableMetaInfo
->
vgroupIndex
;
int32_t
numOfVgroups
=
taosArrayGetSize
(
pTableMetaInfo
->
pVgroupTables
);
assert
(
index
>=
0
&&
index
<
numOfVgroups
);
...
...
@@ -590,7 +590,7 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
SVgroupTableInfo
*
pTableIdList
=
taosArrayGet
(
pTableMetaInfo
->
pVgroupTables
,
index
);
// set the vgroup info
tscSetDnode
IpLis
t
(
pSql
,
&
pTableIdList
->
vgInfo
);
tscSetDnode
EpSe
t
(
pSql
,
&
pTableIdList
->
vgInfo
);
pQueryMsg
->
head
.
vgId
=
htonl
(
pTableIdList
->
vgInfo
.
vgId
);
int32_t
numOfTables
=
taosArrayGetSize
(
pTableIdList
->
itemList
);
...
...
@@ -1323,7 +1323,7 @@ int tscBuildUpdateTagMsg(SSqlObj* pSql, SSqlInfo *pInfo) {
SQueryInfo
*
pQueryInfo
=
tscGetQueryInfoDetail
(
pCmd
,
0
);
STableMetaInfo
*
pTableMetaInfo
=
tscGetMetaInfo
(
pQueryInfo
,
0
);
tscSetDnode
IpLis
t
(
pSql
,
&
pTableMetaInfo
->
pTableMeta
->
vgroupInfo
);
tscSetDnode
EpSe
t
(
pSql
,
&
pTableMetaInfo
->
pTableMeta
->
vgroupInfo
);
return
TSDB_CODE_SUCCESS
;
}
...
...
@@ -1658,8 +1658,8 @@ int tscProcessTableMetaRsp(SSqlObj *pSql) {
pMetaMsg
->
contLen
=
htons
(
pMetaMsg
->
contLen
);
pMetaMsg
->
numOfColumns
=
htons
(
pMetaMsg
->
numOfColumns
);
if
(
pMetaMsg
->
sid
<
0
||
pMetaMsg
->
vgroup
.
numOf
I
ps
<
0
)
{
tscError
(
"invalid meter vgId:%d, sid%d"
,
pMetaMsg
->
vgroup
.
numOf
I
ps
,
pMetaMsg
->
sid
);
if
(
pMetaMsg
->
sid
<
0
||
pMetaMsg
->
vgroup
.
numOf
E
ps
<
0
)
{
tscError
(
"invalid meter vgId:%d, sid%d"
,
pMetaMsg
->
vgroup
.
numOf
E
ps
,
pMetaMsg
->
sid
);
return
TSDB_CODE_TSC_INVALID_VALUE
;
}
...
...
@@ -1673,8 +1673,8 @@ int tscProcessTableMetaRsp(SSqlObj *pSql) {
return
TSDB_CODE_TSC_INVALID_VALUE
;
}
for
(
int
i
=
0
;
i
<
pMetaMsg
->
vgroup
.
numOf
I
ps
;
++
i
)
{
pMetaMsg
->
vgroup
.
ipAddr
[
i
].
port
=
htons
(
pMetaMsg
->
vgroup
.
i
pAddr
[
i
].
port
);
for
(
int
i
=
0
;
i
<
pMetaMsg
->
vgroup
.
numOf
E
ps
;
++
i
)
{
pMetaMsg
->
vgroup
.
epAddr
[
i
].
port
=
htons
(
pMetaMsg
->
vgroup
.
e
pAddr
[
i
].
port
);
}
SSchema
*
pSchema
=
pMetaMsg
->
schema
;
...
...
@@ -1850,10 +1850,10 @@ int tscProcessSTableVgroupRsp(SSqlObj *pSql) {
SCMVgroupInfo
*
pVgroups
=
&
pInfo
->
vgroupList
->
vgroups
[
j
];
pVgroups
->
vgId
=
htonl
(
pVgroups
->
vgId
);
assert
(
pVgroups
->
numOf
I
ps
>=
1
);
assert
(
pVgroups
->
numOf
E
ps
>=
1
);
for
(
int32_t
k
=
0
;
k
<
pVgroups
->
numOf
I
ps
;
++
k
)
{
pVgroups
->
ipAddr
[
k
].
port
=
htons
(
pVgroups
->
i
pAddr
[
k
].
port
);
for
(
int32_t
k
=
0
;
k
<
pVgroups
->
numOf
E
ps
;
++
k
)
{
pVgroups
->
epAddr
[
k
].
port
=
htons
(
pVgroups
->
e
pAddr
[
k
].
port
);
}
pMsg
+=
size
;
...
...
@@ -1946,8 +1946,8 @@ int tscProcessConnectRsp(SSqlObj *pSql) {
assert
(
len
<=
sizeof
(
pObj
->
db
));
tstrncpy
(
pObj
->
db
,
temp
,
sizeof
(
pObj
->
db
));
if
(
pConnect
->
ipList
.
numOfI
ps
>
0
)
tscSetMgmt
IpList
(
&
pConnect
->
ipLis
t
);
if
(
pConnect
->
epSet
.
numOfE
ps
>
0
)
tscSetMgmt
EpSet
(
&
pConnect
->
epSe
t
);
strcpy
(
pObj
->
sversion
,
pConnect
->
serverVersion
);
pObj
->
writeAuth
=
pConnect
->
writeAuth
;
...
...
src/client/src/tscSql.c
浏览文件 @
fdfa789c
...
...
@@ -62,8 +62,8 @@ SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pass, con
}
if
(
ip
)
{
if
(
tscSetMgmt
IpLis
tFromCfg
(
ip
,
NULL
)
<
0
)
return
NULL
;
if
(
port
)
tscMgmt
I
pSet
.
port
[
0
]
=
port
;
if
(
tscSetMgmt
EpSe
tFromCfg
(
ip
,
NULL
)
<
0
)
return
NULL
;
if
(
port
)
tscMgmt
E
pSet
.
port
[
0
]
=
port
;
}
void
*
pDnodeConn
=
NULL
;
...
...
src/client/src/tscSubquery.c
浏览文件 @
fdfa789c
...
...
@@ -458,7 +458,7 @@ void tscBuildVgroupTableInfo(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, SArr
break
;
}
}
assert
(
info
.
vgInfo
.
numOf
I
ps
!=
0
);
assert
(
info
.
vgInfo
.
numOf
E
ps
!=
0
);
vgTables
=
taosArrayInit
(
4
,
sizeof
(
STableIdInfo
));
info
.
itemList
=
vgTables
;
...
...
@@ -1600,8 +1600,8 @@ static void tscAllDataRetrievedFromDnode(SRetrieveSupport *trsupport, SSqlObj* p
// data in from current vnode is stored in cache and disk
uint32_t
numOfRowsFromSubquery
=
trsupport
->
pExtMemBuffer
[
idx
]
->
numOfTotalElems
+
trsupport
->
localBuffer
->
num
;
tscDebug
(
"%p sub:%p all data retrieved from
i
p:%s, vgId:%d, numOfRows:%d, orderOfSub:%d"
,
pParentSql
,
pSql
,
pTableMetaInfo
->
vgroupList
->
vgroups
[
0
].
i
pAddr
[
0
].
fqdn
,
pTableMetaInfo
->
vgroupList
->
vgroups
[
0
].
vgId
,
tscDebug
(
"%p sub:%p all data retrieved from
e
p:%s, vgId:%d, numOfRows:%d, orderOfSub:%d"
,
pParentSql
,
pSql
,
pTableMetaInfo
->
vgroupList
->
vgroups
[
0
].
e
pAddr
[
0
].
fqdn
,
pTableMetaInfo
->
vgroupList
->
vgroups
[
0
].
vgId
,
numOfRowsFromSubquery
,
idx
);
tColModelCompact
(
pDesc
->
pColumnModel
,
trsupport
->
localBuffer
,
pDesc
->
pColumnModel
->
capacity
);
...
...
@@ -1719,8 +1719,8 @@ static void tscRetrieveFromDnodeCallBack(void *param, TAOS_RES *tres, int numOfR
assert
(
pRes
->
numOfRows
==
numOfRows
);
int64_t
num
=
atomic_add_fetch_64
(
&
pState
->
numOfRetrievedRows
,
numOfRows
);
tscDebug
(
"%p sub:%p retrieve numOfRows:%"
PRId64
" totalNumOfRows:%"
PRIu64
" from
i
p:%s, orderOfSub:%d"
,
pParentSql
,
pSql
,
pRes
->
numOfRows
,
pState
->
numOfRetrievedRows
,
pSql
->
ipList
.
fqdn
[
pSql
->
ipLis
t
.
inUse
],
idx
);
tscDebug
(
"%p sub:%p retrieve numOfRows:%"
PRId64
" totalNumOfRows:%"
PRIu64
" from
e
p:%s, orderOfSub:%d"
,
pParentSql
,
pSql
,
pRes
->
numOfRows
,
pState
->
numOfRetrievedRows
,
pSql
->
epSet
.
fqdn
[
pSql
->
epSe
t
.
inUse
],
idx
);
if
(
num
>
tsMaxNumOfOrderedResults
&&
tscIsProjectionQueryOnSTable
(
pQueryInfo
,
0
))
{
tscError
(
"%p sub:%p num of OrderedRes is too many, max allowed:%"
PRId32
" , current:%"
PRId64
,
...
...
@@ -1828,8 +1828,8 @@ void tscRetrieveDataRes(void *param, TAOS_RES *tres, int code) {
return
;
}
tscTrace
(
"%p sub:%p query complete,
i
p:%s, vgId:%d, orderOfSub:%d, retrieve data"
,
trsupport
->
pParentSql
,
pSql
,
pVgroup
->
i
pAddr
[
0
].
fqdn
,
pVgroup
->
vgId
,
trsupport
->
subqueryIndex
);
tscTrace
(
"%p sub:%p query complete,
e
p:%s, vgId:%d, orderOfSub:%d, retrieve data"
,
trsupport
->
pParentSql
,
pSql
,
pVgroup
->
e
pAddr
[
0
].
fqdn
,
pVgroup
->
vgId
,
trsupport
->
subqueryIndex
);
if
(
pSql
->
res
.
qhandle
==
0
)
{
// qhandle is NULL, code is TSDB_CODE_SUCCESS means no results generated from this vnode
tscRetrieveFromDnodeCallBack
(
param
,
pSql
,
0
);
...
...
src/client/src/tscSystem.c
浏览文件 @
fdfa789c
...
...
@@ -41,7 +41,7 @@ int tscNumOfThreads;
static
pthread_once_t
tscinit
=
PTHREAD_ONCE_INIT
;
void
taosInitNote
(
int
numOfNoteLines
,
int
maxNotes
,
char
*
lable
);
void
tscUpdate
IpSet
(
void
*
ahandle
,
SRpcIpSet
*
pI
pSet
);
void
tscUpdate
EpSet
(
void
*
ahandle
,
SRpcEpSet
*
pE
pSet
);
void
tscCheckDiskUsage
(
void
*
UNUSED_PARAM
(
para
),
void
*
UNUSED_PARAM
(
param
))
{
taosGetDisk
();
...
...
@@ -116,8 +116,8 @@ void taos_init_imp() {
taosInitNote
(
tsNumOfLogLines
/
10
,
1
,
(
char
*
)
"tsc_note"
);
}
if
(
tscSetMgmt
IpLis
tFromCfg
(
tsFirst
,
tsSecond
)
<
0
)
{
tscError
(
"failed to init mnode
I
P list"
);
if
(
tscSetMgmt
EpSe
tFromCfg
(
tsFirst
,
tsSecond
)
<
0
)
{
tscError
(
"failed to init mnode
E
P list"
);
return
;
}
...
...
src/client/src/tscUtil.c
浏览文件 @
fdfa789c
...
...
@@ -2145,17 +2145,17 @@ char* strdup_throw(const char* str) {
return
p
;
}
int
tscSetMgmt
IpLis
tFromCfg
(
const
char
*
first
,
const
char
*
second
)
{
tscMgmt
IpSet
.
numOfI
ps
=
0
;
tscMgmt
I
pSet
.
inUse
=
0
;
int
tscSetMgmt
EpSe
tFromCfg
(
const
char
*
first
,
const
char
*
second
)
{
tscMgmt
EpSet
.
numOfE
ps
=
0
;
tscMgmt
E
pSet
.
inUse
=
0
;
if
(
first
&&
first
[
0
]
!=
0
)
{
if
(
strlen
(
first
)
>=
TSDB_EP_LEN
)
{
terrno
=
TSDB_CODE_TSC_INVALID_FQDN
;
return
-
1
;
}
taosGetFqdnPortFromEp
(
first
,
tscMgmt
IpSet
.
fqdn
[
tscMgmtIpSet
.
numOfIps
],
&
tscMgmtIpSet
.
port
[
tscMgmtIpSet
.
numOfI
ps
]);
tscMgmt
IpSet
.
numOfI
ps
++
;
taosGetFqdnPortFromEp
(
first
,
tscMgmt
EpSet
.
fqdn
[
tscMgmtEpSet
.
numOfEps
],
&
tscMgmtEpSet
.
port
[
tscMgmtEpSet
.
numOfE
ps
]);
tscMgmt
EpSet
.
numOfE
ps
++
;
}
if
(
second
&&
second
[
0
]
!=
0
)
{
...
...
@@ -2163,11 +2163,11 @@ int tscSetMgmtIpListFromCfg(const char *first, const char *second) {
terrno
=
TSDB_CODE_TSC_INVALID_FQDN
;
return
-
1
;
}
taosGetFqdnPortFromEp
(
second
,
tscMgmt
IpSet
.
fqdn
[
tscMgmtIpSet
.
numOfIps
],
&
tscMgmtIpSet
.
port
[
tscMgmtIpSet
.
numOfI
ps
]);
tscMgmt
IpSet
.
numOfI
ps
++
;
taosGetFqdnPortFromEp
(
second
,
tscMgmt
EpSet
.
fqdn
[
tscMgmtEpSet
.
numOfEps
],
&
tscMgmtEpSet
.
port
[
tscMgmtEpSet
.
numOfE
ps
]);
tscMgmt
EpSet
.
numOfE
ps
++
;
}
if
(
tscMgmt
IpSet
.
numOfI
ps
==
0
)
{
if
(
tscMgmt
EpSet
.
numOfE
ps
==
0
)
{
terrno
=
TSDB_CODE_TSC_INVALID_FQDN
;
return
-
1
;
}
...
...
src/dnode/inc/dnodeMgmt.h
浏览文件 @
fdfa789c
...
...
@@ -35,8 +35,8 @@ void* dnodeGetVnodeTsdb(void *pVnode);
void
dnodeReleaseVnode
(
void
*
pVnode
);
void
dnodeSendRedirectMsg
(
SRpcMsg
*
rpcMsg
,
bool
forShell
);
void
dnodeGetMnode
IpSetForPeer
(
void
*
i
pSet
);
void
dnodeGetMnode
IpSetForShell
(
void
*
i
pSet
);
void
dnodeGetMnode
EpSetForPeer
(
void
*
e
pSet
);
void
dnodeGetMnode
EpSetForShell
(
void
*
e
pSet
);
#ifdef __cplusplus
}
...
...
src/dnode/src/dnodeMgmt.c
浏览文件 @
fdfa789c
...
...
@@ -52,7 +52,7 @@ void * tsDnodeTmr = NULL;
static
void
*
tsStatusTimer
=
NULL
;
static
uint32_t
tsRebootTime
;
static
SRpc
IpSet
tsDMnodeI
pSet
=
{
0
};
static
SRpc
EpSet
tsDMnodeE
pSet
=
{
0
};
static
SDMMnodeInfos
tsDMnodeInfos
=
{
0
};
static
SDMDnodeCfg
tsDnodeCfg
=
{
0
};
static
taos_qset
tsMgmtQset
=
NULL
;
...
...
@@ -90,21 +90,21 @@ int32_t dnodeInitMgmt() {
tsRebootTime
=
taosGetTimestampSec
();
if
(
!
dnodeReadMnodeInfos
())
{
memset
(
&
tsDMnode
IpSet
,
0
,
sizeof
(
SRpcI
pSet
));
memset
(
&
tsDMnode
EpSet
,
0
,
sizeof
(
SRpcE
pSet
));
memset
(
&
tsDMnodeInfos
,
0
,
sizeof
(
SDMMnodeInfos
));
tsDMnode
IpSet
.
numOfI
ps
=
1
;
taosGetFqdnPortFromEp
(
tsFirst
,
tsDMnode
IpSet
.
fqdn
[
0
],
&
tsDMnodeI
pSet
.
port
[
0
]);
tsDMnode
EpSet
.
numOfE
ps
=
1
;
taosGetFqdnPortFromEp
(
tsFirst
,
tsDMnode
EpSet
.
fqdn
[
0
],
&
tsDMnodeE
pSet
.
port
[
0
]);
if
(
strcmp
(
tsSecond
,
tsFirst
)
!=
0
)
{
tsDMnode
IpSet
.
numOfI
ps
=
2
;
taosGetFqdnPortFromEp
(
tsSecond
,
tsDMnode
IpSet
.
fqdn
[
1
],
&
tsDMnodeI
pSet
.
port
[
1
]);
tsDMnode
EpSet
.
numOfE
ps
=
2
;
taosGetFqdnPortFromEp
(
tsSecond
,
tsDMnode
EpSet
.
fqdn
[
1
],
&
tsDMnodeE
pSet
.
port
[
1
]);
}
}
else
{
tsDMnode
I
pSet
.
inUse
=
tsDMnodeInfos
.
inUse
;
tsDMnode
IpSet
.
numOfI
ps
=
tsDMnodeInfos
.
nodeNum
;
tsDMnode
E
pSet
.
inUse
=
tsDMnodeInfos
.
inUse
;
tsDMnode
EpSet
.
numOfE
ps
=
tsDMnodeInfos
.
nodeNum
;
for
(
int32_t
i
=
0
;
i
<
tsDMnodeInfos
.
nodeNum
;
i
++
)
{
taosGetFqdnPortFromEp
(
tsDMnodeInfos
.
nodeInfos
[
i
].
nodeEp
,
tsDMnode
IpSet
.
fqdn
[
i
],
&
tsDMnodeI
pSet
.
port
[
i
]);
taosGetFqdnPortFromEp
(
tsDMnodeInfos
.
nodeInfos
[
i
].
nodeEp
,
tsDMnode
EpSet
.
fqdn
[
i
],
&
tsDMnodeE
pSet
.
port
[
i
]);
}
}
...
...
@@ -450,27 +450,27 @@ static int32_t dnodeProcessConfigDnodeMsg(SRpcMsg *pMsg) {
return
taosCfgDynamicOptions
(
pCfg
->
config
);
}
void
dnodeUpdateMnode
IpSetForPeer
(
SRpcIpSet
*
pI
pSet
)
{
dInfo
(
"mnode
IP list for is changed, numOfIps:%d inUse:%d"
,
pIpSet
->
numOfIps
,
pI
pSet
->
inUse
);
for
(
int
i
=
0
;
i
<
p
IpSet
->
numOfI
ps
;
++
i
)
{
p
I
pSet
->
port
[
i
]
-=
TSDB_PORT_DNODEDNODE
;
dInfo
(
"mnode index:%d %s:%u"
,
i
,
p
IpSet
->
fqdn
[
i
],
pI
pSet
->
port
[
i
])
void
dnodeUpdateMnode
EpSetForPeer
(
SRpcEpSet
*
pE
pSet
)
{
dInfo
(
"mnode
EP list for is changed, numOfEps:%d inUse:%d"
,
pEpSet
->
numOfEps
,
pE
pSet
->
inUse
);
for
(
int
i
=
0
;
i
<
p
EpSet
->
numOfE
ps
;
++
i
)
{
p
E
pSet
->
port
[
i
]
-=
TSDB_PORT_DNODEDNODE
;
dInfo
(
"mnode index:%d %s:%u"
,
i
,
p
EpSet
->
fqdn
[
i
],
pE
pSet
->
port
[
i
])
}
tsDMnode
IpSet
=
*
pI
pSet
;
tsDMnode
EpSet
=
*
pE
pSet
;
}
void
dnodeGetMnode
IpSetForPeer
(
void
*
i
pSetRaw
)
{
SRpc
IpSet
*
ipSet
=
i
pSetRaw
;
*
ipSet
=
tsDMnodeI
pSet
;
void
dnodeGetMnode
EpSetForPeer
(
void
*
e
pSetRaw
)
{
SRpc
EpSet
*
epSet
=
e
pSetRaw
;
*
epSet
=
tsDMnodeE
pSet
;
for
(
int
i
=
0
;
i
<
ipSet
->
numOfI
ps
;
++
i
)
i
pSet
->
port
[
i
]
+=
TSDB_PORT_DNODEDNODE
;
for
(
int
i
=
0
;
i
<
epSet
->
numOfE
ps
;
++
i
)
e
pSet
->
port
[
i
]
+=
TSDB_PORT_DNODEDNODE
;
}
void
dnodeGetMnode
IpSetForShell
(
void
*
i
pSetRaw
)
{
SRpc
IpSet
*
ipSet
=
i
pSetRaw
;
*
ipSet
=
tsDMnodeI
pSet
;
void
dnodeGetMnode
EpSetForShell
(
void
*
e
pSetRaw
)
{
SRpc
EpSet
*
epSet
=
e
pSetRaw
;
*
epSet
=
tsDMnodeE
pSet
;
}
static
void
dnodeProcessStatusRsp
(
SRpcMsg
*
pMsg
)
{
...
...
@@ -536,10 +536,10 @@ static void dnodeUpdateMnodeInfos(SDMMnodeInfos *pMnodes) {
dInfo
(
"mnode index:%d, %s"
,
tsDMnodeInfos
.
nodeInfos
[
i
].
nodeId
,
tsDMnodeInfos
.
nodeInfos
[
i
].
nodeEp
);
}
tsDMnode
I
pSet
.
inUse
=
tsDMnodeInfos
.
inUse
;
tsDMnode
IpSet
.
numOfI
ps
=
tsDMnodeInfos
.
nodeNum
;
tsDMnode
E
pSet
.
inUse
=
tsDMnodeInfos
.
inUse
;
tsDMnode
EpSet
.
numOfE
ps
=
tsDMnodeInfos
.
nodeNum
;
for
(
int32_t
i
=
0
;
i
<
tsDMnodeInfos
.
nodeNum
;
i
++
)
{
taosGetFqdnPortFromEp
(
tsDMnodeInfos
.
nodeInfos
[
i
].
nodeEp
,
tsDMnode
IpSet
.
fqdn
[
i
],
&
tsDMnodeI
pSet
.
port
[
i
]);
taosGetFqdnPortFromEp
(
tsDMnodeInfos
.
nodeInfos
[
i
].
nodeEp
,
tsDMnode
EpSet
.
fqdn
[
i
],
&
tsDMnodeE
pSet
.
port
[
i
]);
}
dnodeSaveMnodeInfos
();
...
...
@@ -549,10 +549,10 @@ static void dnodeUpdateMnodeInfos(SDMMnodeInfos *pMnodes) {
static
bool
dnodeReadMnodeInfos
()
{
char
ipFile
[
TSDB_FILENAME_LEN
*
2
]
=
{
0
};
sprintf
(
ipFile
,
"%s/mnode
IpLis
t.json"
,
tsDnodeDir
);
sprintf
(
ipFile
,
"%s/mnode
EpSe
t.json"
,
tsDnodeDir
);
FILE
*
fp
=
fopen
(
ipFile
,
"r"
);
if
(
!
fp
)
{
dDebug
(
"failed to read mnode
IpLis
t.json, file not exist"
);
dDebug
(
"failed to read mnode
EpSe
t.json, file not exist"
);
return
false
;
}
...
...
@@ -563,40 +563,40 @@ static bool dnodeReadMnodeInfos() {
if
(
len
<=
0
)
{
free
(
content
);
fclose
(
fp
);
dError
(
"failed to read mnode
IpLis
t.json, content is null"
);
dError
(
"failed to read mnode
EpSe
t.json, content is null"
);
return
false
;
}
content
[
len
]
=
0
;
cJSON
*
root
=
cJSON_Parse
(
content
);
if
(
root
==
NULL
)
{
dError
(
"failed to read mnode
IpLis
t.json, invalid json format"
);
dError
(
"failed to read mnode
EpSe
t.json, invalid json format"
);
goto
PARSE_OVER
;
}
cJSON
*
inUse
=
cJSON_GetObjectItem
(
root
,
"inUse"
);
if
(
!
inUse
||
inUse
->
type
!=
cJSON_Number
)
{
dError
(
"failed to read mnode
IpLis
t.json, inUse not found"
);
dError
(
"failed to read mnode
EpSe
t.json, inUse not found"
);
goto
PARSE_OVER
;
}
tsDMnodeInfos
.
inUse
=
inUse
->
valueint
;
cJSON
*
nodeNum
=
cJSON_GetObjectItem
(
root
,
"nodeNum"
);
if
(
!
nodeNum
||
nodeNum
->
type
!=
cJSON_Number
)
{
dError
(
"failed to read mnode
IpLis
t.json, nodeNum not found"
);
dError
(
"failed to read mnode
EpSe
t.json, nodeNum not found"
);
goto
PARSE_OVER
;
}
tsDMnodeInfos
.
nodeNum
=
nodeNum
->
valueint
;
cJSON
*
nodeInfos
=
cJSON_GetObjectItem
(
root
,
"nodeInfos"
);
if
(
!
nodeInfos
||
nodeInfos
->
type
!=
cJSON_Array
)
{
dError
(
"failed to read mnode
IpLis
t.json, nodeInfos not found"
);
dError
(
"failed to read mnode
EpSe
t.json, nodeInfos not found"
);
goto
PARSE_OVER
;
}
int
size
=
cJSON_GetArraySize
(
nodeInfos
);
if
(
size
!=
tsDMnodeInfos
.
nodeNum
)
{
dError
(
"failed to read mnode
IpLis
t.json, nodeInfos size not matched"
);
dError
(
"failed to read mnode
EpSe
t.json, nodeInfos size not matched"
);
goto
PARSE_OVER
;
}
...
...
@@ -606,14 +606,14 @@ static bool dnodeReadMnodeInfos() {
cJSON
*
nodeId
=
cJSON_GetObjectItem
(
nodeInfo
,
"nodeId"
);
if
(
!
nodeId
||
nodeId
->
type
!=
cJSON_Number
)
{
dError
(
"failed to read mnode
IpLis
t.json, nodeId not found"
);
dError
(
"failed to read mnode
EpSe
t.json, nodeId not found"
);
goto
PARSE_OVER
;
}
tsDMnodeInfos
.
nodeInfos
[
i
].
nodeId
=
nodeId
->
valueint
;
cJSON
*
nodeEp
=
cJSON_GetObjectItem
(
nodeInfo
,
"nodeEp"
);
if
(
!
nodeEp
||
nodeEp
->
type
!=
cJSON_String
||
nodeEp
->
valuestring
==
NULL
)
{
dError
(
"failed to read mnode
IpLis
t.json, nodeName not found"
);
dError
(
"failed to read mnode
EpSe
t.json, nodeName not found"
);
goto
PARSE_OVER
;
}
strncpy
(
tsDMnodeInfos
.
nodeInfos
[
i
].
nodeEp
,
nodeEp
->
valuestring
,
TSDB_EP_LEN
);
...
...
@@ -621,7 +621,7 @@ static bool dnodeReadMnodeInfos() {
ret
=
true
;
dInfo
(
"read mnode
iplist successed, numOfI
ps:%d inUse:%d"
,
tsDMnodeInfos
.
nodeNum
,
tsDMnodeInfos
.
inUse
);
dInfo
(
"read mnode
epSet successed, numOfE
ps:%d inUse:%d"
,
tsDMnodeInfos
.
nodeNum
,
tsDMnodeInfos
.
inUse
);
for
(
int32_t
i
=
0
;
i
<
tsDMnodeInfos
.
nodeNum
;
i
++
)
{
dInfo
(
"mnode:%d, %s"
,
tsDMnodeInfos
.
nodeInfos
[
i
].
nodeId
,
tsDMnodeInfos
.
nodeInfos
[
i
].
nodeEp
);
}
...
...
@@ -635,7 +635,7 @@ PARSE_OVER:
static
void
dnodeSaveMnodeInfos
()
{
char
ipFile
[
TSDB_FILENAME_LEN
]
=
{
0
};
sprintf
(
ipFile
,
"%s/mnode
IpLis
t.json"
,
tsDnodeDir
);
sprintf
(
ipFile
,
"%s/mnode
EpSe
t.json"
,
tsDnodeDir
);
FILE
*
fp
=
fopen
(
ipFile
,
"w"
);
if
(
!
fp
)
return
;
...
...
@@ -663,11 +663,11 @@ static void dnodeSaveMnodeInfos() {
fclose
(
fp
);
free
(
content
);
dInfo
(
"save mnode
iplis
t successed"
);
dInfo
(
"save mnode
epSe
t successed"
);
}
char
*
dnodeGetMnodeMasterEp
()
{
return
tsDMnodeInfos
.
nodeInfos
[
tsDMnode
I
pSet
.
inUse
].
nodeEp
;
return
tsDMnodeInfos
.
nodeInfos
[
tsDMnode
E
pSet
.
inUse
].
nodeEp
;
}
void
*
dnodeGetMnodeInfos
()
{
...
...
@@ -726,9 +726,9 @@ static void dnodeSendStatusMsg(void *handle, void *tmrId) {
.
msgType
=
TSDB_MSG_TYPE_DM_STATUS
};
SRpc
IpSet
i
pSet
;
dnodeGetMnode
IpSetForPeer
(
&
i
pSet
);
dnodeSendMsgToDnode
(
&
i
pSet
,
&
rpcMsg
);
SRpc
EpSet
e
pSet
;
dnodeGetMnode
EpSetForPeer
(
&
e
pSet
);
dnodeSendMsgToDnode
(
&
e
pSet
,
&
rpcMsg
);
}
static
bool
dnodeReadDnodeCfg
()
{
...
...
@@ -817,20 +817,20 @@ void dnodeSendRedirectMsg(SRpcMsg *rpcMsg, bool forShell) {
SRpcConnInfo
connInfo
=
{
0
};
rpcGetConnInfo
(
rpcMsg
->
handle
,
&
connInfo
);
SRpc
IpSet
i
pSet
=
{
0
};
SRpc
EpSet
e
pSet
=
{
0
};
if
(
forShell
)
{
dnodeGetMnode
IpSetForShell
(
&
i
pSet
);
dnodeGetMnode
EpSetForShell
(
&
e
pSet
);
}
else
{
dnodeGetMnode
IpSetForPeer
(
&
i
pSet
);
dnodeGetMnode
EpSetForPeer
(
&
e
pSet
);
}
dDebug
(
"msg:%s will be redirected, dnodeIp:%s user:%s, numOf
I
ps:%d inUse:%d"
,
taosMsg
[
rpcMsg
->
msgType
],
taosIpStr
(
connInfo
.
clientIp
),
connInfo
.
user
,
ipSet
.
numOfIps
,
i
pSet
.
inUse
);
dDebug
(
"msg:%s will be redirected, dnodeIp:%s user:%s, numOf
E
ps:%d inUse:%d"
,
taosMsg
[
rpcMsg
->
msgType
],
taosIpStr
(
connInfo
.
clientIp
),
connInfo
.
user
,
epSet
.
numOfEps
,
e
pSet
.
inUse
);
for
(
int
i
=
0
;
i
<
ipSet
.
numOfI
ps
;
++
i
)
{
dDebug
(
"mnode index:%d %s:%d"
,
i
,
ipSet
.
fqdn
[
i
],
i
pSet
.
port
[
i
]);
ipSet
.
port
[
i
]
=
htons
(
i
pSet
.
port
[
i
]);
for
(
int
i
=
0
;
i
<
epSet
.
numOfE
ps
;
++
i
)
{
dDebug
(
"mnode index:%d %s:%d"
,
i
,
epSet
.
fqdn
[
i
],
e
pSet
.
port
[
i
]);
epSet
.
port
[
i
]
=
htons
(
e
pSet
.
port
[
i
]);
}
rpcSendRedirectRsp
(
rpcMsg
->
handle
,
&
i
pSet
);
rpcSendRedirectRsp
(
rpcMsg
->
handle
,
&
e
pSet
);
}
src/dnode/src/dnodePeer.c
浏览文件 @
fdfa789c
...
...
@@ -29,11 +29,11 @@
#include "dnodeVWrite.h"
#include "dnodeMPeer.h"
extern
void
dnodeUpdateMnode
IpSetForPeer
(
SRpcIpSet
*
pI
pSet
);
extern
void
dnodeUpdateMnode
EpSetForPeer
(
SRpcEpSet
*
pE
pSet
);
static
void
(
*
dnodeProcessReqMsgFp
[
TSDB_MSG_TYPE_MAX
])(
SRpcMsg
*
);
static
void
dnodeProcessReqMsgFromDnode
(
SRpcMsg
*
pMsg
,
SRpc
I
pSet
*
);
static
void
dnodeProcessReqMsgFromDnode
(
SRpcMsg
*
pMsg
,
SRpc
E
pSet
*
);
static
void
(
*
dnodeProcessRspMsgFp
[
TSDB_MSG_TYPE_MAX
])(
SRpcMsg
*
rpcMsg
);
static
void
dnodeProcessRspFromDnode
(
SRpcMsg
*
pMsg
,
SRpc
IpSet
*
pI
pSet
);
static
void
dnodeProcessRspFromDnode
(
SRpcMsg
*
pMsg
,
SRpc
EpSet
*
pE
pSet
);
static
void
*
tsDnodeServerRpc
=
NULL
;
static
void
*
tsDnodeClientRpc
=
NULL
;
...
...
@@ -83,7 +83,7 @@ void dnodeCleanupServer() {
}
}
static
void
dnodeProcessReqMsgFromDnode
(
SRpcMsg
*
pMsg
,
SRpc
IpSet
*
pI
pSet
)
{
static
void
dnodeProcessReqMsgFromDnode
(
SRpcMsg
*
pMsg
,
SRpc
EpSet
*
pE
pSet
)
{
SRpcMsg
rspMsg
=
{
.
handle
=
pMsg
->
handle
,
.
pCont
=
NULL
,
...
...
@@ -148,9 +148,9 @@ void dnodeCleanupClient() {
}
}
static
void
dnodeProcessRspFromDnode
(
SRpcMsg
*
pMsg
,
SRpc
IpSet
*
pI
pSet
)
{
if
(
pMsg
->
msgType
==
TSDB_MSG_TYPE_DM_STATUS_RSP
&&
p
I
pSet
)
{
dnodeUpdateMnode
IpSetForPeer
(
pI
pSet
);
static
void
dnodeProcessRspFromDnode
(
SRpcMsg
*
pMsg
,
SRpc
EpSet
*
pE
pSet
)
{
if
(
pMsg
->
msgType
==
TSDB_MSG_TYPE_DM_STATUS_RSP
&&
p
E
pSet
)
{
dnodeUpdateMnode
EpSetForPeer
(
pE
pSet
);
}
if
(
dnodeProcessRspMsgFp
[
pMsg
->
msgType
])
{
...
...
@@ -166,12 +166,12 @@ void dnodeAddClientRspHandle(uint8_t msgType, void (*fp)(SRpcMsg *rpcMsg)) {
dnodeProcessRspMsgFp
[
msgType
]
=
fp
;
}
void
dnodeSendMsgToDnode
(
SRpc
IpSet
*
i
pSet
,
SRpcMsg
*
rpcMsg
)
{
rpcSendRequest
(
tsDnodeClientRpc
,
i
pSet
,
rpcMsg
);
void
dnodeSendMsgToDnode
(
SRpc
EpSet
*
e
pSet
,
SRpcMsg
*
rpcMsg
)
{
rpcSendRequest
(
tsDnodeClientRpc
,
e
pSet
,
rpcMsg
);
}
void
dnodeSendMsgToDnodeRecv
(
SRpcMsg
*
rpcMsg
,
SRpcMsg
*
rpcRsp
)
{
SRpc
IpSet
i
pSet
=
{
0
};
dnodeGetMnode
IpSetForPeer
(
&
i
pSet
);
rpcSendRecv
(
tsDnodeClientRpc
,
&
i
pSet
,
rpcMsg
,
rpcRsp
);
SRpc
EpSet
e
pSet
=
{
0
};
dnodeGetMnode
EpSetForPeer
(
&
e
pSet
);
rpcSendRecv
(
tsDnodeClientRpc
,
&
e
pSet
,
rpcMsg
,
rpcRsp
);
}
src/dnode/src/dnodeShell.c
浏览文件 @
fdfa789c
...
...
@@ -31,7 +31,7 @@
#include "dnodeShell.h"
static
void
(
*
dnodeProcessShellMsgFp
[
TSDB_MSG_TYPE_MAX
])(
SRpcMsg
*
);
static
void
dnodeProcessMsgFromShell
(
SRpcMsg
*
pMsg
,
SRpc
I
pSet
*
);
static
void
dnodeProcessMsgFromShell
(
SRpcMsg
*
pMsg
,
SRpc
E
pSet
*
);
static
int
dnodeRetrieveUserAuthInfo
(
char
*
user
,
char
*
spi
,
char
*
encrypt
,
char
*
secret
,
char
*
ckey
);
static
void
*
tsDnodeShellRpc
=
NULL
;
static
int32_t
tsDnodeQueryReqNum
=
0
;
...
...
@@ -108,7 +108,7 @@ void dnodeCleanupShell() {
}
}
void
dnodeProcessMsgFromShell
(
SRpcMsg
*
pMsg
,
SRpc
IpSet
*
pI
pSet
)
{
void
dnodeProcessMsgFromShell
(
SRpcMsg
*
pMsg
,
SRpc
EpSet
*
pE
pSet
)
{
SRpcMsg
rpcMsg
=
{
.
handle
=
pMsg
->
handle
,
.
pCont
=
NULL
,
...
...
src/inc/dnode.h
浏览文件 @
fdfa789c
...
...
@@ -39,13 +39,13 @@ SDnodeStatisInfo dnodeGetStatisInfo();
bool
dnodeIsFirstDeploy
();
char
*
dnodeGetMnodeMasterEp
();
void
dnodeGetMnode
IpSetForPeer
(
void
*
i
pSet
);
void
dnodeGetMnode
IpSetForShell
(
void
*
i
pSet
);
void
dnodeGetMnode
EpSetForPeer
(
void
*
e
pSet
);
void
dnodeGetMnode
EpSetForShell
(
void
*
e
pSet
);
void
*
dnodeGetMnodeInfos
();
int32_t
dnodeGetDnodeId
();
void
dnodeAddClientRspHandle
(
uint8_t
msgType
,
void
(
*
fp
)(
SRpcMsg
*
rpcMsg
));
void
dnodeSendMsgToDnode
(
SRpc
IpSet
*
i
pSet
,
SRpcMsg
*
rpcMsg
);
void
dnodeSendMsgToDnode
(
SRpc
EpSet
*
e
pSet
,
SRpcMsg
*
rpcMsg
);
void
dnodeSendMsgToDnodeRecv
(
SRpcMsg
*
rpcMsg
,
SRpcMsg
*
rpcRsp
);
void
*
dnodeSendCfgTableToRecv
(
int32_t
vgId
,
int32_t
sid
);
...
...
src/inc/taosmsg.h
浏览文件 @
fdfa789c
...
...
@@ -176,7 +176,7 @@ extern char *taosMsg[];
typedef
struct
{
char
fqdn
[
TSDB_FQDN_LEN
];
uint16_t
port
;
}
S
I
pAddr
;
}
S
E
pAddr
;
typedef
struct
{
int32_t
numOfVnodes
;
...
...
@@ -306,7 +306,7 @@ typedef struct {
int8_t
reserved1
;
int8_t
reserved2
;
int32_t
connId
;
SRpc
IpSet
ipLis
t
;
SRpc
EpSet
epSe
t
;
}
SCMConnectRsp
;
typedef
struct
{
...
...
@@ -648,8 +648,8 @@ typedef struct SCMSTableVgroupMsg {
typedef
struct
{
int32_t
vgId
;
int8_t
numOf
I
ps
;
S
IpAddr
i
pAddr
[
TSDB_MAX_REPLICA
];
int8_t
numOf
E
ps
;
S
EpAddr
e
pAddr
[
TSDB_MAX_REPLICA
];
}
SCMVgroupInfo
;
typedef
struct
{
...
...
@@ -753,7 +753,7 @@ typedef struct {
uint32_t
onlineDnodes
;
uint32_t
connId
;
int8_t
killConnection
;
SRpc
IpSet
ipLis
t
;
SRpc
EpSet
epSe
t
;
}
SCMHeartBeatRsp
;
typedef
struct
{
...
...
src/inc/trpc.h
浏览文件 @
fdfa789c
...
...
@@ -28,12 +28,12 @@ extern "C" {
extern
int
tsRpcHeadSize
;
typedef
struct
SRpc
I
pSet
{
typedef
struct
SRpc
E
pSet
{
int8_t
inUse
;
int8_t
numOf
I
ps
;
int8_t
numOf
E
ps
;
uint16_t
port
[
TSDB_MAX_REPLICA
];
char
fqdn
[
TSDB_MAX_REPLICA
][
TSDB_FQDN_LEN
];
}
SRpc
I
pSet
;
}
SRpc
E
pSet
;
typedef
struct
SRpcConnInfo
{
uint32_t
clientIp
;
...
...
@@ -67,7 +67,7 @@ typedef struct SRpcInit {
char
*
ckey
;
// ciphering key
// call back to process incoming msg, code shall be ignored by server app
void
(
*
cfp
)(
SRpcMsg
*
,
SRpc
I
pSet
*
);
void
(
*
cfp
)(
SRpcMsg
*
,
SRpc
E
pSet
*
);
// call back to retrieve the client auth info, for server app only
int
(
*
afp
)(
char
*
tableId
,
char
*
spi
,
char
*
encrypt
,
char
*
secret
,
char
*
ckey
);
...
...
@@ -78,11 +78,11 @@ void rpcClose(void *);
void
*
rpcMallocCont
(
int
contLen
);
void
rpcFreeCont
(
void
*
pCont
);
void
*
rpcReallocCont
(
void
*
ptr
,
int
contLen
);
void
rpcSendRequest
(
void
*
thandle
,
const
SRpc
IpSet
*
pI
pSet
,
SRpcMsg
*
pMsg
);
void
rpcSendRequest
(
void
*
thandle
,
const
SRpc
EpSet
*
pE
pSet
,
SRpcMsg
*
pMsg
);
void
rpcSendResponse
(
const
SRpcMsg
*
pMsg
);
void
rpcSendRedirectRsp
(
void
*
pConn
,
const
SRpc
IpSet
*
pI
pSet
);
void
rpcSendRedirectRsp
(
void
*
pConn
,
const
SRpc
EpSet
*
pE
pSet
);
int
rpcGetConnInfo
(
void
*
thandle
,
SRpcConnInfo
*
pInfo
);
void
rpcSendRecv
(
void
*
shandle
,
SRpc
IpSet
*
pI
pSet
,
SRpcMsg
*
pReq
,
SRpcMsg
*
pRsp
);
void
rpcSendRecv
(
void
*
shandle
,
SRpc
EpSet
*
pE
pSet
,
SRpcMsg
*
pReq
,
SRpcMsg
*
pRsp
);
int
rpcReportProgress
(
void
*
pConn
,
char
*
pCont
,
int
contLen
);
void
rpcCancelRequest
(
void
*
pContext
);
...
...
src/kit/taosmigrate/taosmigrate.c
浏览文件 @
fdfa789c
...
...
@@ -210,10 +210,10 @@ int32_t main(int32_t argc, char *argv[]) {
(
void
)
snprintf
(
mnodeWal
,
TSDB_FILENAME_LEN
*
2
,
"%s/mnode/wal/wal0"
,
arguments
.
dataDir
);
walModWalFile
(
mnodeWal
);
// 2. modfiy dnode config: mnode
IpLis
t.json
char
dnode
IpLis
t
[
TSDB_FILENAME_LEN
*
2
]
=
{
0
};
(
void
)
snprintf
(
dnode
IpList
,
TSDB_FILENAME_LEN
*
2
,
"%s/dnode/mnodeIpLis
t.json"
,
arguments
.
dataDir
);
modDnode
IpList
(
dnodeIpLis
t
);
// 2. modfiy dnode config: mnode
EpSe
t.json
char
dnode
EpSe
t
[
TSDB_FILENAME_LEN
*
2
]
=
{
0
};
(
void
)
snprintf
(
dnode
EpSet
,
TSDB_FILENAME_LEN
*
2
,
"%s/dnode/mnodeEpSe
t.json"
,
arguments
.
dataDir
);
modDnode
EpSet
(
dnodeEpSe
t
);
// 3. modify vnode config: config.json
char
vnodeDir
[
TSDB_FILENAME_LEN
*
2
]
=
{
0
};
...
...
src/kit/taosmigrate/taosmigrate.h
浏览文件 @
fdfa789c
...
...
@@ -71,7 +71,7 @@ int tSystemShell(const char * cmd);
void
taosMvFile
(
char
*
destFile
,
char
*
srcFile
)
;
void
walModWalFile
(
char
*
walfile
);
SdnodeIfo
*
getDnodeInfo
(
int32_t
dnodeId
);
void
modDnode
IpList
(
char
*
dnodeIpLis
t
);
void
modDnode
EpSet
(
char
*
dnodeEpSe
t
);
void
modAllVnode
(
char
*
vnodeDir
);
#endif
src/kit/taosmigrate/taosmigrateDnodeCfg.c
浏览文件 @
fdfa789c
...
...
@@ -23,10 +23,10 @@
static
SDMMnodeInfos
tsDnodeIpInfos
=
{
0
};
static
bool
dnodeReadMnodeInfos
(
char
*
dnode
IpLis
t
)
{
FILE
*
fp
=
fopen
(
dnode
IpLis
t
,
"r"
);
static
bool
dnodeReadMnodeInfos
(
char
*
dnode
EpSe
t
)
{
FILE
*
fp
=
fopen
(
dnode
EpSe
t
,
"r"
);
if
(
!
fp
)
{
printf
(
"failed to read mnode
IpLis
t.json, file not exist
\n
"
);
printf
(
"failed to read mnode
EpSe
t.json, file not exist
\n
"
);
return
false
;
}
...
...
@@ -37,40 +37,40 @@ static bool dnodeReadMnodeInfos(char* dnodeIpList) {
if
(
len
<=
0
)
{
free
(
content
);
fclose
(
fp
);
printf
(
"failed to read mnode
IpLis
t.json, content is null
\n
"
);
printf
(
"failed to read mnode
EpSe
t.json, content is null
\n
"
);
return
false
;
}
content
[
len
]
=
0
;
cJSON
*
root
=
cJSON_Parse
(
content
);
if
(
root
==
NULL
)
{
printf
(
"failed to read mnode
IpLis
t.json, invalid json format
\n
"
);
printf
(
"failed to read mnode
EpSe
t.json, invalid json format
\n
"
);
goto
PARSE_OVER
;
}
cJSON
*
inUse
=
cJSON_GetObjectItem
(
root
,
"inUse"
);
if
(
!
inUse
||
inUse
->
type
!=
cJSON_Number
)
{
printf
(
"failed to read mnode
IpLis
t.json, inUse not found
\n
"
);
printf
(
"failed to read mnode
EpSe
t.json, inUse not found
\n
"
);
goto
PARSE_OVER
;
}
tsDnodeIpInfos
.
inUse
=
inUse
->
valueint
;
cJSON
*
nodeNum
=
cJSON_GetObjectItem
(
root
,
"nodeNum"
);
if
(
!
nodeNum
||
nodeNum
->
type
!=
cJSON_Number
)
{
printf
(
"failed to read mnode
IpLis
t.json, nodeNum not found
\n
"
);
printf
(
"failed to read mnode
EpSe
t.json, nodeNum not found
\n
"
);
goto
PARSE_OVER
;
}
tsDnodeIpInfos
.
nodeNum
=
nodeNum
->
valueint
;
cJSON
*
nodeInfos
=
cJSON_GetObjectItem
(
root
,
"nodeInfos"
);
if
(
!
nodeInfos
||
nodeInfos
->
type
!=
cJSON_Array
)
{
printf
(
"failed to read mnode
IpLis
t.json, nodeInfos not found
\n
"
);
printf
(
"failed to read mnode
EpSe
t.json, nodeInfos not found
\n
"
);
goto
PARSE_OVER
;
}
int
size
=
cJSON_GetArraySize
(
nodeInfos
);
if
(
size
!=
tsDnodeIpInfos
.
nodeNum
)
{
printf
(
"failed to read mnode
IpLis
t.json, nodeInfos size not matched
\n
"
);
printf
(
"failed to read mnode
EpSe
t.json, nodeInfos size not matched
\n
"
);
goto
PARSE_OVER
;
}
...
...
@@ -80,14 +80,14 @@ static bool dnodeReadMnodeInfos(char* dnodeIpList) {
cJSON
*
nodeId
=
cJSON_GetObjectItem
(
nodeInfo
,
"nodeId"
);
if
(
!
nodeId
||
nodeId
->
type
!=
cJSON_Number
)
{
printf
(
"failed to read mnode
IpLis
t.json, nodeId not found
\n
"
);
printf
(
"failed to read mnode
EpSe
t.json, nodeId not found
\n
"
);
goto
PARSE_OVER
;
}
tsDnodeIpInfos
.
nodeInfos
[
i
].
nodeId
=
nodeId
->
valueint
;
cJSON
*
nodeEp
=
cJSON_GetObjectItem
(
nodeInfo
,
"nodeEp"
);
if
(
!
nodeEp
||
nodeEp
->
type
!=
cJSON_String
||
nodeEp
->
valuestring
==
NULL
)
{
printf
(
"failed to read mnode
IpLis
t.json, nodeName not found
\n
"
);
printf
(
"failed to read mnode
EpSe
t.json, nodeName not found
\n
"
);
goto
PARSE_OVER
;
}
strncpy
(
tsDnodeIpInfos
.
nodeInfos
[
i
].
nodeEp
,
nodeEp
->
valuestring
,
TSDB_EP_LEN
);
...
...
@@ -102,7 +102,7 @@ static bool dnodeReadMnodeInfos(char* dnodeIpList) {
ret
=
true
;
//printf("read mnode
iplist successed, numOfI
ps:%d inUse:%d\n", tsDnodeIpInfos.nodeNum, tsDnodeIpInfos.inUse);
//printf("read mnode
epSet successed, numOfE
ps:%d inUse:%d\n", tsDnodeIpInfos.nodeNum, tsDnodeIpInfos.inUse);
//for (int32_t i = 0; i < tsDnodeIpInfos.nodeNum; i++) {
// printf("mnode:%d, %s\n", tsDnodeIpInfos.nodeInfos[i].nodeId, tsDnodeIpInfos.nodeInfos[i].nodeEp);
//}
...
...
@@ -115,8 +115,8 @@ PARSE_OVER:
}
static
void
dnodeSaveMnodeInfos
(
char
*
dnode
IpLis
t
)
{
FILE
*
fp
=
fopen
(
dnode
IpLis
t
,
"w"
);
static
void
dnodeSaveMnodeInfos
(
char
*
dnode
EpSe
t
)
{
FILE
*
fp
=
fopen
(
dnode
EpSe
t
,
"w"
);
if
(
!
fp
)
return
;
int32_t
len
=
0
;
...
...
@@ -143,13 +143,13 @@ static void dnodeSaveMnodeInfos(char* dnodeIpList) {
fclose
(
fp
);
free
(
content
);
printf
(
"mod mnode
iplis
t successed
\n
"
);
printf
(
"mod mnode
epSe
t successed
\n
"
);
}
void
modDnode
IpList
(
char
*
dnodeIpLis
t
)
void
modDnode
EpSet
(
char
*
dnodeEpSe
t
)
{
(
void
)
dnodeReadMnodeInfos
(
dnode
IpLis
t
);
dnodeSaveMnodeInfos
(
dnode
IpLis
t
);
(
void
)
dnodeReadMnodeInfos
(
dnode
EpSe
t
);
dnodeSaveMnodeInfos
(
dnode
EpSe
t
);
return
;
}
...
...
src/mnode/inc/mnodeMnode.h
浏览文件 @
fdfa789c
...
...
@@ -42,12 +42,12 @@ void mnodeIncMnodeRef(struct SMnodeObj *pMnode);
void
mnodeDecMnodeRef
(
struct
SMnodeObj
*
pMnode
);
char
*
mnodeGetMnodeRoleStr
();
void
mnodeGetMnode
IpSetForPeer
(
SRpcIpSet
*
i
pSet
);
void
mnodeGetMnode
IpSetForShell
(
SRpcIpSet
*
i
pSet
);
void
mnodeGetMnode
EpSetForPeer
(
SRpcEpSet
*
e
pSet
);
void
mnodeGetMnode
EpSetForShell
(
SRpcEpSet
*
e
pSet
);
char
*
mnodeGetMnodeMasterEp
();
void
mnodeGetMnodeInfos
(
void
*
mnodes
);
void
mnodeUpdateMnode
I
pSet
();
void
mnodeUpdateMnode
E
pSet
();
#ifdef __cplusplus
}
...
...
src/mnode/inc/mnodeVgroup.h
浏览文件 @
fdfa789c
...
...
@@ -44,12 +44,12 @@ int32_t mnodeGetAvailableVgroup(struct SMnodeMsg *pMsg, SVgObj **pVgroup, int32_
void
mnodeAddTableIntoVgroup
(
SVgObj
*
pVgroup
,
SChildTableObj
*
pTable
);
void
mnodeRemoveTableFromVgroup
(
SVgObj
*
pVgroup
,
SChildTableObj
*
pTable
);
void
mnodeSendDropVnodeMsg
(
int32_t
vgId
,
SRpc
IpSet
*
i
pSet
,
void
*
ahandle
);
void
mnodeSendDropVnodeMsg
(
int32_t
vgId
,
SRpc
EpSet
*
e
pSet
,
void
*
ahandle
);
void
mnodeSendCreateVgroupMsg
(
SVgObj
*
pVgroup
,
void
*
ahandle
);
void
mnodeSendAlterVgroupMsg
(
SVgObj
*
pVgroup
);
SRpc
IpSet
mnodeGetI
pSetFromVgroup
(
SVgObj
*
pVgroup
);
SRpc
IpSet
mnodeGetI
pSetFromIp
(
char
*
ep
);
SRpc
EpSet
mnodeGetE
pSetFromVgroup
(
SVgObj
*
pVgroup
);
SRpc
EpSet
mnodeGetE
pSetFromIp
(
char
*
ep
);
#ifdef __cplusplus
}
...
...
src/mnode/src/mnodeDnode.c
浏览文件 @
fdfa789c
...
...
@@ -289,14 +289,14 @@ static int32_t mnodeProcessCfgDnodeMsg(SMnodeMsg *pMsg) {
}
}
SRpc
IpSet
ipSet
=
mnodeGetI
pSetFromIp
(
pCmCfgDnode
->
ep
);
SRpc
EpSet
epSet
=
mnodeGetE
pSetFromIp
(
pCmCfgDnode
->
ep
);
if
(
dnodeId
!=
0
)
{
SDnodeObj
*
pDnode
=
mnodeGetDnode
(
dnodeId
);
if
(
pDnode
==
NULL
)
{
mError
(
"failed to cfg dnode, invalid dnodeId:%d"
,
dnodeId
);
return
TSDB_CODE_MND_DNODE_NOT_EXIST
;
}
ipSet
=
mnodeGetI
pSetFromIp
(
pDnode
->
dnodeEp
);
epSet
=
mnodeGetE
pSetFromIp
(
pDnode
->
dnodeEp
);
mnodeDecDnodeRef
(
pDnode
);
}
...
...
@@ -313,7 +313,7 @@ static int32_t mnodeProcessCfgDnodeMsg(SMnodeMsg *pMsg) {
};
mInfo
(
"dnode:%s, is configured by %s"
,
pCmCfgDnode
->
ep
,
pMsg
->
pUser
->
user
);
dnodeSendMsgToDnode
(
&
i
pSet
,
&
rpcMdCfgDnodeMsg
);
dnodeSendMsgToDnode
(
&
e
pSet
,
&
rpcMdCfgDnodeMsg
);
return
TSDB_CODE_SUCCESS
;
}
...
...
@@ -399,9 +399,9 @@ static int32_t mnodeProcessDnodeStatusMsg(SMnodeMsg *pMsg) {
SVgObj
*
pVgroup
=
mnodeGetVgroup
(
pVload
->
vgId
);
if
(
pVgroup
==
NULL
)
{
SRpc
IpSet
ipSet
=
mnodeGetI
pSetFromIp
(
pDnode
->
dnodeEp
);
SRpc
EpSet
epSet
=
mnodeGetE
pSetFromIp
(
pDnode
->
dnodeEp
);
mInfo
(
"dnode:%d, vgId:%d not exist in mnode, drop it"
,
pDnode
->
dnodeId
,
pVload
->
vgId
);
mnodeSendDropVnodeMsg
(
pVload
->
vgId
,
&
i
pSet
,
NULL
);
mnodeSendDropVnodeMsg
(
pVload
->
vgId
,
&
e
pSet
,
NULL
);
}
else
{
mnodeUpdateVgroupStatus
(
pVgroup
,
pDnode
,
pVload
);
pAccess
->
vgId
=
htonl
(
pVload
->
vgId
);
...
...
src/mnode/src/mnodeMnode.c
浏览文件 @
fdfa789c
...
...
@@ -35,8 +35,8 @@
static
void
*
tsMnodeSdb
=
NULL
;
static
int32_t
tsMnodeUpdateSize
=
0
;
static
SRpc
IpSet
tsMnodeI
pSetForShell
;
static
SRpc
IpSet
tsMnodeI
pSetForPeer
;
static
SRpc
EpSet
tsMnodeE
pSetForShell
;
static
SRpc
EpSet
tsMnodeE
pSetForPeer
;
static
SDMMnodeInfos
tsMnodeInfos
;
static
int32_t
mnodeGetMnodeMeta
(
STableMetaMsg
*
pMeta
,
SShowObj
*
pShow
,
void
*
pConn
);
static
int32_t
mnodeRetrieveMnodes
(
SShowObj
*
pShow
,
char
*
data
,
int32_t
rows
,
void
*
pConn
);
...
...
@@ -123,7 +123,7 @@ static int32_t mnodeMnodeActionRestored() {
sdbFreeIter
(
pIter
);
}
mnodeUpdateMnode
I
pSet
();
mnodeUpdateMnode
E
pSet
();
return
TSDB_CODE_SUCCESS
;
}
...
...
@@ -204,13 +204,13 @@ char *mnodeGetMnodeRoleStr(int32_t role) {
}
}
void
mnodeUpdateMnode
I
pSet
()
{
mInfo
(
"update mnodes
ipset, numOfI
ps:%d "
,
mnodeGetMnodesNum
());
void
mnodeUpdateMnode
E
pSet
()
{
mInfo
(
"update mnodes
epSet, numOfE
ps:%d "
,
mnodeGetMnodesNum
());
mnodeMnodeWrLock
();
memset
(
&
tsMnode
IpSetForShell
,
0
,
sizeof
(
SRpcI
pSet
));
memset
(
&
tsMnode
IpSetForPeer
,
0
,
sizeof
(
SRpcI
pSet
));
memset
(
&
tsMnode
EpSetForShell
,
0
,
sizeof
(
SRpcE
pSet
));
memset
(
&
tsMnode
EpSetForPeer
,
0
,
sizeof
(
SRpcE
pSet
));
memset
(
&
tsMnodeInfos
,
0
,
sizeof
(
SDMMnodeInfos
));
int32_t
index
=
0
;
...
...
@@ -222,20 +222,20 @@ void mnodeUpdateMnodeIpSet() {
SDnodeObj
*
pDnode
=
mnodeGetDnode
(
pMnode
->
mnodeId
);
if
(
pDnode
!=
NULL
)
{
strcpy
(
tsMnode
I
pSetForShell
.
fqdn
[
index
],
pDnode
->
dnodeFqdn
);
tsMnode
I
pSetForShell
.
port
[
index
]
=
htons
(
pDnode
->
dnodePort
);
mDebug
(
"mnode:%d, for shell fqdn:%s %d"
,
pDnode
->
dnodeId
,
tsMnode
IpSetForShell
.
fqdn
[
index
],
htons
(
tsMnodeI
pSetForShell
.
port
[
index
]));
strcpy
(
tsMnode
E
pSetForShell
.
fqdn
[
index
],
pDnode
->
dnodeFqdn
);
tsMnode
E
pSetForShell
.
port
[
index
]
=
htons
(
pDnode
->
dnodePort
);
mDebug
(
"mnode:%d, for shell fqdn:%s %d"
,
pDnode
->
dnodeId
,
tsMnode
EpSetForShell
.
fqdn
[
index
],
htons
(
tsMnodeE
pSetForShell
.
port
[
index
]));
strcpy
(
tsMnode
I
pSetForPeer
.
fqdn
[
index
],
pDnode
->
dnodeFqdn
);
tsMnode
I
pSetForPeer
.
port
[
index
]
=
htons
(
pDnode
->
dnodePort
+
TSDB_PORT_DNODEDNODE
);
mDebug
(
"mnode:%d, for peer fqdn:%s %d"
,
pDnode
->
dnodeId
,
tsMnode
IpSetForPeer
.
fqdn
[
index
],
htons
(
tsMnodeI
pSetForPeer
.
port
[
index
]));
strcpy
(
tsMnode
E
pSetForPeer
.
fqdn
[
index
],
pDnode
->
dnodeFqdn
);
tsMnode
E
pSetForPeer
.
port
[
index
]
=
htons
(
pDnode
->
dnodePort
+
TSDB_PORT_DNODEDNODE
);
mDebug
(
"mnode:%d, for peer fqdn:%s %d"
,
pDnode
->
dnodeId
,
tsMnode
EpSetForPeer
.
fqdn
[
index
],
htons
(
tsMnodeE
pSetForPeer
.
port
[
index
]));
tsMnodeInfos
.
nodeInfos
[
index
].
nodeId
=
htonl
(
pMnode
->
mnodeId
);
strcpy
(
tsMnodeInfos
.
nodeInfos
[
index
].
nodeEp
,
pDnode
->
dnodeEp
);
if
(
pMnode
->
role
==
TAOS_SYNC_ROLE_MASTER
)
{
tsMnode
I
pSetForShell
.
inUse
=
index
;
tsMnode
I
pSetForPeer
.
inUse
=
index
;
tsMnode
E
pSetForShell
.
inUse
=
index
;
tsMnode
E
pSetForPeer
.
inUse
=
index
;
tsMnodeInfos
.
inUse
=
index
;
}
...
...
@@ -248,23 +248,23 @@ void mnodeUpdateMnodeIpSet() {
}
tsMnodeInfos
.
nodeNum
=
index
;
tsMnode
IpSetForShell
.
numOfI
ps
=
index
;
tsMnode
IpSetForPeer
.
numOfI
ps
=
index
;
tsMnode
EpSetForShell
.
numOfE
ps
=
index
;
tsMnode
EpSetForPeer
.
numOfE
ps
=
index
;
sdbFreeIter
(
pIter
);
mnodeMnodeUnLock
();
}
void
mnodeGetMnode
IpSetForPeer
(
SRpcIpSet
*
i
pSet
)
{
void
mnodeGetMnode
EpSetForPeer
(
SRpcEpSet
*
e
pSet
)
{
mnodeMnodeRdLock
();
*
ipSet
=
tsMnodeI
pSetForPeer
;
*
epSet
=
tsMnodeE
pSetForPeer
;
mnodeMnodeUnLock
();
}
void
mnodeGetMnode
IpSetForShell
(
SRpcIpSet
*
i
pSet
)
{
void
mnodeGetMnode
EpSetForShell
(
SRpcEpSet
*
e
pSet
)
{
mnodeMnodeRdLock
();
*
ipSet
=
tsMnodeI
pSetForShell
;
*
epSet
=
tsMnodeE
pSetForShell
;
mnodeMnodeUnLock
();
}
...
...
@@ -295,7 +295,7 @@ int32_t mnodeAddMnode(int32_t dnodeId) {
code
=
TSDB_CODE_MND_SDB_ERROR
;
}
mnodeUpdateMnode
I
pSet
();
mnodeUpdateMnode
E
pSet
();
return
code
;
}
...
...
@@ -308,7 +308,7 @@ void mnodeDropMnodeLocal(int32_t dnodeId) {
mnodeDecMnodeRef
(
pMnode
);
}
mnodeUpdateMnode
I
pSet
();
mnodeUpdateMnode
E
pSet
();
}
int32_t
mnodeDropMnode
(
int32_t
dnodeId
)
{
...
...
@@ -330,7 +330,7 @@ int32_t mnodeDropMnode(int32_t dnodeId) {
sdbDecRef
(
tsMnodeSdb
,
pMnode
);
mnodeUpdateMnode
I
pSet
();
mnodeUpdateMnode
E
pSet
();
return
code
;
}
...
...
src/mnode/src/mnodePeer.c
浏览文件 @
fdfa789c
...
...
@@ -53,14 +53,14 @@ int32_t mnodeProcessPeerReq(SMnodeMsg *pMsg) {
if
(
!
sdbIsMaster
())
{
SMnodeRsp
*
rpcRsp
=
&
pMsg
->
rpcRsp
;
SRpc
IpSet
*
ipSet
=
rpcMallocCont
(
sizeof
(
SRpcI
pSet
));
mnodeGetMnode
IpSetForPeer
(
i
pSet
);
rpcRsp
->
rsp
=
i
pSet
;
rpcRsp
->
len
=
sizeof
(
SRpc
I
pSet
);
SRpc
EpSet
*
epSet
=
rpcMallocCont
(
sizeof
(
SRpcE
pSet
));
mnodeGetMnode
EpSetForPeer
(
e
pSet
);
rpcRsp
->
rsp
=
e
pSet
;
rpcRsp
->
len
=
sizeof
(
SRpc
E
pSet
);
mDebug
(
"%p, msg:%s in mpeer queue, will be redireced inUse:%d"
,
pMsg
->
rpcMsg
.
ahandle
,
taosMsg
[
pMsg
->
rpcMsg
.
msgType
],
i
pSet
->
inUse
);
for
(
int32_t
i
=
0
;
i
<
ipSet
->
numOfI
ps
;
++
i
)
{
mDebug
(
"mnode index:%d
ip:%s:%d"
,
i
,
ipSet
->
fqdn
[
i
],
htons
(
i
pSet
->
port
[
i
]));
mDebug
(
"%p, msg:%s in mpeer queue, will be redireced inUse:%d"
,
pMsg
->
rpcMsg
.
ahandle
,
taosMsg
[
pMsg
->
rpcMsg
.
msgType
],
e
pSet
->
inUse
);
for
(
int32_t
i
=
0
;
i
<
epSet
->
numOfE
ps
;
++
i
)
{
mDebug
(
"mnode index:%d
ep:%s:%d"
,
i
,
epSet
->
fqdn
[
i
],
htons
(
e
pSet
->
port
[
i
]));
}
return
TSDB_CODE_RPC_REDIRECT
;
...
...
src/mnode/src/mnodeRead.c
浏览文件 @
fdfa789c
...
...
@@ -49,14 +49,14 @@ int32_t mnodeProcessRead(SMnodeMsg *pMsg) {
if
(
!
sdbIsMaster
())
{
SMnodeRsp
*
rpcRsp
=
&
pMsg
->
rpcRsp
;
SRpc
IpSet
*
ipSet
=
rpcMallocCont
(
sizeof
(
SRpcI
pSet
));
mnodeGetMnode
IpSetForShell
(
i
pSet
);
rpcRsp
->
rsp
=
i
pSet
;
rpcRsp
->
len
=
sizeof
(
SRpc
I
pSet
);
SRpc
EpSet
*
epSet
=
rpcMallocCont
(
sizeof
(
SRpcE
pSet
));
mnodeGetMnode
EpSetForShell
(
e
pSet
);
rpcRsp
->
rsp
=
e
pSet
;
rpcRsp
->
len
=
sizeof
(
SRpc
E
pSet
);
mDebug
(
"%p, msg:%s in mread queue, will be redireced, inUse:%d"
,
pMsg
->
rpcMsg
.
ahandle
,
taosMsg
[
pMsg
->
rpcMsg
.
msgType
],
i
pSet
->
inUse
);
for
(
int32_t
i
=
0
;
i
<
ipSet
->
numOfI
ps
;
++
i
)
{
mDebug
(
"mnode index:%d
ip:%s:%d"
,
i
,
ipSet
->
fqdn
[
i
],
htons
(
i
pSet
->
port
[
i
]));
mDebug
(
"%p, msg:%s in mread queue, will be redireced, inUse:%d"
,
pMsg
->
rpcMsg
.
ahandle
,
taosMsg
[
pMsg
->
rpcMsg
.
msgType
],
e
pSet
->
inUse
);
for
(
int32_t
i
=
0
;
i
<
epSet
->
numOfE
ps
;
++
i
)
{
mDebug
(
"mnode index:%d
ep:%s:%d"
,
i
,
epSet
->
fqdn
[
i
],
htons
(
e
pSet
->
port
[
i
]));
}
return
TSDB_CODE_RPC_REDIRECT
;
...
...
src/mnode/src/mnodeSdb.c
浏览文件 @
fdfa789c
...
...
@@ -219,7 +219,7 @@ void sdbUpdateMnodeRoles() {
}
}
mnodeUpdateMnode
I
pSet
();
mnodeUpdateMnode
E
pSet
();
}
static
uint32_t
sdbGetFileInfo
(
void
*
ahandle
,
char
*
name
,
uint32_t
*
index
,
uint32_t
eindex
,
int32_t
*
size
,
uint64_t
*
fversion
)
{
...
...
src/mnode/src/mnodeShow.c
浏览文件 @
fdfa789c
...
...
@@ -270,7 +270,7 @@ static int32_t mnodeProcessHeartBeatMsg(SMnodeMsg *pMsg) {
pHBRsp
->
onlineDnodes
=
htonl
(
mnodeGetOnlinDnodesNum
());
pHBRsp
->
totalDnodes
=
htonl
(
mnodeGetDnodesNum
());
mnodeGetMnode
IpSetForShell
(
&
pHBRsp
->
ipLis
t
);
mnodeGetMnode
EpSetForShell
(
&
pHBRsp
->
epSe
t
);
pMsg
->
rpcRsp
.
rsp
=
pHBRsp
;
pMsg
->
rpcRsp
.
len
=
sizeof
(
SCMHeartBeatRsp
);
...
...
@@ -335,7 +335,7 @@ static int32_t mnodeProcessConnectMsg(SMnodeMsg *pMsg) {
pConnectRsp
->
writeAuth
=
pUser
->
writeAuth
;
pConnectRsp
->
superAuth
=
pUser
->
superAuth
;
mnodeGetMnode
IpSetForShell
(
&
pConnectRsp
->
ipLis
t
);
mnodeGetMnode
EpSetForShell
(
&
pConnectRsp
->
epSe
t
);
connect_over:
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
...
...
src/mnode/src/mnodeTable.c
浏览文件 @
fdfa789c
...
...
@@ -910,9 +910,9 @@ static int32_t mnodeProcessDropSuperTableMsg(SMnodeMsg *pMsg) {
mInfo
(
"app:%p:%p, stable:%s, send drop stable msg to vgId:%d"
,
pMsg
->
rpcMsg
.
ahandle
,
pMsg
,
pStable
->
info
.
tableId
,
pVgroup
->
vgId
);
SRpc
IpSet
ipSet
=
mnodeGetI
pSetFromVgroup
(
pVgroup
);
SRpc
EpSet
epSet
=
mnodeGetE
pSetFromVgroup
(
pVgroup
);
SRpcMsg
rpcMsg
=
{.
pCont
=
pDrop
,
.
contLen
=
sizeof
(
SMDDropSTableMsg
),
.
msgType
=
TSDB_MSG_TYPE_MD_DROP_STABLE
};
dnodeSendMsgToDnode
(
&
i
pSet
,
&
rpcMsg
);
dnodeSendMsgToDnode
(
&
e
pSet
,
&
rpcMsg
);
mnodeDecVgroupRef
(
pVgroup
);
}
taosHashDestroyIter
(
pIter
);
...
...
@@ -1484,10 +1484,10 @@ static int32_t mnodeProcessSuperTableVgroupMsg(SMnodeMsg *pMsg) {
SDnodeObj
*
pDnode
=
pVgroup
->
vnodeGid
[
vn
].
pDnode
;
if
(
pDnode
==
NULL
)
break
;
tstrncpy
(
pVgroupInfo
->
vgroups
[
vgSize
].
i
pAddr
[
vn
].
fqdn
,
pDnode
->
dnodeFqdn
,
TSDB_FQDN_LEN
);
pVgroupInfo
->
vgroups
[
vgSize
].
i
pAddr
[
vn
].
port
=
htons
(
pDnode
->
dnodePort
);
tstrncpy
(
pVgroupInfo
->
vgroups
[
vgSize
].
e
pAddr
[
vn
].
fqdn
,
pDnode
->
dnodeFqdn
,
TSDB_FQDN_LEN
);
pVgroupInfo
->
vgroups
[
vgSize
].
e
pAddr
[
vn
].
port
=
htons
(
pDnode
->
dnodePort
);
pVgroupInfo
->
vgroups
[
vgSize
].
numOf
I
ps
++
;
pVgroupInfo
->
vgroups
[
vgSize
].
numOf
E
ps
++
;
}
vgSize
++
;
...
...
@@ -1615,7 +1615,7 @@ static int32_t mnodeDoCreateChildTableCb(SMnodeMsg *pMsg, int32_t code) {
return
terrno
;
}
SRpc
IpSet
ipSet
=
mnodeGetI
pSetFromVgroup
(
pMsg
->
pVgroup
);
SRpc
EpSet
epSet
=
mnodeGetE
pSetFromVgroup
(
pMsg
->
pVgroup
);
SRpcMsg
rpcMsg
=
{
.
ahandle
=
pMsg
,
.
pCont
=
pMDCreate
,
...
...
@@ -1624,7 +1624,7 @@ static int32_t mnodeDoCreateChildTableCb(SMnodeMsg *pMsg, int32_t code) {
.
msgType
=
TSDB_MSG_TYPE_MD_CREATE_TABLE
};
dnodeSendMsgToDnode
(
&
i
pSet
,
&
rpcMsg
);
dnodeSendMsgToDnode
(
&
e
pSet
,
&
rpcMsg
);
return
TSDB_CODE_MND_ACTION_IN_PROGRESS
;
}
...
...
@@ -1788,7 +1788,7 @@ static int32_t mnodeProcessDropChildTableMsg(SMnodeMsg *pMsg, bool needReturn) {
pDrop
->
sid
=
htonl
(
pTable
->
sid
);
pDrop
->
uid
=
htobe64
(
pTable
->
uid
);
SRpc
IpSet
ipSet
=
mnodeGetI
pSetFromVgroup
(
pMsg
->
pVgroup
);
SRpc
EpSet
epSet
=
mnodeGetE
pSetFromVgroup
(
pMsg
->
pVgroup
);
mInfo
(
"app:%p:%p, table:%s, send drop ctable msg, vgId:%d sid:%d uid:%"
PRIu64
,
pMsg
->
rpcMsg
.
ahandle
,
pMsg
,
pDrop
->
tableId
,
pTable
->
vgId
,
pTable
->
sid
,
pTable
->
uid
);
...
...
@@ -1803,7 +1803,7 @@ static int32_t mnodeProcessDropChildTableMsg(SMnodeMsg *pMsg, bool needReturn) {
if
(
!
needReturn
)
rpcMsg
.
ahandle
=
NULL
;
dnodeSendMsgToDnode
(
&
i
pSet
,
&
rpcMsg
);
dnodeSendMsgToDnode
(
&
e
pSet
,
&
rpcMsg
);
return
TSDB_CODE_MND_ACTION_IN_PROGRESS
;
}
...
...
@@ -1842,7 +1842,7 @@ static int32_t mnodeAlterNormalTableColumnCb(SMnodeMsg *pMsg, int32_t code) {
}
}
SRpc
IpSet
ipSet
=
mnodeGetI
pSetFromVgroup
(
pMsg
->
pVgroup
);
SRpc
EpSet
epSet
=
mnodeGetE
pSetFromVgroup
(
pMsg
->
pVgroup
);
SRpcMsg
rpcMsg
=
{
.
ahandle
=
pMsg
,
.
pCont
=
pMDCreate
,
...
...
@@ -1854,7 +1854,7 @@ static int32_t mnodeAlterNormalTableColumnCb(SMnodeMsg *pMsg, int32_t code) {
mDebug
(
"app:%p:%p, ctable %s, send alter column msg to vgId:%d"
,
pMsg
->
rpcMsg
.
ahandle
,
pMsg
,
pTable
->
info
.
tableId
,
pMsg
->
pVgroup
->
vgId
);
dnodeSendMsgToDnode
(
&
i
pSet
,
&
rpcMsg
);
dnodeSendMsgToDnode
(
&
e
pSet
,
&
rpcMsg
);
return
TSDB_CODE_MND_ACTION_IN_PROGRESS
;
}
...
...
@@ -1996,9 +1996,9 @@ static int32_t mnodeDoGetChildTableMeta(SMnodeMsg *pMsg, STableMetaMsg *pMeta) {
for
(
int32_t
i
=
0
;
i
<
pMsg
->
pVgroup
->
numOfVnodes
;
++
i
)
{
SDnodeObj
*
pDnode
=
mnodeGetDnode
(
pMsg
->
pVgroup
->
vnodeGid
[
i
].
dnodeId
);
if
(
pDnode
==
NULL
)
break
;
strcpy
(
pMeta
->
vgroup
.
i
pAddr
[
i
].
fqdn
,
pDnode
->
dnodeFqdn
);
pMeta
->
vgroup
.
i
pAddr
[
i
].
port
=
htons
(
pDnode
->
dnodePort
+
TSDB_PORT_DNODESHELL
);
pMeta
->
vgroup
.
numOf
I
ps
++
;
strcpy
(
pMeta
->
vgroup
.
e
pAddr
[
i
].
fqdn
,
pDnode
->
dnodeFqdn
);
pMeta
->
vgroup
.
e
pAddr
[
i
].
port
=
htons
(
pDnode
->
dnodePort
+
TSDB_PORT_DNODESHELL
);
pMeta
->
vgroup
.
numOf
E
ps
++
;
mnodeDecDnodeRef
(
pDnode
);
}
pMeta
->
vgroup
.
vgId
=
htonl
(
pMsg
->
pVgroup
->
vgId
);
...
...
src/mnode/src/mnodeVgroup.c
浏览文件 @
fdfa789c
...
...
@@ -317,9 +317,9 @@ void mnodeUpdateVgroupStatus(SVgObj *pVgroup, SDnodeObj *pDnode, SVnodeLoad *pVl
}
if
(
!
dnodeExist
)
{
SRpc
IpSet
ipSet
=
mnodeGetI
pSetFromIp
(
pDnode
->
dnodeEp
);
SRpc
EpSet
epSet
=
mnodeGetE
pSetFromIp
(
pDnode
->
dnodeEp
);
mError
(
"vgId:%d, dnode:%d not exist in mnode, drop it"
,
pVload
->
vgId
,
pDnode
->
dnodeId
);
mnodeSendDropVnodeMsg
(
pVload
->
vgId
,
&
i
pSet
,
NULL
);
mnodeSendDropVnodeMsg
(
pVload
->
vgId
,
&
e
pSet
,
NULL
);
return
;
}
...
...
@@ -809,29 +809,29 @@ static SMDCreateVnodeMsg *mnodeBuildVnodeMsg(SVgObj *pVgroup) {
return
pVnode
;
}
SRpc
IpSet
mnodeGetI
pSetFromVgroup
(
SVgObj
*
pVgroup
)
{
SRpc
IpSet
i
pSet
=
{
.
numOf
I
ps
=
pVgroup
->
numOfVnodes
,
SRpc
EpSet
mnodeGetE
pSetFromVgroup
(
SVgObj
*
pVgroup
)
{
SRpc
EpSet
e
pSet
=
{
.
numOf
E
ps
=
pVgroup
->
numOfVnodes
,
.
inUse
=
0
,
};
for
(
int
i
=
0
;
i
<
pVgroup
->
numOfVnodes
;
++
i
)
{
strcpy
(
i
pSet
.
fqdn
[
i
],
pVgroup
->
vnodeGid
[
i
].
pDnode
->
dnodeFqdn
);
i
pSet
.
port
[
i
]
=
pVgroup
->
vnodeGid
[
i
].
pDnode
->
dnodePort
+
TSDB_PORT_DNODEDNODE
;
strcpy
(
e
pSet
.
fqdn
[
i
],
pVgroup
->
vnodeGid
[
i
].
pDnode
->
dnodeFqdn
);
e
pSet
.
port
[
i
]
=
pVgroup
->
vnodeGid
[
i
].
pDnode
->
dnodePort
+
TSDB_PORT_DNODEDNODE
;
}
return
i
pSet
;
return
e
pSet
;
}
SRpc
IpSet
mnodeGetI
pSetFromIp
(
char
*
ep
)
{
SRpc
IpSet
i
pSet
;
SRpc
EpSet
mnodeGetE
pSetFromIp
(
char
*
ep
)
{
SRpc
EpSet
e
pSet
;
ipSet
.
numOfI
ps
=
1
;
i
pSet
.
inUse
=
0
;
taosGetFqdnPortFromEp
(
ep
,
ipSet
.
fqdn
[
0
],
&
i
pSet
.
port
[
0
]);
i
pSet
.
port
[
0
]
+=
TSDB_PORT_DNODEDNODE
;
return
i
pSet
;
epSet
.
numOfE
ps
=
1
;
e
pSet
.
inUse
=
0
;
taosGetFqdnPortFromEp
(
ep
,
epSet
.
fqdn
[
0
],
&
e
pSet
.
port
[
0
]);
e
pSet
.
port
[
0
]
+=
TSDB_PORT_DNODEDNODE
;
return
e
pSet
;
}
static
void
mnodeSendAlterVnodeMsg
(
SVgObj
*
pVgroup
,
SRpc
IpSet
*
i
pSet
)
{
static
void
mnodeSendAlterVnodeMsg
(
SVgObj
*
pVgroup
,
SRpc
EpSet
*
e
pSet
)
{
SMDAlterVnodeMsg
*
pAlter
=
mnodeBuildVnodeMsg
(
pVgroup
);
SRpcMsg
rpcMsg
=
{
.
ahandle
=
NULL
,
...
...
@@ -840,21 +840,21 @@ static void mnodeSendAlterVnodeMsg(SVgObj *pVgroup, SRpcIpSet *ipSet) {
.
code
=
0
,
.
msgType
=
TSDB_MSG_TYPE_MD_ALTER_VNODE
};
dnodeSendMsgToDnode
(
i
pSet
,
&
rpcMsg
);
dnodeSendMsgToDnode
(
e
pSet
,
&
rpcMsg
);
}
void
mnodeSendAlterVgroupMsg
(
SVgObj
*
pVgroup
)
{
mDebug
(
"vgId:%d, send alter all vnodes msg, numOfVnodes:%d db:%s"
,
pVgroup
->
vgId
,
pVgroup
->
numOfVnodes
,
pVgroup
->
dbName
);
for
(
int32_t
i
=
0
;
i
<
pVgroup
->
numOfVnodes
;
++
i
)
{
SRpc
IpSet
ipSet
=
mnodeGetI
pSetFromIp
(
pVgroup
->
vnodeGid
[
i
].
pDnode
->
dnodeEp
);
SRpc
EpSet
epSet
=
mnodeGetE
pSetFromIp
(
pVgroup
->
vnodeGid
[
i
].
pDnode
->
dnodeEp
);
mDebug
(
"vgId:%d, index:%d, send alter vnode msg to dnode %s"
,
pVgroup
->
vgId
,
i
,
pVgroup
->
vnodeGid
[
i
].
pDnode
->
dnodeEp
);
mnodeSendAlterVnodeMsg
(
pVgroup
,
&
i
pSet
);
mnodeSendAlterVnodeMsg
(
pVgroup
,
&
e
pSet
);
}
}
static
void
mnodeSendCreateVnodeMsg
(
SVgObj
*
pVgroup
,
SRpc
IpSet
*
i
pSet
,
void
*
ahandle
)
{
static
void
mnodeSendCreateVnodeMsg
(
SVgObj
*
pVgroup
,
SRpc
EpSet
*
e
pSet
,
void
*
ahandle
)
{
SMDCreateVnodeMsg
*
pCreate
=
mnodeBuildVnodeMsg
(
pVgroup
);
SRpcMsg
rpcMsg
=
{
.
ahandle
=
ahandle
,
...
...
@@ -863,17 +863,17 @@ static void mnodeSendCreateVnodeMsg(SVgObj *pVgroup, SRpcIpSet *ipSet, void *aha
.
code
=
0
,
.
msgType
=
TSDB_MSG_TYPE_MD_CREATE_VNODE
};
dnodeSendMsgToDnode
(
i
pSet
,
&
rpcMsg
);
dnodeSendMsgToDnode
(
e
pSet
,
&
rpcMsg
);
}
void
mnodeSendCreateVgroupMsg
(
SVgObj
*
pVgroup
,
void
*
ahandle
)
{
mDebug
(
"vgId:%d, send create all vnodes msg, numOfVnodes:%d db:%s"
,
pVgroup
->
vgId
,
pVgroup
->
numOfVnodes
,
pVgroup
->
dbName
);
for
(
int32_t
i
=
0
;
i
<
pVgroup
->
numOfVnodes
;
++
i
)
{
SRpc
IpSet
ipSet
=
mnodeGetI
pSetFromIp
(
pVgroup
->
vnodeGid
[
i
].
pDnode
->
dnodeEp
);
SRpc
EpSet
epSet
=
mnodeGetE
pSetFromIp
(
pVgroup
->
vnodeGid
[
i
].
pDnode
->
dnodeEp
);
mDebug
(
"vgId:%d, index:%d, send create vnode msg to dnode %s, ahandle:%p"
,
pVgroup
->
vgId
,
i
,
pVgroup
->
vnodeGid
[
i
].
pDnode
->
dnodeEp
,
ahandle
);
mnodeSendCreateVnodeMsg
(
pVgroup
,
&
i
pSet
,
ahandle
);
mnodeSendCreateVnodeMsg
(
pVgroup
,
&
e
pSet
,
ahandle
);
}
}
...
...
@@ -926,7 +926,7 @@ static SMDDropVnodeMsg *mnodeBuildDropVnodeMsg(int32_t vgId) {
return
pDrop
;
}
void
mnodeSendDropVnodeMsg
(
int32_t
vgId
,
SRpc
IpSet
*
i
pSet
,
void
*
ahandle
)
{
void
mnodeSendDropVnodeMsg
(
int32_t
vgId
,
SRpc
EpSet
*
e
pSet
,
void
*
ahandle
)
{
SMDDropVnodeMsg
*
pDrop
=
mnodeBuildDropVnodeMsg
(
vgId
);
SRpcMsg
rpcMsg
=
{
.
ahandle
=
ahandle
,
...
...
@@ -935,16 +935,16 @@ void mnodeSendDropVnodeMsg(int32_t vgId, SRpcIpSet *ipSet, void *ahandle) {
.
code
=
0
,
.
msgType
=
TSDB_MSG_TYPE_MD_DROP_VNODE
};
dnodeSendMsgToDnode
(
i
pSet
,
&
rpcMsg
);
dnodeSendMsgToDnode
(
e
pSet
,
&
rpcMsg
);
}
static
void
mnodeSendDropVgroupMsg
(
SVgObj
*
pVgroup
,
void
*
ahandle
)
{
pVgroup
->
status
=
TAOS_VG_STATUS_DROPPING
;
// deleting
mDebug
(
"vgId:%d, send drop all vnodes msg, ahandle:%p"
,
pVgroup
->
vgId
,
ahandle
);
for
(
int32_t
i
=
0
;
i
<
pVgroup
->
numOfVnodes
;
++
i
)
{
SRpc
IpSet
ipSet
=
mnodeGetI
pSetFromIp
(
pVgroup
->
vnodeGid
[
i
].
pDnode
->
dnodeEp
);
SRpc
EpSet
epSet
=
mnodeGetE
pSetFromIp
(
pVgroup
->
vnodeGid
[
i
].
pDnode
->
dnodeEp
);
mDebug
(
"vgId:%d, send drop vnode msg to dnode:%d, ahandle:%p"
,
pVgroup
->
vgId
,
pVgroup
->
vnodeGid
[
i
].
dnodeId
,
ahandle
);
mnodeSendDropVnodeMsg
(
pVgroup
->
vgId
,
&
i
pSet
,
ahandle
);
mnodeSendDropVnodeMsg
(
pVgroup
->
vgId
,
&
e
pSet
,
ahandle
);
}
}
...
...
@@ -998,8 +998,8 @@ static int32_t mnodeProcessVnodeCfgMsg(SMnodeMsg *pMsg) {
}
mDebug
(
"vgId:%d, send create vnode msg to dnode %s for vnode cfg msg"
,
pVgroup
->
vgId
,
pDnode
->
dnodeEp
);
SRpc
IpSet
ipSet
=
mnodeGetI
pSetFromIp
(
pDnode
->
dnodeEp
);
mnodeSendCreateVnodeMsg
(
pVgroup
,
&
i
pSet
,
NULL
);
SRpc
EpSet
epSet
=
mnodeGetE
pSetFromIp
(
pDnode
->
dnodeEp
);
mnodeSendCreateVnodeMsg
(
pVgroup
,
&
e
pSet
,
NULL
);
mnodeDecDnodeRef
(
pDnode
);
mnodeDecVgroupRef
(
pVgroup
);
...
...
src/mnode/src/mnodeWrite.c
浏览文件 @
fdfa789c
...
...
@@ -49,16 +49,16 @@ int32_t mnodeProcessWrite(SMnodeMsg *pMsg) {
if
(
!
sdbIsMaster
())
{
SMnodeRsp
*
rpcRsp
=
&
pMsg
->
rpcRsp
;
SRpc
IpSet
*
ipSet
=
rpcMallocCont
(
sizeof
(
SRpcI
pSet
));
mnodeGetMnode
IpSetForShell
(
i
pSet
);
rpcRsp
->
rsp
=
i
pSet
;
rpcRsp
->
len
=
sizeof
(
SRpc
I
pSet
);
SRpc
EpSet
*
epSet
=
rpcMallocCont
(
sizeof
(
SRpcE
pSet
));
mnodeGetMnode
EpSetForShell
(
e
pSet
);
rpcRsp
->
rsp
=
e
pSet
;
rpcRsp
->
len
=
sizeof
(
SRpc
E
pSet
);
mDebug
(
"app:%p:%p, msg:%s will be redireced inUse:%d"
,
pMsg
->
rpcMsg
.
ahandle
,
pMsg
,
taosMsg
[
pMsg
->
rpcMsg
.
msgType
],
i
pSet
->
inUse
);
for
(
int32_t
i
=
0
;
i
<
ipSet
->
numOfI
ps
;
++
i
)
{
mDebug
(
"app:%p:%p, mnode index:%d
ip:%s:%d"
,
pMsg
->
rpcMsg
.
ahandle
,
pMsg
,
i
,
i
pSet
->
fqdn
[
i
],
htons
(
i
pSet
->
port
[
i
]));
e
pSet
->
inUse
);
for
(
int32_t
i
=
0
;
i
<
epSet
->
numOfE
ps
;
++
i
)
{
mDebug
(
"app:%p:%p, mnode index:%d
ep:%s:%d"
,
pMsg
->
rpcMsg
.
ahandle
,
pMsg
,
i
,
e
pSet
->
fqdn
[
i
],
htons
(
e
pSet
->
port
[
i
]));
}
return
TSDB_CODE_RPC_REDIRECT
;
...
...
src/rpc/src/rpcMain.c
浏览文件 @
fdfa789c
...
...
@@ -55,7 +55,7 @@ typedef struct {
char
secret
[
TSDB_KEY_LEN
];
// secret for the link
char
ckey
[
TSDB_KEY_LEN
];
// ciphering key
void
(
*
cfp
)(
SRpcMsg
*
,
SRpc
I
pSet
*
);
void
(
*
cfp
)(
SRpcMsg
*
,
SRpc
E
pSet
*
);
int
(
*
afp
)(
char
*
user
,
char
*
spi
,
char
*
encrypt
,
char
*
secret
,
char
*
ckey
);
int32_t
refCount
;
...
...
@@ -71,7 +71,7 @@ typedef struct {
typedef
struct
{
SRpcInfo
*
pRpc
;
// associated SRpcInfo
SRpc
IpSet
i
pSet
;
// ip list provided by app
SRpc
EpSet
e
pSet
;
// ip list provided by app
void
*
ahandle
;
// handle provided by app
void
*
signature
;
// for validation
struct
SRpcConn
*
pConn
;
// pConn allocated
...
...
@@ -80,12 +80,12 @@ typedef struct {
int32_t
contLen
;
// content length
int32_t
code
;
// error code
int16_t
numOfTry
;
// number of try for different servers
int8_t
oldInUse
;
// server
I
P inUse passed by app
int8_t
oldInUse
;
// server
E
P inUse passed by app
int8_t
redirect
;
// flag to indicate redirect
int8_t
connType
;
// connection type
SRpcMsg
*
pRsp
;
// for synchronous API
tsem_t
*
pSem
;
// for synchronous API
SRpc
I
pSet
*
pSet
;
// for synchronous API
SRpc
E
pSet
*
pSet
;
// for synchronous API
char
msg
[
0
];
// RpcHead starts from here
}
SRpcReqContext
;
...
...
@@ -355,7 +355,7 @@ void *rpcReallocCont(void *ptr, int contLen) {
return
start
+
sizeof
(
SRpcReqContext
)
+
sizeof
(
SRpcHead
);
}
void
rpcSendRequest
(
void
*
shandle
,
const
SRpc
IpSet
*
pI
pSet
,
SRpcMsg
*
pMsg
)
{
void
rpcSendRequest
(
void
*
shandle
,
const
SRpc
EpSet
*
pE
pSet
,
SRpcMsg
*
pMsg
)
{
SRpcInfo
*
pRpc
=
(
SRpcInfo
*
)
shandle
;
SRpcReqContext
*
pContext
;
...
...
@@ -364,11 +364,11 @@ void rpcSendRequest(void *shandle, const SRpcIpSet *pIpSet, SRpcMsg *pMsg) {
pContext
->
ahandle
=
pMsg
->
ahandle
;
pContext
->
signature
=
pContext
;
pContext
->
pRpc
=
(
SRpcInfo
*
)
shandle
;
pContext
->
ipSet
=
*
pI
pSet
;
pContext
->
epSet
=
*
pE
pSet
;
pContext
->
contLen
=
contLen
;
pContext
->
pCont
=
pMsg
->
pCont
;
pContext
->
msgType
=
pMsg
->
msgType
;
pContext
->
oldInUse
=
p
I
pSet
->
inUse
;
pContext
->
oldInUse
=
p
E
pSet
->
inUse
;
pContext
->
connType
=
RPC_CONN_UDPC
;
if
(
contLen
>
tsRpcMaxUdpSize
)
pContext
->
connType
=
RPC_CONN_TCPC
;
...
...
@@ -458,15 +458,15 @@ void rpcSendResponse(const SRpcMsg *pRsp) {
return
;
}
void
rpcSendRedirectRsp
(
void
*
thandle
,
const
SRpc
IpSet
*
pI
pSet
)
{
void
rpcSendRedirectRsp
(
void
*
thandle
,
const
SRpc
EpSet
*
pE
pSet
)
{
SRpcMsg
rpcMsg
;
memset
(
&
rpcMsg
,
0
,
sizeof
(
rpcMsg
));
rpcMsg
.
contLen
=
sizeof
(
SRpc
I
pSet
);
rpcMsg
.
contLen
=
sizeof
(
SRpc
E
pSet
);
rpcMsg
.
pCont
=
rpcMallocCont
(
rpcMsg
.
contLen
);
if
(
rpcMsg
.
pCont
==
NULL
)
return
;
memcpy
(
rpcMsg
.
pCont
,
p
IpSet
,
sizeof
(
SRpcI
pSet
));
memcpy
(
rpcMsg
.
pCont
,
p
EpSet
,
sizeof
(
SRpcE
pSet
));
rpcMsg
.
code
=
TSDB_CODE_RPC_REDIRECT
;
rpcMsg
.
handle
=
thandle
;
...
...
@@ -488,7 +488,7 @@ int rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo) {
return
0
;
}
void
rpcSendRecv
(
void
*
shandle
,
SRpc
IpSet
*
pI
pSet
,
SRpcMsg
*
pMsg
,
SRpcMsg
*
pRsp
)
{
void
rpcSendRecv
(
void
*
shandle
,
SRpc
EpSet
*
pE
pSet
,
SRpcMsg
*
pMsg
,
SRpcMsg
*
pRsp
)
{
SRpcReqContext
*
pContext
;
pContext
=
(
SRpcReqContext
*
)
(
pMsg
->
pCont
-
sizeof
(
SRpcHead
)
-
sizeof
(
SRpcReqContext
));
...
...
@@ -498,9 +498,9 @@ void rpcSendRecv(void *shandle, SRpcIpSet *pIpSet, SRpcMsg *pMsg, SRpcMsg *pRsp)
tsem_init
(
&
sem
,
0
,
0
);
pContext
->
pSem
=
&
sem
;
pContext
->
pRsp
=
pRsp
;
pContext
->
pSet
=
p
I
pSet
;
pContext
->
pSet
=
p
E
pSet
;
rpcSendRequest
(
shandle
,
p
I
pSet
,
pMsg
);
rpcSendRequest
(
shandle
,
p
E
pSet
,
pMsg
);
tsem_wait
(
&
sem
);
tsem_destroy
(
&
sem
);
...
...
@@ -755,11 +755,11 @@ static SRpcConn *rpcGetConnObj(SRpcInfo *pRpc, int sid, SRecvInfo *pRecv) {
static
SRpcConn
*
rpcSetupConnToServer
(
SRpcReqContext
*
pContext
)
{
SRpcConn
*
pConn
;
SRpcInfo
*
pRpc
=
pContext
->
pRpc
;
SRpc
IpSet
*
pIpSet
=
&
pContext
->
i
pSet
;
SRpc
EpSet
*
pEpSet
=
&
pContext
->
e
pSet
;
pConn
=
rpcGetConnFromCache
(
pRpc
->
pCache
,
p
IpSet
->
fqdn
[
pIpSet
->
inUse
],
pIpSet
->
port
[
pI
pSet
->
inUse
],
pContext
->
connType
);
pConn
=
rpcGetConnFromCache
(
pRpc
->
pCache
,
p
EpSet
->
fqdn
[
pEpSet
->
inUse
],
pEpSet
->
port
[
pE
pSet
->
inUse
],
pContext
->
connType
);
if
(
pConn
==
NULL
||
pConn
->
user
[
0
]
==
0
)
{
pConn
=
rpcOpenConn
(
pRpc
,
p
IpSet
->
fqdn
[
pIpSet
->
inUse
],
pIpSet
->
port
[
pI
pSet
->
inUse
],
pContext
->
connType
);
pConn
=
rpcOpenConn
(
pRpc
,
p
EpSet
->
fqdn
[
pEpSet
->
inUse
],
pEpSet
->
port
[
pE
pSet
->
inUse
],
pContext
->
connType
);
}
if
(
pConn
)
{
...
...
@@ -1020,16 +1020,16 @@ static void rpcNotifyClient(SRpcReqContext *pContext, SRpcMsg *pMsg) {
pContext
->
pConn
=
NULL
;
if
(
pContext
->
pRsp
)
{
// for synchronous API
memcpy
(
pContext
->
pSet
,
&
pContext
->
ipSet
,
sizeof
(
SRpcI
pSet
));
memcpy
(
pContext
->
pSet
,
&
pContext
->
epSet
,
sizeof
(
SRpcE
pSet
));
memcpy
(
pContext
->
pRsp
,
pMsg
,
sizeof
(
SRpcMsg
));
tsem_post
(
pContext
->
pSem
);
}
else
{
// for asynchronous API
SRpc
IpSet
*
pI
pSet
=
NULL
;
if
(
pContext
->
i
pSet
.
inUse
!=
pContext
->
oldInUse
||
pContext
->
redirect
)
p
IpSet
=
&
pContext
->
i
pSet
;
SRpc
EpSet
*
pE
pSet
=
NULL
;
if
(
pContext
->
e
pSet
.
inUse
!=
pContext
->
oldInUse
||
pContext
->
redirect
)
p
EpSet
=
&
pContext
->
e
pSet
;
(
*
pRpc
->
cfp
)(
pMsg
,
p
I
pSet
);
(
*
pRpc
->
cfp
)(
pMsg
,
p
E
pSet
);
}
// free the request message
...
...
@@ -1070,9 +1070,9 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead) {
pConn
->
pContext
=
NULL
;
pConn
->
pReqMsg
=
NULL
;
// for UDP, port may be changed by server, the port in
i
pSet shall be used for cache
// for UDP, port may be changed by server, the port in
e
pSet shall be used for cache
if
(
pHead
->
code
!=
TSDB_CODE_RPC_TOO_SLOW
)
{
rpcAddConnIntoCache
(
pRpc
->
pCache
,
pConn
,
pConn
->
peerFqdn
,
pContext
->
ipSet
.
port
[
pContext
->
i
pSet
.
inUse
],
pConn
->
connType
);
rpcAddConnIntoCache
(
pRpc
->
pCache
,
pConn
,
pConn
->
peerFqdn
,
pContext
->
epSet
.
port
[
pContext
->
e
pSet
.
inUse
],
pConn
->
connType
);
}
else
{
rpcCloseConn
(
pConn
);
}
...
...
@@ -1087,10 +1087,10 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead) {
if
(
pHead
->
code
==
TSDB_CODE_RPC_REDIRECT
)
{
pContext
->
numOfTry
=
0
;
memcpy
(
&
pContext
->
ipSet
,
pHead
->
content
,
sizeof
(
pContext
->
i
pSet
));
tDebug
(
"%s, redirect is received, numOf
Ips:%d"
,
pConn
->
info
,
pContext
->
ipSet
.
numOfI
ps
);
for
(
int
i
=
0
;
i
<
pContext
->
ipSet
.
numOfI
ps
;
++
i
)
pContext
->
ipSet
.
port
[
i
]
=
htons
(
pContext
->
i
pSet
.
port
[
i
]);
memcpy
(
&
pContext
->
epSet
,
pHead
->
content
,
sizeof
(
pContext
->
e
pSet
));
tDebug
(
"%s, redirect is received, numOf
Eps:%d"
,
pConn
->
info
,
pContext
->
epSet
.
numOfE
ps
);
for
(
int
i
=
0
;
i
<
pContext
->
epSet
.
numOfE
ps
;
++
i
)
pContext
->
epSet
.
port
[
i
]
=
htons
(
pContext
->
e
pSet
.
port
[
i
]);
rpcSendReqToServer
(
pRpc
,
pContext
);
rpcFreeCont
(
rpcMsg
.
pCont
);
}
else
if
(
pHead
->
code
==
TSDB_CODE_RPC_NOT_READY
)
{
...
...
@@ -1269,7 +1269,7 @@ static void rpcProcessConnError(void *param, void *id) {
tDebug
(
"%s %p, connection error happens"
,
pRpc
->
label
,
pContext
->
ahandle
);
if
(
pContext
->
numOfTry
>=
pContext
->
ipSet
.
numOfI
ps
)
{
if
(
pContext
->
numOfTry
>=
pContext
->
epSet
.
numOfE
ps
)
{
rpcMsg
.
msgType
=
pContext
->
msgType
+
1
;
rpcMsg
.
ahandle
=
pContext
->
ahandle
;
rpcMsg
.
code
=
pContext
->
code
;
...
...
@@ -1279,8 +1279,8 @@ static void rpcProcessConnError(void *param, void *id) {
rpcNotifyClient
(
pContext
,
&
rpcMsg
);
}
else
{
// move to next IP
pContext
->
i
pSet
.
inUse
++
;
pContext
->
ipSet
.
inUse
=
pContext
->
ipSet
.
inUse
%
pContext
->
ipSet
.
numOfI
ps
;
pContext
->
e
pSet
.
inUse
++
;
pContext
->
epSet
.
inUse
=
pContext
->
epSet
.
inUse
%
pContext
->
epSet
.
numOfE
ps
;
rpcSendReqToServer
(
pRpc
,
pContext
);
}
}
...
...
src/rpc/test/rclient.c
浏览文件 @
fdfa789c
...
...
@@ -22,7 +22,7 @@
typedef
struct
{
int
index
;
SRpc
IpSet
i
pSet
;
SRpc
EpSet
e
pSet
;
int
num
;
int
numOfReqs
;
int
msgSize
;
...
...
@@ -32,11 +32,11 @@ typedef struct {
void
*
pRpc
;
}
SInfo
;
static
void
processResponse
(
SRpcMsg
*
pMsg
,
SRpc
IpSet
*
pI
pSet
)
{
static
void
processResponse
(
SRpcMsg
*
pMsg
,
SRpc
EpSet
*
pE
pSet
)
{
SInfo
*
pInfo
=
(
SInfo
*
)
pMsg
->
ahandle
;
tDebug
(
"thread:%d, response is received, type:%d contLen:%d code:0x%x"
,
pInfo
->
index
,
pMsg
->
msgType
,
pMsg
->
contLen
,
pMsg
->
code
);
if
(
p
IpSet
)
pInfo
->
ipSet
=
*
pI
pSet
;
if
(
p
EpSet
)
pInfo
->
epSet
=
*
pE
pSet
;
rpcFreeCont
(
pMsg
->
pCont
);
sem_post
(
&
pInfo
->
rspSem
);
...
...
@@ -57,7 +57,7 @@ static void *sendRequest(void *param) {
rpcMsg
.
ahandle
=
pInfo
;
rpcMsg
.
msgType
=
1
;
tDebug
(
"thread:%d, send request, contLen:%d num:%d"
,
pInfo
->
index
,
pInfo
->
msgSize
,
pInfo
->
num
);
rpcSendRequest
(
pInfo
->
pRpc
,
&
pInfo
->
i
pSet
,
&
rpcMsg
);
rpcSendRequest
(
pInfo
->
pRpc
,
&
pInfo
->
e
pSet
,
&
rpcMsg
);
if
(
pInfo
->
num
%
20000
==
0
)
tInfo
(
"thread:%d, %d requests have been sent"
,
pInfo
->
index
,
pInfo
->
num
);
sem_wait
(
&
pInfo
->
rspSem
);
...
...
@@ -71,7 +71,7 @@ static void *sendRequest(void *param) {
int
main
(
int
argc
,
char
*
argv
[])
{
SRpcInit
rpcInit
;
SRpc
IpSet
i
pSet
;
SRpc
EpSet
e
pSet
;
int
msgSize
=
128
;
int
numOfReqs
=
0
;
int
appThreads
=
1
;
...
...
@@ -82,12 +82,12 @@ int main(int argc, char *argv[]) {
pthread_attr_t
thattr
;
// server info
ipSet
.
numOfI
ps
=
1
;
i
pSet
.
inUse
=
0
;
i
pSet
.
port
[
0
]
=
7000
;
i
pSet
.
port
[
1
]
=
7000
;
strcpy
(
i
pSet
.
fqdn
[
0
],
serverIp
);
strcpy
(
i
pSet
.
fqdn
[
1
],
"192.168.0.1"
);
epSet
.
numOfE
ps
=
1
;
e
pSet
.
inUse
=
0
;
e
pSet
.
port
[
0
]
=
7000
;
e
pSet
.
port
[
1
]
=
7000
;
strcpy
(
e
pSet
.
fqdn
[
0
],
serverIp
);
strcpy
(
e
pSet
.
fqdn
[
1
],
"192.168.0.1"
);
// client info
memset
(
&
rpcInit
,
0
,
sizeof
(
rpcInit
));
...
...
@@ -105,9 +105,9 @@ int main(int argc, char *argv[]) {
for
(
int
i
=
1
;
i
<
argc
;
++
i
)
{
if
(
strcmp
(
argv
[
i
],
"-p"
)
==
0
&&
i
<
argc
-
1
)
{
i
pSet
.
port
[
0
]
=
atoi
(
argv
[
++
i
]);
e
pSet
.
port
[
0
]
=
atoi
(
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-i"
)
==
0
&&
i
<
argc
-
1
)
{
tstrncpy
(
ipSet
.
fqdn
[
0
],
argv
[
++
i
],
sizeof
(
i
pSet
.
fqdn
[
0
]));
tstrncpy
(
epSet
.
fqdn
[
0
],
argv
[
++
i
],
sizeof
(
e
pSet
.
fqdn
[
0
]));
}
else
if
(
strcmp
(
argv
[
i
],
"-t"
)
==
0
&&
i
<
argc
-
1
)
{
rpcInit
.
numOfThreads
=
atoi
(
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-m"
)
==
0
&&
i
<
argc
-
1
)
{
...
...
@@ -131,7 +131,7 @@ int main(int argc, char *argv[]) {
}
else
{
printf
(
"
\n
usage: %s [options]
\n
"
,
argv
[
0
]);
printf
(
" [-i ip]: first server IP address, default is:%s
\n
"
,
serverIp
);
printf
(
" [-p port]: server port number, default is:%d
\n
"
,
i
pSet
.
port
[
0
]);
printf
(
" [-p port]: server port number, default is:%d
\n
"
,
e
pSet
.
port
[
0
]);
printf
(
" [-t threads]: number of rpc threads, default is:%d
\n
"
,
rpcInit
.
numOfThreads
);
printf
(
" [-s sessions]: number of rpc sessions, default is:%d
\n
"
,
rpcInit
.
sessions
);
printf
(
" [-m msgSize]: message body size, default is:%d
\n
"
,
msgSize
);
...
...
@@ -168,7 +168,7 @@ int main(int argc, char *argv[]) {
for
(
int
i
=
0
;
i
<
appThreads
;
++
i
)
{
pInfo
->
index
=
i
;
pInfo
->
ipSet
=
i
pSet
;
pInfo
->
epSet
=
e
pSet
;
pInfo
->
numOfReqs
=
numOfReqs
;
pInfo
->
msgSize
=
msgSize
;
sem_init
(
&
pInfo
->
rspSem
,
0
,
0
);
...
...
src/rpc/test/rsclient.c
浏览文件 @
fdfa789c
...
...
@@ -23,7 +23,7 @@
typedef
struct
{
int
index
;
SRpc
IpSet
i
pSet
;
SRpc
EpSet
e
pSet
;
int
num
;
int
numOfReqs
;
int
msgSize
;
...
...
@@ -51,7 +51,7 @@ static void *sendRequest(void *param) {
rpcMsg
.
msgType
=
1
;
tDebug
(
"thread:%d, send request, contLen:%d num:%d"
,
pInfo
->
index
,
pInfo
->
msgSize
,
pInfo
->
num
);
rpcSendRecv
(
pInfo
->
pRpc
,
&
pInfo
->
i
pSet
,
&
rpcMsg
,
&
rspMsg
);
rpcSendRecv
(
pInfo
->
pRpc
,
&
pInfo
->
e
pSet
,
&
rpcMsg
,
&
rspMsg
);
// handle response
if
(
rspMsg
.
code
!=
0
)
terror
++
;
...
...
@@ -72,7 +72,7 @@ static void *sendRequest(void *param) {
int
main
(
int
argc
,
char
*
argv
[])
{
SRpcInit
rpcInit
;
SRpc
IpSet
i
pSet
;
SRpc
EpSet
e
pSet
;
int
msgSize
=
128
;
int
numOfReqs
=
0
;
int
appThreads
=
1
;
...
...
@@ -83,12 +83,12 @@ int main(int argc, char *argv[]) {
pthread_attr_t
thattr
;
// server info
ipSet
.
numOfI
ps
=
1
;
i
pSet
.
inUse
=
0
;
i
pSet
.
port
[
0
]
=
7000
;
i
pSet
.
port
[
1
]
=
7000
;
strcpy
(
i
pSet
.
fqdn
[
0
],
serverIp
);
strcpy
(
i
pSet
.
fqdn
[
1
],
"192.168.0.1"
);
epSet
.
numOfE
ps
=
1
;
e
pSet
.
inUse
=
0
;
e
pSet
.
port
[
0
]
=
7000
;
e
pSet
.
port
[
1
]
=
7000
;
strcpy
(
e
pSet
.
fqdn
[
0
],
serverIp
);
strcpy
(
e
pSet
.
fqdn
[
1
],
"192.168.0.1"
);
// client info
memset
(
&
rpcInit
,
0
,
sizeof
(
rpcInit
));
...
...
@@ -106,9 +106,9 @@ int main(int argc, char *argv[]) {
for
(
int
i
=
1
;
i
<
argc
;
++
i
)
{
if
(
strcmp
(
argv
[
i
],
"-p"
)
==
0
&&
i
<
argc
-
1
)
{
i
pSet
.
port
[
0
]
=
atoi
(
argv
[
++
i
]);
e
pSet
.
port
[
0
]
=
atoi
(
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-i"
)
==
0
&&
i
<
argc
-
1
)
{
tstrncpy
(
ipSet
.
fqdn
[
0
],
argv
[
++
i
],
sizeof
(
i
pSet
.
fqdn
[
0
]));
tstrncpy
(
epSet
.
fqdn
[
0
],
argv
[
++
i
],
sizeof
(
e
pSet
.
fqdn
[
0
]));
}
else
if
(
strcmp
(
argv
[
i
],
"-t"
)
==
0
&&
i
<
argc
-
1
)
{
rpcInit
.
numOfThreads
=
atoi
(
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-m"
)
==
0
&&
i
<
argc
-
1
)
{
...
...
@@ -132,7 +132,7 @@ int main(int argc, char *argv[]) {
}
else
{
printf
(
"
\n
usage: %s [options]
\n
"
,
argv
[
0
]);
printf
(
" [-i ip]: first server IP address, default is:%s
\n
"
,
serverIp
);
printf
(
" [-p port]: server port number, default is:%d
\n
"
,
i
pSet
.
port
[
0
]);
printf
(
" [-p port]: server port number, default is:%d
\n
"
,
e
pSet
.
port
[
0
]);
printf
(
" [-t threads]: number of rpc threads, default is:%d
\n
"
,
rpcInit
.
numOfThreads
);
printf
(
" [-s sessions]: number of rpc sessions, default is:%d
\n
"
,
rpcInit
.
sessions
);
printf
(
" [-m msgSize]: message body size, default is:%d
\n
"
,
msgSize
);
...
...
@@ -168,7 +168,7 @@ int main(int argc, char *argv[]) {
for
(
int
i
=
0
;
i
<
appThreads
;
++
i
)
{
pInfo
->
index
=
i
;
pInfo
->
ipSet
=
i
pSet
;
pInfo
->
epSet
=
e
pSet
;
pInfo
->
numOfReqs
=
numOfReqs
;
pInfo
->
msgSize
=
msgSize
;
sem_init
(
&
pInfo
->
rspSem
,
0
,
0
);
...
...
src/rpc/test/rserver.c
浏览文件 @
fdfa789c
...
...
@@ -103,7 +103,7 @@ int retrieveAuthInfo(char *meterId, char *spi, char *encrypt, char *secret, char
return
ret
;
}
void
processRequestMsg
(
SRpcMsg
*
pMsg
,
SRpc
IpSet
*
pI
pSet
)
{
void
processRequestMsg
(
SRpcMsg
*
pMsg
,
SRpc
EpSet
*
pE
pSet
)
{
SRpcMsg
*
pTemp
;
pTemp
=
taosAllocateQitem
(
sizeof
(
SRpcMsg
));
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录