Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
Mr、小李
TDengine
提交
0f202d9f
T
TDengine
项目概览
Mr、小李
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
0f202d9f
编写于
4月 16, 2020
作者:
S
slguan
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
rearrage some codes
上级
fe12d0fe
变更
19
隐藏空白更改
内联
并排
Showing
19 changed file
with
294 addition
and
190 deletion
+294
-190
cmake/define.inc
cmake/define.inc
+6
-13
src/dnode/CMakeLists.txt
src/dnode/CMakeLists.txt
+3
-10
src/dnode/src/dnodeMClient.c
src/dnode/src/dnodeMClient.c
+2
-2
src/inc/mnode.h
src/inc/mnode.h
+11
-13
src/inc/treplica.h
src/inc/treplica.h
+9
-13
src/mnode/inc/mgmtMnode.h
src/mnode/inc/mgmtMnode.h
+13
-21
src/mnode/src/mgmtDClient.c
src/mnode/src/mgmtDClient.c
+1
-1
src/mnode/src/mgmtDServer.c
src/mnode/src/mgmtDServer.c
+1
-1
src/mnode/src/mgmtDb.c
src/mnode/src/mgmtDb.c
+1
-2
src/mnode/src/mgmtDnode.c
src/mnode/src/mgmtDnode.c
+19
-17
src/mnode/src/mgmtMain.c
src/mnode/src/mgmtMain.c
+7
-7
src/mnode/src/mgmtMnode.c
src/mnode/src/mgmtMnode.c
+193
-60
src/mnode/src/mgmtProfile.c
src/mnode/src/mgmtProfile.c
+1
-1
src/mnode/src/mgmtReplica.c
src/mnode/src/mgmtReplica.c
+12
-6
src/mnode/src/mgmtSdb.c
src/mnode/src/mgmtSdb.c
+4
-3
src/mnode/src/mgmtShell.c
src/mnode/src/mgmtShell.c
+6
-15
src/mnode/src/mgmtTable.c
src/mnode/src/mgmtTable.c
+1
-1
src/mnode/src/mgmtUser.c
src/mnode/src/mgmtUser.c
+1
-1
src/mnode/src/mgmtVgroup.c
src/mnode/src/mgmtVgroup.c
+3
-3
未找到文件。
cmake/define.inc
浏览文件 @
0f202d9f
CMAKE_MINIMUM_REQUIRED
(
VERSION
2.8
)
PROJECT
(
TDengine
)
IF
(
TD_CLUSTER
)
ADD_DEFINITIONS
(
-
D_CLUSTER
)
ENDIF
()
IF
(
TD_MPEER
)
ADD_DEFINITIONS
(
-
D_MPEER
)
ENDIF
()
IF
(
TD_VPEER
)
ADD_DEFINITIONS
(
-
D_VPEER
)
#ADD_DEFINITIONS(-DTSDB_REPLICA_MAX_NUM=3)
ELSE
()
#ADD_DEFINITIONS(-DTSDB_REPLICA_MAX_NUM=1)
IF
(
TD_SYNC
)
ADD_DEFINITIONS
(
-
D_SYNC
)
ENDIF
()
IF
(
TD_ACCOUNT
)
ADD_DEFINITIONS
(
-
D_ACCOUNT
)
ENDIF
()
IF
(
TD_ADMIN
)
ADD_DEFINITIONS
(
-
D_ADMIN
)
ENDIF
()
IF
(
TD_GRANT
)
ADD_DEFINITIONS
(
-
D_GRANT
)
ENDIF
()
...
...
src/dnode/CMakeLists.txt
浏览文件 @
0f202d9f
...
...
@@ -21,20 +21,13 @@ IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
IF
(
TD_ACCOUNT
)
TARGET_LINK_LIBRARIES
(
taosd account
)
ENDIF
()
IF
(
TD_GRANT
)
TARGET_LINK_LIBRARIES
(
taosd grant
)
ENDIF
()
IF
(
TD_CLUSTER
)
TARGET_LINK_LIBRARIES
(
taosd cluster
)
ENDIF
()
IF
(
TD_VPEER
)
TARGET_LINK_LIBRARIES
(
taosd balance sync
)
ENDIF
()
IF
(
TD_MPEER
)
TARGET_LINK_LIBRARIES
(
taosd mpeer sync
)
IF
(
TD_SYNC
)
TARGET_LINK_LIBRARIES
(
taosd replica sync
)
ENDIF
()
SET
(
PREPARE_ENV_CMD
"prepare_env_cmd"
)
...
...
src/dnode/src/dnodeMClient.c
浏览文件 @
0f202d9f
...
...
@@ -23,12 +23,12 @@
#include "tsync.h"
#include "ttime.h"
#include "ttimer.h"
#include "treplica.h"
#include "dnode.h"
#include "dnodeMClient.h"
#include "dnodeModule.h"
#include "dnodeMgmt.h"
#include "vnode.h"
#include "mpeer.h"
#define MPEER_CONTENT_LEN 2000
...
...
@@ -181,7 +181,7 @@ static void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
tsMnodeInfos
.
nodeInfos
[
i
].
nodeName
);
}
dnodeSaveMnodeIpList
();
mpeerUpdateSync
();
replicaNotify
();
}
taosTmrReset
(
dnodeSendStatusMsg
,
tsStatusInterval
*
1000
,
NULL
,
tsDnodeTmr
,
&
tsStatusTimer
);
...
...
src/inc/mnode.h
浏览文件 @
0f202d9f
...
...
@@ -43,19 +43,6 @@ struct _acct_obj;
struct
_user_obj
;
struct
_mnode_obj
;
typedef
struct
_mnode_obj
{
int32_t
mnodeId
;
int64_t
createdTime
;
int8_t
reserved
[
14
];
int8_t
updateEnd
[
1
];
int32_t
refCount
;
uint32_t
privateIp
;
uint32_t
publicIp
;
uint16_t
port
;
int8_t
role
;
char
mnodeName
[
TSDB_NODE_NAME_LEN
+
1
];
}
SMnodeObj
;
typedef
struct
_dnode_obj
{
int32_t
dnodeId
;
uint32_t
privateIp
;
...
...
@@ -88,6 +75,17 @@ typedef struct _dnode_obj {
int16_t
bandwidthUsage
;
// calc from sys.band
}
SDnodeObj
;
typedef
struct
_mnode_obj
{
int32_t
mnodeId
;
int64_t
createdTime
;
int8_t
reserved
[
14
];
int8_t
updateEnd
[
1
];
int32_t
refCount
;
int8_t
role
;
SDnodeObj
*
pDnode
;
}
SMnodeObj
;
typedef
struct
{
int32_t
dnodeId
;
uint32_t
privateIp
;
...
...
src/inc/t
balance
.h
→
src/inc/t
replica
.h
浏览文件 @
0f202d9f
...
...
@@ -13,27 +13,23 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_
BALANCE
_H
#define TDENGINE_
BALANCE
_H
#ifndef TDENGINE_
REPLICA
_H
#define TDENGINE_
REPLICA
_H
#ifdef __cplusplus
extern
"C"
{
#endif
#include <stdint.h>
#include <stdbool.h>
#include <pthread.h>
struct
_db_obj
;
struct
_vg_obj
;
struct
_dnode_obj
;
int32_t
balanceInit
();
void
balanceCleanUp
();
void
balanceNotify
();
void
balanceReset
();
int32_t
balanceAllocVnodes
(
struct
_vg_obj
*
pVgroup
);
int32_t
balanceDropDnode
(
struct
_dnode_obj
*
pDnode
);
int32_t
replicaInit
();
void
replicaCleanUp
();
void
replicaNotify
();
void
replicaReset
();
int32_t
replicaAllocVnodes
(
struct
_vg_obj
*
pVgroup
);
int32_t
replicaForwardReqToPeer
(
void
*
pHead
);
int32_t
replicaDropDnode
(
struct
_dnode_obj
*
pDnode
);
#ifdef __cplusplus
}
...
...
src/
inc/mpeer
.h
→
src/
mnode/inc/mgmtMnode
.h
浏览文件 @
0f202d9f
...
...
@@ -13,8 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_M
PEER
_H
#define TDENGINE_M
PEER
_H
#ifndef TDENGINE_M
GMT_MNODE
_H
#define TDENGINE_M
GMT_MNODE
_H
#ifdef __cplusplus
extern
"C"
{
...
...
@@ -28,29 +28,21 @@ enum _TAOS_MN_STATUS {
TAOS_MN_STATUS_READY
};
// general implementation
int32_t
mpeerInit
();
void
mpeerCleanup
();
int32_t
mgmtInitMnodes
();
void
mgmtCleanupMnodes
();
// special implementation
int32_t
mpeerInitMnodes
();
void
mpeerCleanupMnodes
();
int32_t
mpeerAddMnode
(
int32_t
dnodeId
);
int32_t
mpeerRemoveMnode
(
int32_t
dnodeId
);
int32_t
mgmtAddMnode
(
int32_t
dnodeId
);
int32_t
mgmtDropMnode
(
int32_t
dnodeId
);
void
*
m
peer
GetMnode
(
int32_t
mnodeId
);
int32_t
m
peer
GetMnodesNum
();
void
*
m
peer
GetNextMnode
(
void
*
pNode
,
struct
_mnode_obj
**
pMnode
);
void
m
peer
ReleaseMnode
(
struct
_mnode_obj
*
pMnode
);
void
*
m
gmt
GetMnode
(
int32_t
mnodeId
);
int32_t
m
gmt
GetMnodesNum
();
void
*
m
gmt
GetNextMnode
(
void
*
pNode
,
struct
_mnode_obj
**
pMnode
);
void
m
gmt
ReleaseMnode
(
struct
_mnode_obj
*
pMnode
);
bool
m
peer
IsMaster
();
bool
m
gmt
IsMaster
();
void
mpeerGetPrivateIpList
(
SRpcIpSet
*
ipSet
);
void
mpeerGetPublicIpList
(
SRpcIpSet
*
ipSet
);
void
mpeerGetMpeerInfos
(
void
*
mpeers
);
int32_t
mpeerForwardReqToPeer
(
void
*
pHead
);
void
mpeerUpdateSync
();
void
mgmtGetMnodeIpList
(
SRpcIpSet
*
ipSet
,
bool
usePublicIp
);
void
mgmtGetMnodeList
(
void
*
mpeers
);
#ifdef __cplusplus
}
...
...
src/mnode/src/mgmtDClient.c
浏览文件 @
0f202d9f
...
...
@@ -21,7 +21,7 @@
#include "tutil.h"
#include "dnode.h"
#include "mnode.h"
#include "
tbalanc
e.h"
#include "
mgmtMnod
e.h"
#include "mgmtDb.h"
#include "mgmtDnode.h"
#include "tgrant.h"
...
...
src/mnode/src/mgmtDServer.c
浏览文件 @
0f202d9f
...
...
@@ -22,7 +22,7 @@
#include "tutil.h"
#include "dnode.h"
#include "mnode.h"
#include "t
balance
.h"
#include "t
replica
.h"
#include "mgmtDb.h"
#include "mgmtDServer.h"
#include "tgrant.h"
...
...
src/mnode/src/mgmtDb.c
浏览文件 @
0f202d9f
...
...
@@ -19,12 +19,11 @@
#include "tutil.h"
#include "name.h"
#include "mnode.h"
#include "tbalance.h"
#include "mgmtAcct.h"
#include "mgmtDb.h"
#include "mgmtDnode.h"
#include "tgrant.h"
#include "m
peer
.h"
#include "m
gmtMnode
.h"
#include "mgmtShell.h"
#include "mgmtProfile.h"
#include "mgmtSdb.h"
...
...
src/mnode/src/mgmtDnode.c
浏览文件 @
0f202d9f
...
...
@@ -16,13 +16,13 @@
#define _DEFAULT_SOURCE
#include "os.h"
#include "tmodule.h"
#include "tbalance.h"
#include "tgrant.h"
#include "
mgmtDnode
.h"
#include "
treplica
.h"
#include "mnode.h"
#include "mpeer.h"
#include "mgmtDClient.h"
#include "mgmtDServer.h"
#include "mgmtDnode.h"
#include "mgmtMnode.h"
#include "mgmtSdb.h"
#include "mgmtShell.h"
#include "mgmtUser.h"
...
...
@@ -119,13 +119,15 @@ static int32_t mgmtDnodeActionDecode(SSdbOperDesc *pOper) {
static
int32_t
mgmtDnodeActionRestored
()
{
int32_t
numOfRows
=
sdbGetNumOfRows
(
tsDnodeSdb
);
if
(
numOfRows
<=
0
)
{
if
(
strcmp
(
tsMasterIp
,
tsPrivateIp
)
==
0
)
{
mgmtCreateDnode
(
inet_addr
(
tsPrivateIp
));
}
if
(
numOfRows
<=
0
&&
strcmp
(
tsMasterIp
,
tsPrivateIp
)
==
0
)
{
uint32_t
ip
=
inet_addr
(
tsPrivateIp
);
mgmtCreateDnode
(
ip
);
SDnodeObj
*
pDnode
=
mgmtGetDnodeByIp
(
ip
);
mgmtAddMnode
(
pDnode
->
dnodeId
);
mgmtReleaseDnode
(
pDnode
);
}
return
0
;
return
TSDB_CODE_SUCCESS
;
}
int32_t
mgmtInitDnodes
()
{
...
...
@@ -326,7 +328,7 @@ void mgmtProcessDnodeStatusMsg(SRpcMsg *rpcMsg) {
if
(
pDnode
->
status
==
TAOS_DN_STATUS_OFFLINE
)
{
mTrace
(
"dnode:%d, from offline to online"
,
pDnode
->
dnodeId
);
pDnode
->
status
=
TAOS_DN_STATUS_READY
;
balance
Notify
();
replica
Notify
();
mgmtMonitorDnodeModule
();
}
...
...
@@ -339,7 +341,7 @@ void mgmtProcessDnodeStatusMsg(SRpcMsg *rpcMsg) {
return
;
}
m
peerGetMpeerInfos
(
&
pRsp
->
mpeers
);
m
gmtGetMnodeList
(
&
pRsp
->
mpeers
);
pRsp
->
dnodeState
.
dnodeId
=
htonl
(
pDnode
->
dnodeId
);
pRsp
->
dnodeState
.
moduleStatus
=
htonl
(
pDnode
->
moduleStatus
);
...
...
@@ -417,7 +419,7 @@ int32_t mgmtDropDnode(SDnodeObj *pDnode) {
return
code
;
}
static
int32_t
cluster
DropDnodeByIp
(
uint32_t
ip
)
{
static
int32_t
mgmt
DropDnodeByIp
(
uint32_t
ip
)
{
SDnodeObj
*
pDnode
=
mgmtGetDnodeByIp
(
ip
);
if
(
pDnode
==
NULL
)
{
mError
(
"dnode:%s, is not exist"
,
taosIpStr
(
ip
));
...
...
@@ -465,7 +467,7 @@ static void mgmtProcessDropDnodeMsg(SQueuedMsg *pMsg) {
rpcRsp
.
code
=
TSDB_CODE_NO_RIGHTS
;
}
else
{
uint32_t
ip
=
inet_addr
(
pDrop
->
ip
);
rpcRsp
.
code
=
cluster
DropDnodeByIp
(
ip
);
rpcRsp
.
code
=
mgmt
DropDnodeByIp
(
ip
);
if
(
rpcRsp
.
code
==
TSDB_CODE_SUCCESS
)
{
mLPrint
(
"dnode:%s is dropped by %s"
,
pDrop
->
ip
,
pMsg
->
pUser
->
user
);
}
else
{
...
...
@@ -709,7 +711,7 @@ int32_t mgmtRetrieveModules(SShowObj *pShow, char *data, int32_t rows, void *pCo
return
numOfRows
;
}
static
bool
cluster
CheckConfigShow
(
SGlobalConfig
*
cfg
)
{
static
bool
mgmt
CheckConfigShow
(
SGlobalConfig
*
cfg
)
{
if
(
!
(
cfg
->
cfgType
&
TSDB_CFG_CTYPE_B_SHOW
))
return
false
;
return
true
;
...
...
@@ -746,7 +748,7 @@ static int32_t mgmtGetConfigMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pC
pShow
->
numOfRows
=
0
;
for
(
int32_t
i
=
tsGlobalConfigNum
-
1
;
i
>=
0
;
--
i
)
{
SGlobalConfig
*
cfg
=
tsGlobalConfig
+
i
;
if
(
!
cluster
CheckConfigShow
(
cfg
))
continue
;
if
(
!
mgmt
CheckConfigShow
(
cfg
))
continue
;
pShow
->
numOfRows
++
;
}
...
...
@@ -762,7 +764,7 @@ static int32_t mgmtRetrieveConfigs(SShowObj *pShow, char *data, int32_t rows, vo
for
(
int32_t
i
=
tsGlobalConfigNum
-
1
;
i
>=
0
&&
numOfRows
<
rows
;
--
i
)
{
SGlobalConfig
*
cfg
=
tsGlobalConfig
+
i
;
if
(
!
cluster
CheckConfigShow
(
cfg
))
continue
;
if
(
!
mgmt
CheckConfigShow
(
cfg
))
continue
;
char
*
pWrite
;
int32_t
cols
=
0
;
...
...
@@ -924,7 +926,7 @@ static void clusterSetModuleInDnode(SDnodeObj *pDnode, int32_t moduleType) {
mgmtUpdateDnode
(
pDnode
);
if
(
moduleType
==
TSDB_MOD_MGMT
)
{
m
peer
AddMnode
(
pDnode
->
dnodeId
);
m
gmt
AddMnode
(
pDnode
->
dnodeId
);
mPrint
(
"dnode:%d, add it into mnode list"
,
pDnode
->
dnodeId
);
}
}
...
...
@@ -934,7 +936,7 @@ static void clusterUnSetModuleInDnode(SDnodeObj *pDnode, int32_t moduleType) {
mgmtUpdateDnode
(
pDnode
);
if
(
moduleType
==
TSDB_MOD_MGMT
)
{
m
peerRemove
Mnode
(
pDnode
->
dnodeId
);
m
gmtDrop
Mnode
(
pDnode
->
dnodeId
);
mPrint
(
"dnode:%d, remove it from mnode list"
,
pDnode
->
dnodeId
);
}
}
...
...
src/mnode/src/mgmtMain.c
浏览文件 @
0f202d9f
...
...
@@ -20,10 +20,10 @@
#include "tsched.h"
#include "mnode.h"
#include "mgmtAcct.h"
#include "t
balance
.h"
#include "t
replica
.h"
#include "mgmtDnode.h"
#include "tgrant.h"
#include "m
peer
.h"
#include "m
gmtMnode
.h"
#include "mgmtDb.h"
#include "mgmtDClient.h"
#include "mgmtDServer.h"
...
...
@@ -109,7 +109,7 @@ int32_t mgmtStartSystem() {
return
-
1
;
}
if
(
m
peerInit
()
<
0
)
{
if
(
m
gmtInitMnodes
()
<
0
)
{
mError
(
"failed to init mpeers"
);
return
-
1
;
}
...
...
@@ -127,7 +127,7 @@ int32_t mgmtStartSystem() {
return
-
1
;
}
if
(
balance
Init
()
<
0
)
{
if
(
replica
Init
()
<
0
)
{
mError
(
"failed to init dnode balance"
)
}
...
...
@@ -140,7 +140,7 @@ int32_t mgmtStartSystem() {
void
mgmtStopSystem
()
{
if
(
m
peer
IsMaster
())
{
if
(
m
gmt
IsMaster
())
{
mTrace
(
"it is a master mgmt node, it could not be stopped"
);
return
;
}
...
...
@@ -152,8 +152,8 @@ void mgmtStopSystem() {
void
mgmtCleanUpSystem
()
{
mPrint
(
"starting to clean up mgmt"
);
grantCleanUp
();
m
peerCleanup
();
balance
CleanUp
();
m
gmtCleanupMnodes
();
replica
CleanUp
();
mgmtCleanUpShell
();
mgmtCleanupDClient
();
mgmtCleanupDServer
();
...
...
src/mnode/src/mgmtMnode.c
浏览文件 @
0f202d9f
...
...
@@ -16,91 +16,129 @@
#define _DEFAULT_SOURCE
#include "os.h"
#include "taoserror.h"
#include "tmodule.h"
#include "trpc.h"
#include "tsync.h"
#include "mpeer.h"
#include "treplica.h"
#include "mnode.h"
#include "mgmtMnode.h"
#include "mgmtDnode.h"
#include "mgmtSdb.h"
#include "mgmtShell.h"
#include "mgmtUser.h"
static
void
*
tsMnodeSdb
=
NULL
;
static
int32_t
tsMnodeUpdateSize
=
0
;
static
int32_t
tsMnodeIsMaster
=
true
;
static
int32_t
mgmtGetMnodeMeta
(
STableMetaMsg
*
pMeta
,
SShowObj
*
pShow
,
void
*
pConn
);
static
int32_t
mgmtRetrieveMnodes
(
SShowObj
*
pShow
,
char
*
data
,
int32_t
rows
,
void
*
pConn
);
#ifndef _MPEER
static
SMnodeObj
tsMnodeObj
=
{
0
};
static
int32_t
mgmtMnodeActionDestroy
(
SSdbOperDesc
*
pOper
)
{
tfree
(
pOper
->
pObj
);
return
TSDB_CODE_SUCCESS
;
}
int32_t
mpeerInitMnodes
()
{
tsMnodeObj
.
mnodeId
=
1
;
tsMnodeObj
.
privateIp
=
inet_addr
(
tsPrivateIp
);
tsMnodeObj
.
publicIp
=
inet_addr
(
tsPublicIp
);
tsMnodeObj
.
createdTime
=
taosGetTimestampMs
();
tsMnodeObj
.
role
=
TAOS_SYNC_ROLE_MASTER
;
tsMnodeObj
.
port
=
tsMnodeDnodePort
;
sprintf
(
tsMnodeObj
.
mnodeName
,
"m%d"
,
tsMnodeObj
.
mnodeId
);
static
int32_t
mgmtMnodeActionInsert
(
SSdbOperDesc
*
pOper
)
{
SMnodeObj
*
pMnode
=
pOper
->
pObj
;
SDnodeObj
*
pDnode
=
mgmtGetDnode
(
pMnode
->
mnodeId
);
if
(
pDnode
==
NULL
)
return
TSDB_CODE_DNODE_NOT_EXIST
;
pMnode
->
pDnode
=
pDnode
;
mgmtReleaseDnode
(
pDnode
);
return
TSDB_CODE_SUCCESS
;
}
void
mpeerCleanupMnodes
()
{}
int32_t
mpeerAddMnode
(
int32_t
dnodeId
)
{
return
TSDB_CODE_SUCCESS
;
}
int32_t
mpeerRemoveMnode
(
int32_t
dnodeId
)
{
return
TSDB_CODE_SUCCESS
;
}
void
*
mpeerGetMnode
(
int32_t
mnodeId
)
{
return
&
tsMnodeObj
;
}
int32_t
mpeerGetMnodesNum
()
{
return
1
;
}
void
mpeerReleaseMnode
(
struct
_mnode_obj
*
pMnode
)
{}
bool
mpeerIsMaster
()
{
return
tsMnodeObj
.
role
==
TAOS_SYNC_ROLE_MASTER
;
}
void
mpeerUpdateSync
()
{}
void
*
mpeerGetNextMnode
(
void
*
pNode
,
SMnodeObj
**
pMnode
)
{
if
(
*
pMnode
==
NULL
)
{
*
pMnode
=
&
tsMnodeObj
;
}
else
{
*
pMnode
=
NULL
;
static
int32_t
mgmtMnodeActionDelete
(
SSdbOperDesc
*
pOper
)
{
SMnodeObj
*
pMnode
=
pOper
->
pObj
;
mTrace
(
"mnode:%d, is dropped from sdb"
,
pMnode
->
mnodeId
);
return
TSDB_CODE_SUCCESS
;
}
static
int32_t
mgmtMnodeActionUpdate
(
SSdbOperDesc
*
pOper
)
{
SMnodeObj
*
pMnode
=
pOper
->
pObj
;
SMnodeObj
*
pSaved
=
mgmtGetMnode
(
pMnode
->
mnodeId
);
if
(
pMnode
!=
pSaved
)
{
memcpy
(
pSaved
,
pMnode
,
pOper
->
rowSize
);
free
(
pMnode
);
}
return
*
pMnode
;
return
TSDB_CODE_SUCCESS
;
}
void
mpeerGetPrivateIpList
(
SRpcIpSet
*
ipSet
)
{
ipSet
->
inUse
=
0
;
ipSet
->
numOfIps
=
1
;
ipSet
->
port
=
htons
(
tsMnodeObj
.
port
)
;
ipSet
->
ip
[
0
]
=
htonl
(
tsMnodeObj
.
privateIp
)
;
static
int32_t
mgmtMnodeActionEncode
(
SSdbOperDesc
*
pOper
)
{
SMnodeObj
*
pMnode
=
pOper
->
pObj
;
memcpy
(
pOper
->
rowData
,
pMnode
,
tsMnodeUpdateSize
)
;
pOper
->
rowSize
=
tsMnodeUpdateSize
;
return
TSDB_CODE_SUCCESS
;
}
void
mpeerGetPublicIpList
(
SRpcIpSet
*
ipSet
)
{
ipSet
->
inUse
=
0
;
ipSet
->
numOfIps
=
1
;
ipSet
->
port
=
htons
(
tsMnodeObj
.
port
);
ipSet
->
ip
[
0
]
=
htonl
(
tsMnodeObj
.
publicIp
);
}
static
int32_t
mgmtMnodeActionDecode
(
SSdbOperDesc
*
pOper
)
{
SMnodeObj
*
pMnode
=
calloc
(
1
,
sizeof
(
SMnodeObj
));
if
(
pMnode
==
NULL
)
return
TSDB_CODE_SERV_OUT_OF_MEMORY
;
void
mpeerGetMpeerInfos
(
void
*
param
)
{
SDMNodeInfos
*
mpeers
=
param
;
mpeers
->
inUse
=
0
;
mpeers
->
nodeNum
=
1
;
mpeers
->
nodeInfos
[
0
].
nodeId
=
htonl
(
tsMnodeObj
.
mnodeId
);
mpeers
->
nodeInfos
[
0
].
nodeIp
=
htonl
(
tsMnodeObj
.
privateIp
);
mpeers
->
nodeInfos
[
0
].
nodePort
=
htons
(
tsMnodeObj
.
port
);
strcpy
(
mpeers
->
nodeInfos
[
0
].
nodeName
,
tsMnodeObj
.
mnodeName
);
memcpy
(
pMnode
,
pOper
->
rowData
,
tsMnodeUpdateSize
);
pOper
->
pObj
=
pMnode
;
return
TSDB_CODE_SUCCESS
;
}
int32_t
mpeerForwardReqToPeer
(
void
*
pHead
)
{
static
int32_t
mgmtMnodeActionRestored
(
)
{
return
TSDB_CODE_SUCCESS
;
}
#endif
int32_t
mgmtInitMnodes
()
{
SMnodeObj
tObj
;
tsMnodeUpdateSize
=
(
int8_t
*
)
tObj
.
updateEnd
-
(
int8_t
*
)
&
tObj
;
SSdbTableDesc
tableDesc
=
{
.
tableId
=
SDB_TABLE_MNODE
,
.
tableName
=
"mnodes"
,
.
hashSessions
=
TSDB_MAX_MNODES
,
.
maxRowSize
=
tsMnodeUpdateSize
,
.
refCountPos
=
(
int8_t
*
)(
&
tObj
.
refCount
)
-
(
int8_t
*
)
&
tObj
,
.
keyType
=
SDB_KEY_INT
,
.
insertFp
=
mgmtMnodeActionInsert
,
.
deleteFp
=
mgmtMnodeActionDelete
,
.
updateFp
=
mgmtMnodeActionUpdate
,
.
encodeFp
=
mgmtMnodeActionEncode
,
.
decodeFp
=
mgmtMnodeActionDecode
,
.
destroyFp
=
mgmtMnodeActionDestroy
,
.
restoredFp
=
mgmtMnodeActionRestored
};
tsMnodeSdb
=
sdbOpenTable
(
&
tableDesc
);
if
(
tsMnodeSdb
==
NULL
)
{
mError
(
"failed to init mnodes data"
);
return
-
1
;
}
int32_t
mpeerInit
()
{
mgmtAddShellShowMetaHandle
(
TSDB_MGMT_TABLE_MNODE
,
mgmtGetMnodeMeta
);
mgmtAddShellShowRetrieveHandle
(
TSDB_MGMT_TABLE_MNODE
,
mgmtRetrieveMnodes
);
return
mpeerInitMnodes
();
mTrace
(
"mnodes table is created"
);
return
TSDB_CODE_SUCCESS
;
}
void
mgmtCleanupMnodes
()
{
sdbCloseTable
(
tsMnodeSdb
);
}
int32_t
mgmtGetMnodesNum
()
{
return
sdbGetNumOfRows
(
tsMnodeSdb
);
}
void
*
mgmtGetMnode
(
int32_t
mnodeId
)
{
return
sdbGetRow
(
tsMnodeSdb
,
&
mnodeId
);
}
void
mgmtReleaseMnode
(
struct
_mnode_obj
*
pMnode
)
{
sdbDecRef
(
tsMnodeSdb
,
pMnode
);
}
void
mpeerCleanup
()
{
mpeerCleanupMnodes
();
void
*
mgmtGetNextMnode
(
void
*
pNode
,
SMnodeObj
**
pMnode
)
{
return
sdbFetchRow
(
tsMnodeSdb
,
pNode
,
(
void
**
)
pMnode
);
}
static
char
*
m
peer
GetMnodeRoleStr
(
int32_t
role
)
{
static
char
*
m
gmt
GetMnodeRoleStr
(
int32_t
role
)
{
switch
(
role
)
{
case
TAOS_SYNC_ROLE_OFFLINE
:
return
"offline"
;
...
...
@@ -115,6 +153,101 @@ static char *mpeerGetMnodeRoleStr(int32_t role) {
}
}
bool
mgmtIsMaster
()
{
return
tsMnodeIsMaster
;
}
void
mgmtGetMnodeIpList
(
SRpcIpSet
*
ipSet
,
bool
usePublicIp
)
{
void
*
pNode
=
NULL
;
while
(
1
)
{
SMnodeObj
*
pMnode
=
NULL
;
pNode
=
mgmtGetNextMnode
(
pNode
,
&
pMnode
);
if
(
pMnode
==
NULL
)
break
;
if
(
usePublicIp
)
{
ipSet
->
ip
[
ipSet
->
numOfIps
]
=
htonl
(
pMnode
->
pDnode
->
publicIp
);
}
else
{
ipSet
->
ip
[
ipSet
->
numOfIps
]
=
htonl
(
pMnode
->
pDnode
->
privateIp
);
}
if
(
pMnode
->
role
==
TAOS_SYNC_ROLE_MASTER
)
{
ipSet
->
inUse
=
ipSet
->
numOfIps
;
}
ipSet
->
numOfIps
++
;
ipSet
->
port
=
htons
(
pMnode
->
pDnode
->
mnodeShellPort
);
mgmtReleaseMnode
(
pMnode
);
}
}
void
mgmtGetMnodeList
(
void
*
param
)
{
SDMNodeInfos
*
mnodes
=
param
;
mnodes
->
inUse
=
0
;
int32_t
index
=
0
;
void
*
pNode
=
NULL
;
while
(
1
)
{
SMnodeObj
*
pMnode
=
NULL
;
pNode
=
mgmtGetNextMnode
(
pNode
,
&
pMnode
);
if
(
pMnode
==
NULL
)
break
;
mnodes
->
nodeInfos
[
index
].
nodeId
=
htonl
(
pMnode
->
mnodeId
);
mnodes
->
nodeInfos
[
index
].
nodeIp
=
htonl
(
pMnode
->
pDnode
->
privateIp
);
mnodes
->
nodeInfos
[
index
].
nodePort
=
htons
(
pMnode
->
pDnode
->
mnodeDnodePort
);
strcpy
(
mnodes
->
nodeInfos
[
index
].
nodeName
,
pMnode
->
pDnode
->
dnodeName
);
mPrint
(
"node:%d role:%s"
,
pMnode
->
mnodeId
,
mgmtGetMnodeRoleStr
(
pMnode
->
role
));
if
(
pMnode
->
role
==
TAOS_SYNC_ROLE_MASTER
)
{
mnodes
->
inUse
=
index
;
mPrint
(
"node:%d inUse:%d"
,
pMnode
->
mnodeId
,
mnodes
->
inUse
);
}
index
++
;
mgmtReleaseMnode
(
pMnode
);
}
mnodes
->
nodeNum
=
index
;
}
int32_t
mgmtAddMnode
(
int32_t
dnodeId
)
{
SMnodeObj
*
pMnode
=
calloc
(
1
,
sizeof
(
SMnodeObj
));
pMnode
->
mnodeId
=
dnodeId
;
pMnode
->
createdTime
=
taosGetTimestampMs
();
SSdbOperDesc
oper
=
{
.
type
=
SDB_OPER_GLOBAL
,
.
table
=
tsMnodeSdb
,
.
pObj
=
pMnode
,
};
int32_t
code
=
sdbInsertRow
(
&
oper
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
tfree
(
pMnode
);
code
=
TSDB_CODE_SDB_ERROR
;
}
return
code
;
}
int32_t
mgmtDropMnode
(
int32_t
dnodeId
)
{
SMnodeObj
*
pMnode
=
sdbGetRow
(
tsMnodeSdb
,
&
dnodeId
);
if
(
pMnode
==
NULL
)
{
return
TSDB_CODE_DNODE_NOT_EXIST
;
}
SSdbOperDesc
oper
=
{
.
type
=
SDB_OPER_GLOBAL
,
.
table
=
tsMnodeSdb
,
.
pObj
=
pMnode
};
int32_t
code
=
sdbDeleteRow
(
&
oper
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
code
=
TSDB_CODE_SDB_ERROR
;
}
sdbDecRef
(
tsMnodeSdb
,
pMnode
);
return
code
;
}
static
int32_t
mgmtGetMnodeMeta
(
STableMetaMsg
*
pMeta
,
SShowObj
*
pShow
,
void
*
pConn
)
{
SUserObj
*
pUser
=
mgmtGetUserFromConn
(
pConn
,
NULL
);
if
(
pUser
==
NULL
)
return
0
;
...
...
@@ -162,7 +295,7 @@ static int32_t mgmtGetMnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pCo
pShow
->
offset
[
i
]
=
pShow
->
offset
[
i
-
1
]
+
pShow
->
bytes
[
i
-
1
];
}
pShow
->
numOfRows
=
m
peer
GetMnodesNum
();
pShow
->
numOfRows
=
m
gmt
GetMnodesNum
();
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
pShow
->
pNode
=
NULL
;
mgmtReleaseUser
(
pUser
);
...
...
@@ -178,7 +311,7 @@ static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, voi
char
ipstr
[
32
];
while
(
numOfRows
<
rows
)
{
pShow
->
pNode
=
m
peer
GetNextMnode
(
pShow
->
pNode
,
&
pMnode
);
pShow
->
pNode
=
m
gmt
GetNextMnode
(
pShow
->
pNode
,
&
pMnode
);
if
(
pMnode
==
NULL
)
break
;
cols
=
0
;
...
...
@@ -187,12 +320,12 @@ static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, voi
*
(
int16_t
*
)
pWrite
=
pMnode
->
mnodeId
;
cols
++
;
tinet_ntoa
(
ipstr
,
pMnode
->
privateIp
);
tinet_ntoa
(
ipstr
,
pMnode
->
p
Dnode
->
p
rivateIp
);
pWrite
=
data
+
pShow
->
offset
[
cols
]
*
rows
+
pShow
->
bytes
[
cols
]
*
numOfRows
;
strcpy
(
pWrite
,
ipstr
);
cols
++
;
tinet_ntoa
(
ipstr
,
pMnode
->
publicIp
);
tinet_ntoa
(
ipstr
,
pMnode
->
p
Dnode
->
p
ublicIp
);
pWrite
=
data
+
pShow
->
offset
[
cols
]
*
rows
+
pShow
->
bytes
[
cols
]
*
numOfRows
;
strcpy
(
pWrite
,
ipstr
);
cols
++
;
...
...
@@ -202,15 +335,15 @@ static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, voi
cols
++
;
pWrite
=
data
+
pShow
->
offset
[
cols
]
*
rows
+
pShow
->
bytes
[
cols
]
*
numOfRows
;
strcpy
(
pWrite
,
m
peer
GetMnodeRoleStr
(
pMnode
->
role
));
strcpy
(
pWrite
,
m
gmt
GetMnodeRoleStr
(
pMnode
->
role
));
cols
++
;
numOfRows
++
;
m
peer
ReleaseMnode
(
pMnode
);
m
gmt
ReleaseMnode
(
pMnode
);
}
pShow
->
numOfReads
+=
numOfRows
;
return
numOfRows
;
}
\ No newline at end of file
}
src/mnode/src/mgmtProfile.c
浏览文件 @
0f202d9f
...
...
@@ -19,7 +19,7 @@
#include "mgmtAcct.h"
#include "mgmtDnode.h"
#include "mgmtDb.h"
#include "m
peer
.h"
#include "m
gmtMnode
.h"
#include "mgmtProfile.h"
#include "mgmtShell.h"
#include "mgmtTable.h"
...
...
src/mnode/src/mgmt
Balance
.c
→
src/mnode/src/mgmt
Replica
.c
浏览文件 @
0f202d9f
...
...
@@ -14,17 +14,23 @@
*/
#define _DEFAULT_SOURCE
#include "tbalance.h"
#include "os.h"
#include "trpc.h"
#include "treplica.h"
#include "mnode.h"
#include "mgmtMnode.h"
#include "mgmtDnode.h"
#include "mgmtVgroup.h"
#ifndef _VPEER
int32_t
balanceInit
()
{
return
0
;
}
void
balanceCleanUp
()
{}
void
balanceNotify
()
{}
#ifndef _SYNC
int32_t
balanceAllocVnodes
(
SVgObj
*
pVgroup
)
{
int32_t
replicaInit
()
{
return
TSDB_CODE_SUCCESS
;
}
void
replicaCleanUp
()
{}
void
replicaNotify
()
{}
void
replicaReset
()
{}
int32_t
replicaForwardReqToPeer
(
void
*
pHead
)
{
return
TSDB_CODE_SUCCESS
;
}
int32_t
replicaAllocVnodes
(
SVgObj
*
pVgroup
)
{
void
*
pNode
=
NULL
;
SDnodeObj
*
pDnode
=
NULL
;
SDnodeObj
*
pSelDnode
=
NULL
;
...
...
src/mnode/src/mgmtSdb.c
浏览文件 @
0f202d9f
...
...
@@ -18,11 +18,12 @@
#include "taoserror.h"
#include "tlog.h"
#include "trpc.h"
#include "treplica.h"
#include "tqueue.h"
#include "twal.h"
#include "hashint.h"
#include "hashstr.h"
#include "m
peer
.h"
#include "m
gmtMnode
.h"
#include "mgmtSdb.h"
typedef
struct
_SSdbTable
{
...
...
@@ -131,7 +132,7 @@ int32_t sdbInit() {
sdbTrace
(
"sdb is initialized, version:%d totalRows:%d numOfTables:%d"
,
tsSdbObj
->
version
,
totalRows
,
numOfTables
);
mpeerUpdateSync
();
replicaNotify
();
return
TSDB_CODE_SUCCESS
;
}
...
...
@@ -264,7 +265,7 @@ static int32_t sdbProcessWriteFromApp(SSdbTable *pTable, SWalHead *pHead, int32_
tsSdbObj
->
version
++
;
pHead
->
version
=
tsSdbObj
->
version
;
code
=
mpeer
ForwardReqToPeer
(
pHead
);
code
=
replica
ForwardReqToPeer
(
pHead
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
pthread_mutex_unlock
(
&
tsSdbObj
->
mutex
);
sdbError
(
"table:%s, failed to forward %s record:%s from file, version:%"
PRId64
", reason:%s"
,
pTable
->
tableName
,
...
...
src/mnode/src/mgmtShell.c
浏览文件 @
0f202d9f
...
...
@@ -23,11 +23,10 @@
#include "dnode.h"
#include "mnode.h"
#include "mgmtAcct.h"
#include "tbalance.h"
#include "mgmtDb.h"
#include "mgmtDnode.h"
#include "tgrant.h"
#include "m
peer
.h"
#include "m
gmtMnode
.h"
#include "mgmtProfile.h"
#include "mgmtSdb.h"
#include "mgmtShell.h"
...
...
@@ -141,7 +140,7 @@ static void mgmtProcessMsgFromShell(SRpcMsg *rpcMsg) {
return
;
}
if
(
!
m
peer
IsMaster
())
{
if
(
!
m
gmt
IsMaster
())
{
// rpcSendRedirectRsp(rpcMsg->handle, mgmtGetMnodeIpListForRedirect());
mgmtSendSimpleResp
(
rpcMsg
->
handle
,
TSDB_CODE_NO_MASTER
);
rpcFreeCont
(
rpcMsg
->
pCont
);
...
...
@@ -329,12 +328,8 @@ static void mgmtProcessHeartBeatMsg(SQueuedMsg *pMsg) {
return
;
}
if
(
pMsg
->
usePublicIp
)
{
mpeerGetPublicIpList
(
&
pHBRsp
->
ipList
);
}
else
{
mpeerGetPrivateIpList
(
&
pHBRsp
->
ipList
);
}
mgmtGetMnodeIpList
(
&
pHBRsp
->
ipList
,
pMsg
->
usePublicIp
);
/*
* TODO
* Dispose kill stream or kill query message
...
...
@@ -415,12 +410,8 @@ static void mgmtProcessConnectMsg(SQueuedMsg *pMsg) {
pConnectRsp
->
writeAuth
=
pUser
->
writeAuth
;
pConnectRsp
->
superAuth
=
pUser
->
superAuth
;
if
(
pMsg
->
usePublicIp
)
{
mpeerGetPublicIpList
(
&
pConnectRsp
->
ipList
);
}
else
{
mpeerGetPrivateIpList
(
&
pConnectRsp
->
ipList
);
}
mgmtGetMnodeIpList
(
&
pConnectRsp
->
ipList
,
pMsg
->
usePublicIp
);
connect_over:
rpcRsp
.
code
=
code
;
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
...
...
src/mnode/src/mgmtTable.c
浏览文件 @
0f202d9f
...
...
@@ -29,7 +29,7 @@
#include "mgmtDnode.h"
#include "mgmtDServer.h"
#include "tgrant.h"
#include "m
peer
.h"
#include "m
gmtMnode
.h"
#include "mgmtProfile.h"
#include "mgmtSdb.h"
#include "mgmtShell.h"
...
...
src/mnode/src/mgmtUser.c
浏览文件 @
0f202d9f
...
...
@@ -20,7 +20,7 @@
#include "tutil.h"
#include "mgmtAcct.h"
#include "tgrant.h"
#include "m
peer
.h"
#include "m
gmtMnode
.h"
#include "mgmtSdb.h"
#include "mgmtShell.h"
#include "mgmtUser.h"
...
...
src/mnode/src/mgmtVgroup.c
浏览文件 @
0f202d9f
...
...
@@ -17,14 +17,14 @@
#include "os.h"
#include "taoserror.h"
#include "tlog.h"
#include "tbalance.h"
#include "tsync.h"
#include "treplica.h"
#include "mgmtDnode.h"
#include "mnode.h"
#include "mgmtDb.h"
#include "mgmtDClient.h"
#include "mgmtDServer.h"
#include "m
peer
.h"
#include "m
gmtMnode
.h"
#include "mgmtProfile.h"
#include "mgmtSdb.h"
#include "mgmtShell.h"
...
...
@@ -244,7 +244,7 @@ void mgmtCreateVgroup(SQueuedMsg *pMsg, SDbObj *pDb) {
strcpy
(
pVgroup
->
dbName
,
pDb
->
name
);
pVgroup
->
numOfVnodes
=
pDb
->
cfg
.
replications
;
pVgroup
->
createdTime
=
taosGetTimestampMs
();
if
(
balance
AllocVnodes
(
pVgroup
)
!=
0
)
{
if
(
replica
AllocVnodes
(
pVgroup
)
!=
0
)
{
mError
(
"db:%s, no enough dnode to alloc %d vnodes to vgroup"
,
pDb
->
name
,
pVgroup
->
numOfVnodes
);
free
(
pVgroup
);
mgmtSendSimpleResp
(
pMsg
->
thandle
,
TSDB_CODE_NO_ENOUGH_DNODES
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录