Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
4131fbea
T
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
提交
4131fbea
编写于
3月 18, 2022
作者:
S
Shengliang Guan
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
shm
上级
846129fd
变更
22
隐藏空白更改
内联
并排
Showing
22 changed file
with
204 addition
and
396 deletion
+204
-396
include/common/tmsg.h
include/common/tmsg.h
+1
-1
include/util/taoserror.h
include/util/taoserror.h
+8
-17
source/dnode/mgmt/bnode/inc/bmInt.h
source/dnode/mgmt/bnode/inc/bmInt.h
+1
-0
source/dnode/mgmt/bnode/src/bmInt.c
source/dnode/mgmt/bnode/src/bmInt.c
+1
-0
source/dnode/mgmt/container/inc/dnd.h
source/dnode/mgmt/container/inc/dnd.h
+1
-6
source/dnode/mgmt/container/src/dndTransport.c
source/dnode/mgmt/container/src/dndTransport.c
+1
-1
source/dnode/mgmt/dnode/src/dmFile.c
source/dnode/mgmt/dnode/src/dmFile.c
+2
-2
source/dnode/mgmt/mnode/inc/mm.h
source/dnode/mgmt/mnode/inc/mm.h
+0
-4
source/dnode/mgmt/mnode/inc/mmInt.h
source/dnode/mgmt/mnode/inc/mmInt.h
+12
-17
source/dnode/mgmt/mnode/src/mmFile.c
source/dnode/mgmt/mnode/src/mmFile.c
+9
-18
source/dnode/mgmt/mnode/src/mmInt.c
source/dnode/mgmt/mnode/src/mmInt.c
+115
-209
source/dnode/mgmt/mnode/src/mmMsg.c
source/dnode/mgmt/mnode/src/mmMsg.c
+22
-67
source/dnode/mgmt/mnode/src/mmWorker.c
source/dnode/mgmt/mnode/src/mmWorker.c
+9
-27
source/dnode/mgmt/qnode/inc/qmInt.h
source/dnode/mgmt/qnode/inc/qmInt.h
+1
-0
source/dnode/mgmt/qnode/src/qmInt.c
source/dnode/mgmt/qnode/src/qmInt.c
+1
-0
source/dnode/mgmt/snode/inc/smInt.h
source/dnode/mgmt/snode/inc/smInt.h
+1
-0
source/dnode/mgmt/snode/src/smInt.c
source/dnode/mgmt/snode/src/smInt.c
+1
-0
source/dnode/mgmt/test/mnode/dmnode.cpp
source/dnode/mgmt/test/mnode/dmnode.cpp
+8
-8
source/dnode/mgmt/vnode/inc/vmInt.h
source/dnode/mgmt/vnode/inc/vmInt.h
+1
-1
source/dnode/mgmt/vnode/src/vmFile.c
source/dnode/mgmt/vnode/src/vmFile.c
+1
-1
source/dnode/mnode/impl/src/mndMnode.c
source/dnode/mnode/impl/src/mndMnode.c
+4
-4
source/util/src/terror.c
source/util/src/terror.c
+4
-13
未找到文件。
include/common/tmsg.h
浏览文件 @
4131fbea
...
...
@@ -2327,7 +2327,7 @@ struct SMgmtWrapper;
typedef
int32_t
(
*
PutToQueueFp
)(
struct
SMgmtWrapper
*
pWrapper
,
struct
SRpcMsg
*
pReq
);
typedef
int32_t
(
*
SendReqFp
)(
struct
SMgmtWrapper
*
pWrapper
,
struct
SEpSet
*
epSet
,
struct
SRpcMsg
*
rpcMsg
);
typedef
int32_t
(
*
SendMnodeReqFp
)(
struct
SMgmtWrapper
*
pWrapper
,
struct
SRpcMsg
*
rpcMsg
);
typedef
int32_t
(
*
SendRspFp
)(
struct
SMgmtWrapper
*
pWrapper
,
struct
SRpcMsg
*
rpcMsg
);
typedef
void
(
*
SendRspFp
)(
struct
SMgmtWrapper
*
pWrapper
,
struct
SRpcMsg
*
rpcMsg
);
#ifdef __cplusplus
}
...
...
include/util/taoserror.h
浏览文件 @
4131fbea
...
...
@@ -86,10 +86,6 @@ int32_t* taosGetErrno();
#define TSDB_CODE_INVALID_VERSION_STRING TAOS_DEF_ERROR_CODE(0, 0x0121)
#define TSDB_CODE_VERSION_NOT_COMPATIBLE TAOS_DEF_ERROR_CODE(0, 0x0122)
#define TSDB_CODE_COMPRESS_ERROR TAOS_DEF_ERROR_CODE(0, 0x0123)
#define TSDB_CODE_NODE_ALREADY_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0124)
#define TSDB_CODE_NODE_NOT_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0125)
#define TSDB_CODE_NODE_PARSE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0126)
#define TSDB_CODE_NODE_INVALID_OPTION TAOS_DEF_ERROR_CODE(0, 0x0127)
//client
#define TSDB_CODE_TSC_INVALID_OPERATION TAOS_DEF_ERROR_CODE(0, 0x0200)
...
...
@@ -282,19 +278,14 @@ int32_t* taosGetErrno();
#define TSDB_CODE_DND_ACTION_IN_PROGRESS TAOS_DEF_ERROR_CODE(0, 0x0400)
#define TSDB_CODE_DND_OFFLINE TAOS_DEF_ERROR_CODE(0, 0x0401)
#define TSDB_CODE_DND_INVALID_MSG_LEN TAOS_DEF_ERROR_CODE(0, 0x0402)
#define TSDB_CODE_DND_DNODE_READ_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0410)
#define TSDB_CODE_DND_DNODE_WRITE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0411)
#define TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0420)
#define TSDB_CODE_DND_MNODE_NOT_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0421)
#define TSDB_CODE_DND_MNODE_INVALID_OPTION TAOS_DEF_ERROR_CODE(0, 0x0422)
#define TSDB_CODE_DND_MNODE_READ_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0423)
#define TSDB_CODE_DND_MNODE_WRITE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0424)
#define TSDB_CODE_DND_VNODE_ALREADY_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0460)
#define TSDB_CODE_DND_VNODE_NOT_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0461)
#define TSDB_CODE_DND_VNODE_INVALID_OPTION TAOS_DEF_ERROR_CODE(0, 0x0462)
#define TSDB_CODE_DND_VNODE_READ_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0463)
#define TSDB_CODE_DND_VNODE_WRITE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0464)
#define TSDB_CODE_DND_VNODE_TOO_MANY_VNODES TAOS_DEF_ERROR_CODE(0, 0x0465)
#define TSDB_CODE_NODE_ALREADY_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0403)
#define TSDB_CODE_NODE_NOT_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0404)
#define TSDB_CODE_NODE_PARSE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0405)
#define TSDB_CODE_NODE_INVALID_OPTION TAOS_DEF_ERROR_CODE(0, 0x0406)
#define TSDB_CODE_DND_VNODE_ALREADY_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0410)
#define TSDB_CODE_DND_VNODE_NOT_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0411)
#define TSDB_CODE_DND_VNODE_INVALID_OPTION TAOS_DEF_ERROR_CODE(0, 0x0412)
#define TSDB_CODE_DND_VNODE_TOO_MANY_VNODES TAOS_DEF_ERROR_CODE(0, 0x0413)
// vnode
#define TSDB_CODE_VND_ACTION_IN_PROGRESS TAOS_DEF_ERROR_CODE(0, 0x0500)
...
...
source/dnode/mgmt/bnode/inc/bmInt.h
浏览文件 @
4131fbea
...
...
@@ -17,6 +17,7 @@
#define _TD_DND_BNODE_INT_H_
#include "bm.h"
#include "bnode.h"
#ifdef __cplusplus
extern
"C"
{
...
...
source/dnode/mgmt/bnode/src/bmInt.c
浏览文件 @
4131fbea
...
...
@@ -23,6 +23,7 @@ static void bmInitOption(SBnodeMgmt *pMgmt, SBnodeOpt *pOption) {
pOption
->
pWrapper
=
pMgmt
->
pWrapper
;
pOption
->
sendReqFp
=
dndSendReqToDnode
;
pOption
->
sendMnodeReqFp
=
dndSendReqToMnode
;
pOption
->
sendRspFp
=
dndSendRsp
;
pOption
->
dnodeId
=
pDnode
->
dnodeId
;
pOption
->
clusterId
=
pDnode
->
clusterId
;
}
...
...
source/dnode/mgmt/container/inc/dnd.h
浏览文件 @
4131fbea
...
...
@@ -36,13 +36,8 @@
#include "tworker.h"
#include "dnode.h"
#include "bnode.h"
#include "mnode.h"
#include "qnode.h"
#include "snode.h"
#include "tfs.h"
#include "
vnode
.h"
#include "
wal
.h"
#ifdef __cplusplus
extern
"C"
{
...
...
source/dnode/mgmt/container/src/dndTransport.c
浏览文件 @
4131fbea
...
...
@@ -293,7 +293,7 @@ int32_t dndSendReqToMnode(SMgmtWrapper *pWrapper, SRpcMsg *pReq) {
}
void
dndSendRpcRsp
(
SMgmtWrapper
*
pWrapper
,
SRpcMsg
*
pRsp
)
{
if
(
pRsp
->
code
==
TSDB_CODE_
DND_M
NODE_NOT_DEPLOYED
||
pRsp
->
code
==
TSDB_CODE_APP_NOT_READY
)
{
if
(
pRsp
->
code
==
TSDB_CODE_NODE_NOT_DEPLOYED
||
pRsp
->
code
==
TSDB_CODE_APP_NOT_READY
)
{
SMgmtWrapper
*
pDnodeWrapper
=
dndAcquireWrapper
(
pWrapper
->
pDnode
,
DNODE
);
dmSendRedirectRsp
(
pDnodeWrapper
->
pMgmt
,
pRsp
);
}
else
{
...
...
source/dnode/mgmt/dnode/src/dmFile.c
浏览文件 @
4131fbea
...
...
@@ -21,7 +21,7 @@ static bool dmIsEpChanged(SDnodeMgmt *pMgmt, int32_t dnodeId, const char *ep);
static
void
dmResetDnodes
(
SDnodeMgmt
*
pMgmt
,
SArray
*
dnodeEps
);
int32_t
dmReadFile
(
SDnodeMgmt
*
pMgmt
)
{
int32_t
code
=
TSDB_CODE_
DND_DNODE_READ
_FILE_ERROR
;
int32_t
code
=
TSDB_CODE_
NODE_PARSE
_FILE_ERROR
;
int32_t
len
=
0
;
int32_t
maxLen
=
256
*
1024
;
char
*
content
=
calloc
(
1
,
maxLen
+
1
);
...
...
@@ -203,7 +203,7 @@ int32_t dmWriteFile(SDnodeMgmt *pMgmt) {
snprintf
(
realfile
,
sizeof
(
realfile
),
"%s%smnode.json"
,
pMgmt
->
path
,
TD_DIRSEP
);
if
(
taosRenameFile
(
file
,
realfile
)
!=
0
)
{
terrno
=
T
SDB_CODE_DND_MNODE_WRITE_FILE_ERROR
;
terrno
=
T
AOS_SYSTEM_ERROR
(
errno
)
;
dError
(
"failed to rename %s since %s"
,
file
,
terrstr
());
return
-
1
;
}
...
...
source/dnode/mgmt/mnode/inc/mm.h
浏览文件 @
4131fbea
...
...
@@ -23,10 +23,6 @@ extern "C" {
#endif
void
mmGetMgmtFp
(
SMgmtWrapper
*
pMgmt
);
void
mmInitMsgHandles
(
SMgmtWrapper
*
pWrapper
);
int32_t
mmProcessCreateReq
(
SMnodeMgmt
*
pMgmt
,
SNodeMsg
*
pMsg
);
int32_t
mmProcessDropReq
(
SMnodeMgmt
*
pMgmt
,
SNodeMsg
*
pMsg
);
int32_t
mmGetUserAuth
(
SMgmtWrapper
*
pWrapper
,
char
*
user
,
char
*
spi
,
char
*
encrypt
,
char
*
secret
,
char
*
ckey
);
int32_t
mmGetMonitorInfo
(
SMgmtWrapper
*
pWrapper
,
SMonClusterInfo
*
pClusterInfo
,
SMonVgroupInfo
*
pVgroupInfo
,
...
...
source/dnode/mgmt/mnode/inc/mmInt.h
浏览文件 @
4131fbea
...
...
@@ -16,43 +16,39 @@
#ifndef _TD_DND_MNODE_INT_H_
#define _TD_DND_MNODE_INT_H_
#include "dm.h"
#include "mm.h"
#include "mnode.h"
#ifdef __cplusplus
extern
"C"
{
#endif
typedef
struct
SMnodeMgmt
{
int32_t
refCount
;
int8_t
deployed
;
int8_t
dropped
;
int8_t
replica
;
int8_t
selfIndex
;
SReplica
replicas
[
TSDB_MAX_REPLICA
];
SMnode
*
pMnode
;
SDnode
*
pDnode
;
SMgmtWrapper
*
pWrapper
;
const
char
*
path
;
SRWLatch
latch
;
SDnodeWorker
readWorker
;
SDnodeWorker
writeWorker
;
SDnodeWorker
syncWorker
;
SReplica
replicas
[
TSDB_MAX_REPLICA
];
int8_t
replica
;
int8_t
selfIndex
;
}
SMnodeMgmt
;
// mmFile.c
int32_t
mmReadFile
(
SMnodeMgmt
*
pMgmt
);
int32_t
mmWriteFile
(
SMnodeMgmt
*
pMgmt
);
int32_t
mmReadFile
(
SMnodeMgmt
*
pMgmt
,
bool
*
pDeployed
);
int32_t
mmWriteFile
(
SMnodeMgmt
*
pMgmt
,
bool
deployed
);
// mmInt.c
SMnode
*
mmAcquire
(
SMnodeMgmt
*
pMgmt
);
void
mmRelease
(
SMnodeMgmt
*
pMgmt
,
SMnode
*
pMnode
);
int32_t
mmOpen
(
SMnodeMgmt
*
pMgmt
,
SMnodeOpt
*
pOption
);
int32_t
mmAlter
(
SMnodeMgmt
*
pMgmt
,
SMnodeOpt
*
pOption
);
int32_t
mmDrop
(
SMnodeMgmt
*
pMgmt
);
int32_t
mmBuildOptionFromReq
(
SMnodeMgmt
*
pMgmt
,
SMnodeOpt
*
pOption
,
SDCreateMnodeReq
*
pCreate
);
int32_t
mmOpenFromMsg
(
SMgmtWrapper
*
pWrapper
,
SDCreateMnodeReq
*
pReq
);
int32_t
mmDrop
(
SMgmtWrapper
*
pWrapper
);
int32_t
mmAlter
(
SMnodeMgmt
*
pMgmt
,
SDAlterMnodeReq
*
pReq
);
// mmMsg.c
void
mmInitMsgHandles
(
SMgmtWrapper
*
pWrapper
);
int32_t
mmProcessCreateReq
(
SMgmtWrapper
*
pWrapper
,
SNodeMsg
*
pMsg
);
int32_t
mmProcessDropReq
(
SMgmtWrapper
*
pWrapper
,
SNodeMsg
*
pMsg
);
int32_t
mmProcessAlterReq
(
SMnodeMgmt
*
pMgmt
,
SNodeMsg
*
pMsg
);
// mmWorker.c
...
...
@@ -61,7 +57,6 @@ void mmStopWorker(SMnodeMgmt *pMgmt);
int32_t
mmProcessWriteMsg
(
SMnodeMgmt
*
pMgmt
,
SNodeMsg
*
pMsg
);
int32_t
mmProcessSyncMsg
(
SMnodeMgmt
*
pMgmt
,
SNodeMsg
*
pMsg
);
int32_t
mmProcessReadMsg
(
SMnodeMgmt
*
pMgmt
,
SNodeMsg
*
pMsg
);
int32_t
mmPutMsgToWriteQueue
(
SMgmtWrapper
*
pWrapper
,
SRpcMsg
*
pRpcMsg
);
int32_t
mmPutMsgToReadQueue
(
SMgmtWrapper
*
pWrapper
,
SRpcMsg
*
pRpcMsg
);
...
...
source/dnode/mgmt/mnode/src/mmFile.c
浏览文件 @
4131fbea
...
...
@@ -16,8 +16,8 @@
#define _DEFAULT_SOURCE
#include "mmInt.h"
int32_t
mmReadFile
(
SMnodeMgmt
*
pMgmt
)
{
int32_t
code
=
TSDB_CODE_
DND_MNODE_READ
_FILE_ERROR
;
int32_t
mmReadFile
(
SMnodeMgmt
*
pMgmt
,
bool
*
pDeployed
)
{
int32_t
code
=
TSDB_CODE_
NODE_PARSE
_FILE_ERROR
;
int32_t
len
=
0
;
int32_t
maxLen
=
4096
;
char
*
content
=
calloc
(
1
,
maxLen
+
1
);
...
...
@@ -51,14 +51,7 @@ int32_t mmReadFile(SMnodeMgmt *pMgmt) {
dError
(
"failed to read %s since deployed not found"
,
file
);
goto
PRASE_MNODE_OVER
;
}
pMgmt
->
deployed
=
deployed
->
valueint
;
cJSON
*
dropped
=
cJSON_GetObjectItem
(
root
,
"dropped"
);
if
(
!
dropped
||
dropped
->
type
!=
cJSON_Number
)
{
dError
(
"failed to read %s since dropped not found"
,
file
);
goto
PRASE_MNODE_OVER
;
}
pMgmt
->
dropped
=
dropped
->
valueint
;
*
pDeployed
=
deployed
->
valueint
;
cJSON
*
mnodes
=
cJSON_GetObjectItem
(
root
,
"mnodes"
);
if
(
!
mnodes
||
mnodes
->
type
!=
cJSON_Array
)
{
...
...
@@ -101,7 +94,7 @@ int32_t mmReadFile(SMnodeMgmt *pMgmt) {
}
code
=
0
;
dDebug
(
"succcessed to read file %s, deployed:%d
dropped:%d"
,
file
,
pMgmt
->
deployed
,
pMgmt
->
dropp
ed
);
dDebug
(
"succcessed to read file %s, deployed:%d
"
,
file
,
*
pDeploy
ed
);
PRASE_MNODE_OVER:
if
(
content
!=
NULL
)
free
(
content
);
...
...
@@ -112,13 +105,13 @@ PRASE_MNODE_OVER:
return
code
;
}
int32_t
mmWriteFile
(
SMnodeMgmt
*
pMgmt
)
{
int32_t
mmWriteFile
(
SMnodeMgmt
*
pMgmt
,
bool
deployed
)
{
char
file
[
PATH_MAX
];
snprintf
(
file
,
sizeof
(
file
),
"%s%smnode.json.bak"
,
pMgmt
->
path
,
TD_DIRSEP
);
TdFilePtr
pFile
=
taosOpenFile
(
file
,
TD_FILE_CTEATE
|
TD_FILE_WRITE
|
TD_FILE_TRUNC
);
if
(
pFile
==
NULL
)
{
terrno
=
T
SDB_CODE_DND_MNODE_WRITE_FILE_ERROR
;
terrno
=
T
AOS_SYSTEM_ERROR
(
errno
);
;
dError
(
"failed to write %s since %s"
,
file
,
terrstr
());
return
-
1
;
}
...
...
@@ -128,9 +121,7 @@ int32_t mmWriteFile(SMnodeMgmt *pMgmt) {
char
*
content
=
calloc
(
1
,
maxLen
+
1
);
len
+=
snprintf
(
content
+
len
,
maxLen
-
len
,
"{
\n
"
);
len
+=
snprintf
(
content
+
len
,
maxLen
-
len
,
"
\"
deployed
\"
: %d,
\n
"
,
pMgmt
->
deployed
);
len
+=
snprintf
(
content
+
len
,
maxLen
-
len
,
"
\"
dropped
\"
: %d,
\n
"
,
pMgmt
->
dropped
);
len
+=
snprintf
(
content
+
len
,
maxLen
-
len
,
"
\"
deployed
\"
: %d,
\n
"
,
deployed
);
len
+=
snprintf
(
content
+
len
,
maxLen
-
len
,
"
\"
mnodes
\"
: [{
\n
"
);
for
(
int32_t
i
=
0
;
i
<
pMgmt
->
replica
;
++
i
)
{
SReplica
*
pReplica
=
&
pMgmt
->
replicas
[
i
];
...
...
@@ -154,11 +145,11 @@ int32_t mmWriteFile(SMnodeMgmt *pMgmt) {
snprintf
(
realfile
,
sizeof
(
realfile
),
"%s%smnode.json"
,
pMgmt
->
path
,
TD_DIRSEP
);
if
(
taosRenameFile
(
file
,
realfile
)
!=
0
)
{
terrno
=
T
SDB_CODE_DND_MNODE_WRITE_FILE_ERROR
;
terrno
=
T
AOS_SYSTEM_ERROR
(
errno
);
;
dError
(
"failed to rename %s since %s"
,
file
,
terrstr
());
return
-
1
;
}
dInfo
(
"successed to write %s, deployed:%d
dropped:%d"
,
realfile
,
pMgmt
->
deployed
,
pMgmt
->
dropp
ed
);
dInfo
(
"successed to write %s, deployed:%d
"
,
realfile
,
deploy
ed
);
return
0
;
}
source/dnode/mgmt/mnode/src/mmInt.c
浏览文件 @
4131fbea
...
...
@@ -16,138 +16,35 @@
#define _DEFAULT_SOURCE
#include "mmInt.h"
SMnode
*
mmAcquire
(
SMnodeMgmt
*
pMgmt
)
{
SMnode
*
pMnode
=
NULL
;
int32_t
refCount
=
0
;
taosRLockLatch
(
&
pMgmt
->
latch
);
if
(
pMgmt
->
deployed
&&
!
pMgmt
->
dropped
)
{
refCount
=
atomic_add_fetch_32
(
&
pMgmt
->
refCount
,
1
);
pMnode
=
pMgmt
->
pMnode
;
}
else
{
terrno
=
TSDB_CODE_DND_MNODE_NOT_DEPLOYED
;
}
taosRUnLockLatch
(
&
pMgmt
->
latch
);
if
(
pMnode
!=
NULL
)
{
dTrace
(
"acquire mnode, refCount:%d"
,
refCount
);
}
return
pMnode
;
}
void
mmRelease
(
SMnodeMgmt
*
pMgmt
,
SMnode
*
pMnode
)
{
if
(
pMnode
==
NULL
)
return
;
taosRLockLatch
(
&
pMgmt
->
latch
);
int32_t
refCount
=
atomic_sub_fetch_32
(
&
pMgmt
->
refCount
,
1
);
taosRUnLockLatch
(
&
pMgmt
->
latch
);
dTrace
(
"release mnode, refCount:%d"
,
refCount
);
}
int32_t
mmOpen
(
SMnodeMgmt
*
pMgmt
,
SMnodeOpt
*
pOption
)
{
SMnode
*
pMnode
=
mmAcquire
(
pMgmt
);
if
(
pMnode
!=
NULL
)
{
mmRelease
(
pMgmt
,
pMnode
);
terrno
=
TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED
;
dError
(
"failed to create mnode since %s"
,
terrstr
());
return
-
1
;
}
if
(
walInit
()
!=
0
)
{
dError
(
"failed to init wal since %s"
,
terrstr
());
mndDestroy
(
pMgmt
->
path
);
return
-
1
;
}
pMnode
=
mndOpen
(
pMgmt
->
path
,
pOption
);
if
(
pMnode
==
NULL
)
{
dError
(
"failed to open mnode since %s"
,
terrstr
());
return
-
1
;
}
if
(
mmStartWorker
(
pMgmt
)
!=
0
)
{
dError
(
"failed to start mnode worker since %s"
,
terrstr
());
mndClose
(
pMnode
);
mndDestroy
(
pMgmt
->
path
);
return
-
1
;
}
pMgmt
->
deployed
=
1
;
if
(
mmWriteFile
(
pMgmt
)
!=
0
)
{
dError
(
"failed to write mnode file since %s"
,
terrstr
());
pMgmt
->
deployed
=
0
;
mmStopWorker
(
pMgmt
);
mndClose
(
pMnode
);
mndDestroy
(
pMgmt
->
path
);
return
-
1
;
}
taosWLockLatch
(
&
pMgmt
->
latch
);
pMgmt
->
pMnode
=
pMnode
;
pMgmt
->
deployed
=
1
;
taosWUnLockLatch
(
&
pMgmt
->
latch
);
dInfo
(
"mnode open successfully"
);
return
0
;
}
int32_t
mmAlter
(
SMnodeMgmt
*
pMgmt
,
SMnodeOpt
*
pOption
)
{
SMnode
*
pMnode
=
mmAcquire
(
pMgmt
);
if
(
pMnode
==
NULL
)
{
dError
(
"failed to alter mnode since %s"
,
terrstr
());
return
-
1
;
}
if
(
mndAlter
(
pMnode
,
pOption
)
!=
0
)
{
dError
(
"failed to alter mnode since %s"
,
terrstr
());
mmRelease
(
pMgmt
,
pMnode
);
return
-
1
;
}
mmRelease
(
pMgmt
,
pMnode
);
return
0
;
static
bool
mmDeployRequired
(
SDnode
*
pDnode
)
{
if
(
pDnode
->
dnodeId
>
0
)
return
false
;
if
(
pDnode
->
clusterId
>
0
)
return
false
;
if
(
strcmp
(
pDnode
->
localEp
,
pDnode
->
firstEp
)
!=
0
)
return
false
;
return
true
;
}
int32_t
mmDrop
(
SMnodeMgmt
*
pMgmt
)
{
SMnode
*
pMnode
=
mmAcquire
(
pMgmt
)
;
if
(
pMnode
==
NULL
)
{
dError
(
"failed to drop mnode since %s"
,
terrstr
());
static
int32_t
mmRequire
(
SMgmtWrapper
*
pWrapper
,
bool
*
required
)
{
SMnode
Mgmt
mgmt
=
{
0
}
;
mgmt
.
path
=
pWrapper
->
path
;
if
(
mmReadFile
(
&
mgmt
,
required
)
!=
0
)
{
return
-
1
;
}
taosRLockLatch
(
&
pMgmt
->
latch
);
pMgmt
->
dropped
=
1
;
taosRUnLockLatch
(
&
pMgmt
->
latch
);
if
(
mmWriteFile
(
pMgmt
)
!=
0
)
{
taosRLockLatch
(
&
pMgmt
->
latch
);
pMgmt
->
dropped
=
0
;
taosRUnLockLatch
(
&
pMgmt
->
latch
);
mmRelease
(
pMgmt
,
pMnode
);
dError
(
"failed to drop mnode since %s"
,
terrstr
());
return
-
1
;
if
(
!
(
*
required
))
{
*
required
=
mmDeployRequired
(
pWrapper
->
pDnode
);
}
mmRelease
(
pMgmt
,
pMnode
);
mmStopWorker
(
pMgmt
);
pMgmt
->
deployed
=
0
;
mmWriteFile
(
pMgmt
);
mndClose
(
pMnode
);
pMgmt
->
pMnode
=
NULL
;
mndDestroy
(
pMgmt
->
path
);
return
0
;
}
static
void
mmInitOption
(
SMnodeMgmt
*
pMgmt
,
SMnodeOpt
*
pOption
)
{
SDnode
*
pDnode
=
pMgmt
->
pDnode
;
pOption
->
pWrapper
=
pMgmt
->
pWrapper
;
pOption
->
sendReqFp
=
dndSendReqToDnode
;
pOption
->
sendMnodeReqFp
=
dndSendReqToMnode
;
pOption
->
putToWriteQFp
=
mmPutMsgToWriteQueue
;
pOption
->
putToReadQFp
=
mmPutMsgToReadQueue
;
pOption
->
sendReqFp
=
dndSendReqToDnode
;
pOption
->
sendMnodeReqFp
=
dndSendReqToMnode
;
pOption
->
sendRspFp
=
dndSendRsp
;
pOption
->
dnodeId
=
pDnode
->
dnodeId
;
pOption
->
clusterId
=
pDnode
->
clusterId
;
}
...
...
@@ -175,12 +72,8 @@ static void mmBuildOptionForOpen(SMnodeMgmt *pMgmt, SMnodeOpt *pOption) {
memcpy
(
&
pOption
->
replicas
,
pMgmt
->
replicas
,
sizeof
(
SReplica
)
*
TSDB_MAX_REPLICA
);
}
int32_t
mmBuildOptionFromReq
(
SMnodeMgmt
*
pMgmt
,
SMnodeOpt
*
pOption
,
SDCreateMnodeReq
*
pCreate
)
{
SDnode
*
pDnode
=
pMgmt
->
pDnode
;
static
int32_t
mmBuildOptionFromReq
(
SMnodeMgmt
*
pMgmt
,
SMnodeOpt
*
pOption
,
SDCreateMnodeReq
*
pCreate
)
{
mmInitOption
(
pMgmt
,
pOption
);
pOption
->
dnodeId
=
pDnode
->
dnodeId
;
pOption
->
clusterId
=
pDnode
->
clusterId
;
pOption
->
replica
=
pCreate
->
replica
;
pOption
->
selfIndex
=
-
1
;
...
...
@@ -205,109 +98,136 @@ int32_t mmBuildOptionFromReq(SMnodeMgmt *pMgmt, SMnodeOpt *pOption, SDCreateMnod
return
0
;
}
static
void
mmCleanup
(
SMgmtWrapper
*
pWrapper
)
{
SMnodeMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
if
(
pMgmt
==
NULL
)
return
;
static
int32_t
mmOpenImp
(
SMnodeMgmt
*
pMgmt
,
SDCreateMnodeReq
*
pReq
)
{
SMnodeOpt
option
=
{
0
};
if
(
pReq
!=
NULL
)
{
if
(
mmBuildOptionFromReq
(
pMgmt
,
&
option
,
pReq
)
!=
0
)
{
return
-
1
;
}
}
else
{
bool
deployed
=
false
;
if
(
mmReadFile
(
pMgmt
,
&
deployed
)
!=
0
)
{
dError
(
"failed to read file since %s"
,
terrstr
());
return
-
1
;
}
dInfo
(
"mnode-mgmt start to cleanup"
);
if
(
pMgmt
->
pMnode
)
{
mmStopWorker
(
pMgmt
);
mndClose
(
pMgmt
->
pMnode
);
pMgmt
->
pMnode
=
NULL
;
if
(
!
deployed
)
{
dInfo
(
"mnode start to deploy"
);
mmBuildOptionForDeploy
(
pMgmt
,
&
option
);
}
else
{
dInfo
(
"mnode start to open"
);
mmBuildOptionForOpen
(
pMgmt
,
&
option
);
}
}
free
(
pMgmt
);
pWrapper
->
pMgmt
=
NULL
;
dInfo
(
"mnode-mgmt is cleaned up"
);
}
static
int32_t
mmInit
(
SMgmtWrapper
*
pWrapper
)
{
SDnode
*
pDnode
=
pWrapper
->
pDnode
;
SMnodeMgmt
*
pMgmt
=
calloc
(
1
,
sizeof
(
SMnodeMgmt
));
int32_t
code
=
-
1
;
SMnodeOpt
option
=
{
0
};
dInfo
(
"mnode-mgmt start to init"
);
if
(
pMgmt
==
NULL
)
goto
_OVER
;
pMgmt
->
path
=
pWrapper
->
path
;
pMgmt
->
pDnode
=
pWrapper
->
pDnode
;
pMgmt
->
pWrapper
=
pWrapper
;
taosInitRWLatch
(
&
pMgmt
->
latch
);
if
(
mmReadFile
(
pMgmt
)
!=
0
)
{
dError
(
"failed to read file since %s"
,
terrstr
());
goto
_OVER
;
pMgmt
->
pMnode
=
mndOpen
(
pMgmt
->
path
,
&
option
);
if
(
pMgmt
->
pMnode
==
NULL
)
{
dError
(
"failed to open mnode since %s"
,
terrstr
());
return
-
1
;
}
if
(
!
pMgmt
->
deployed
)
{
dInfo
(
"mnode start to deploy"
);
mmBuildOptionForDeploy
(
pMgmt
,
&
option
);
code
=
mmOpen
(
pMgmt
,
&
option
);
}
else
{
dInfo
(
"mnode start to open"
);
mmBuildOptionForOpen
(
pMgmt
,
&
option
);
code
=
mmOpen
(
pMgmt
,
&
option
);
if
(
mmStartWorker
(
pMgmt
)
!=
0
)
{
dError
(
"failed to start mnode worker since %s"
,
terrstr
());
return
-
1
;
}
_OVER:
if
(
code
==
0
)
{
pWrapper
->
pMgmt
=
pMgmt
;
dInfo
(
"mnode-mgmt is initialized"
);
}
else
{
dError
(
"failed to init mnode-mgmt since %s"
,
terrstr
());
mmCleanup
(
pWrapper
);
bool
deployed
=
true
;
if
(
dndWriteFile
(
pMgmt
->
pWrapper
,
deployed
)
!=
0
)
{
dError
(
"failed to write mnode file since %s"
,
terrstr
());
return
-
1
;
}
return
code
;
return
0
;
}
static
bool
mmDeployRequired
(
SDnode
*
pDnode
)
{
if
(
pDnode
->
dnodeId
>
0
)
{
return
false
;
static
void
mmCloseImp
(
SMnodeMgmt
*
pMgmt
)
{
if
(
pMgmt
->
pMnode
!=
NULL
)
{
mmStopWorker
(
pMgmt
);
mndClose
(
pMgmt
->
pMnode
);
pMgmt
->
pMnode
=
NULL
;
}
}
if
(
pDnode
->
clusterId
>
0
)
{
return
false
;
int32_t
mmAlter
(
SMnodeMgmt
*
pMgmt
,
SDAlterMnodeReq
*
pReq
)
{
SMnodeOpt
option
=
{
0
};
if
(
pReq
!=
NULL
)
{
if
(
mmBuildOptionFromReq
(
pMgmt
,
&
option
,
pReq
)
!=
0
)
{
return
-
1
;
}
}
if
(
strcmp
(
pDnode
->
localEp
,
pDnode
->
firstEp
)
!=
0
)
{
return
false
;
return
mndAlter
(
pMgmt
->
pMnode
,
&
option
);
}
int32_t
mmDrop
(
SMgmtWrapper
*
pWrapper
)
{
SMnodeMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
if
(
pMgmt
==
NULL
)
return
0
;
dInfo
(
"mnode-mgmt start to drop"
);
bool
deployed
=
false
;
if
(
mmWriteFile
(
pMgmt
,
deployed
)
!=
0
)
{
dError
(
"failed to drop mnode since %s"
,
terrstr
());
return
-
1
;
}
return
true
;
mmCloseImp
(
pMgmt
);
taosRemoveDir
(
pMgmt
->
path
);
pWrapper
->
pMgmt
=
NULL
;
free
(
pMgmt
);
dInfo
(
"mnode-mgmt is dropped"
);
return
0
;
}
static
int32_t
mmRequire
(
SMgmtWrapper
*
pWrapper
,
bool
*
required
)
{
SMnodeMgmt
mgmt
=
{
0
};
mgmt
.
path
=
pWrapper
->
path
;
if
(
mmReadFile
(
&
mgmt
)
!=
0
)
{
static
void
mmClose
(
SMgmtWrapper
*
pWrapper
)
{
SMnodeMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
if
(
pMgmt
==
NULL
)
return
;
dInfo
(
"mnode-mgmt start to cleanup"
);
mmCloseImp
(
pMgmt
);
pWrapper
->
pMgmt
=
NULL
;
free
(
pMgmt
);
dInfo
(
"mnode-mgmt is cleaned up"
);
}
int32_t
mmOpenFromMsg
(
SMgmtWrapper
*
pWrapper
,
SDCreateMnodeReq
*
pReq
)
{
dInfo
(
"mnode-mgmt start to init"
);
if
(
walInit
()
!=
0
)
{
dError
(
"failed to init wal since %s"
,
terrstr
());
return
-
1
;
}
if
(
mgmt
.
dropped
)
{
dInfo
(
"mnode has been dropped and needs to be deleted"
);
mndDestroy
(
mgmt
.
path
)
;
SMnodeMgmt
*
pMgmt
=
calloc
(
1
,
sizeof
(
SMnodeMgmt
));
if
(
pMgmt
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
return
-
1
;
}
if
(
mgmt
.
deployed
)
{
*
required
=
true
;
dInfo
(
"mnode has been deployed"
);
return
0
;
}
pMgmt
->
path
=
pWrapper
->
path
;
pMgmt
->
pDnode
=
pWrapper
->
pDnode
;
pMgmt
->
pWrapper
=
pWrapper
;
pWrapper
->
pMgmt
=
pMgmt
;
*
required
=
mmDeployRequired
(
pWrapper
->
pDnode
);
if
(
*
required
)
{
dInfo
(
"mnode need to be deployed"
);
int32_t
code
=
mmOpenImp
(
pMgmt
,
pReq
);
if
(
code
!=
0
)
{
dError
(
"failed to init mnode-mgmt since %s"
,
terrstr
());
mmClose
(
pWrapper
);
}
else
{
dInfo
(
"mnode-mgmt is initialized"
);
}
return
0
;
return
code
;
}
static
int32_t
mmOpen
(
SMgmtWrapper
*
pWrapper
)
{
return
mmOpenFromMsg
(
pWrapper
,
NULL
);
}
void
mmGetMgmtFp
(
SMgmtWrapper
*
pWrapper
)
{
SMgmtFp
mgmtFp
=
{
0
};
mgmtFp
.
openFp
=
mmInit
;
mgmtFp
.
closeFp
=
mmCleanup
;
mgmtFp
.
openFp
=
mmOpen
;
mgmtFp
.
closeFp
=
mmClose
;
mgmtFp
.
createMsgFp
=
mmProcessCreateReq
;
mgmtFp
.
dropMsgFp
=
mmProcessDropReq
;
mgmtFp
.
requiredFp
=
mmRequire
;
mmInitMsgHandles
(
pWrapper
);
...
...
@@ -318,16 +238,7 @@ void mmGetMgmtFp(SMgmtWrapper *pWrapper) {
int32_t
mmGetUserAuth
(
SMgmtWrapper
*
pWrapper
,
char
*
user
,
char
*
spi
,
char
*
encrypt
,
char
*
secret
,
char
*
ckey
)
{
SMnodeMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SMnode
*
pMnode
=
mmAcquire
(
pMgmt
);
if
(
pMnode
==
NULL
)
{
terrno
=
TSDB_CODE_APP_NOT_READY
;
dTrace
(
"failed to get user auth since %s"
,
terrstr
());
return
-
1
;
}
int32_t
code
=
mndRetriveAuth
(
pMnode
,
user
,
spi
,
encrypt
,
secret
,
ckey
);
mmRelease
(
pMgmt
,
pMnode
);
int32_t
code
=
mndRetriveAuth
(
pMgmt
->
pMnode
,
user
,
spi
,
encrypt
,
secret
,
ckey
);
dTrace
(
"user:%s, retrieve auth spi:%d encrypt:%d"
,
user
,
*
spi
,
*
encrypt
);
return
code
;
}
...
...
@@ -335,10 +246,5 @@ int32_t mmGetUserAuth(SMgmtWrapper *pWrapper, char *user, char *spi, char *encry
int32_t
mmGetMonitorInfo
(
SMgmtWrapper
*
pWrapper
,
SMonClusterInfo
*
pClusterInfo
,
SMonVgroupInfo
*
pVgroupInfo
,
SMonGrantInfo
*
pGrantInfo
)
{
SMnodeMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SMnode
*
pMnode
=
mmAcquire
(
pMgmt
);
if
(
pMnode
==
NULL
)
return
-
1
;
int32_t
code
=
mndGetMonitorInfo
(
pMnode
,
pClusterInfo
,
pVgroupInfo
,
pGrantInfo
);
mmRelease
(
pMgmt
,
pMnode
);
return
code
;
return
mndGetMonitorInfo
(
pMgmt
->
pMnode
,
pClusterInfo
,
pVgroupInfo
,
pGrantInfo
);
}
\ No newline at end of file
source/dnode/mgmt/mnode/src/mmMsg.c
浏览文件 @
4131fbea
...
...
@@ -16,8 +16,8 @@
#define _DEFAULT_SOURCE
#include "mmInt.h"
int32_t
mmProcessCreateReq
(
SM
nodeMgmt
*
pMgmt
,
SNodeMsg
*
pMsg
)
{
SDnode
*
pDnode
=
p
Mgmt
->
pDnode
;
int32_t
mmProcessCreateReq
(
SM
gmtWrapper
*
pWrapper
,
SNodeMsg
*
pMsg
)
{
SDnode
*
pDnode
=
p
Wrapper
->
pDnode
;
SRpcMsg
*
pReq
=
&
pMsg
->
rpcMsg
;
SDCreateMnodeReq
createReq
=
{
0
};
...
...
@@ -27,32 +27,35 @@ int32_t mmProcessCreateReq(SMnodeMgmt *pMgmt, SNodeMsg *pMsg) {
}
if
(
createReq
.
replica
<=
1
||
createReq
.
dnodeId
!=
pDnode
->
dnodeId
)
{
terrno
=
TSDB_CODE_
DND_M
NODE_INVALID_OPTION
;
terrno
=
TSDB_CODE_NODE_INVALID_OPTION
;
dError
(
"failed to create mnode since %s"
,
terrstr
());
return
-
1
;
}
else
{
return
mmOpenFromMsg
(
pWrapper
,
&
createReq
);
}
}
SMnodeOpt
option
=
{
0
};
if
(
mmBuildOptionFromReq
(
pMgmt
,
&
option
,
&
createReq
)
!=
0
)
{
terrno
=
TSDB_CODE_DND_MNODE_INVALID_OPTION
;
dError
(
"failed to create mnode since %s"
,
terrstr
());
int32_t
mmProcessDropReq
(
SMgmtWrapper
*
pWrapper
,
SNodeMsg
*
pMsg
)
{
SDnode
*
pDnode
=
pWrapper
->
pDnode
;
SRpcMsg
*
pReq
=
&
pMsg
->
rpcMsg
;
SDDropMnodeReq
dropReq
=
{
0
};
if
(
tDeserializeSMCreateDropMnodeReq
(
pReq
->
pCont
,
pReq
->
contLen
,
&
dropReq
)
!=
0
)
{
terrno
=
TSDB_CODE_INVALID_MSG
;
return
-
1
;
}
SMnode
*
pMnode
=
mmAcquire
(
pMgmt
);
if
(
pMnode
!=
NULL
)
{
mmRelease
(
pMgmt
,
pMnode
);
terrno
=
TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED
;
dError
(
"failed to create mnode since %s"
,
terrstr
());
if
(
dropReq
.
dnodeId
!=
pDnode
->
dnodeId
)
{
terrno
=
TSDB_CODE_NODE_INVALID_OPTION
;
dError
(
"failed to drop mnode since %s"
,
terrstr
());
return
-
1
;
}
else
{
return
mmDrop
(
pWrapper
);
}
dDebug
(
"start to create mnode"
);
return
mmOpen
(
pMgmt
,
&
option
);
}
int32_t
mmProcessAlterReq
(
SMnodeMgmt
*
pMgmt
,
SNodeMsg
*
pMsg
)
{
SDnode
*
pDnode
=
pMgmt
->
pDnode
;
SDnode
*
pDnode
=
pMgmt
->
pDnode
;
SRpcMsg
*
pReq
=
&
pMsg
->
rpcMsg
;
SDAlterMnodeReq
alterReq
=
{
0
};
...
...
@@ -62,60 +65,12 @@ int32_t mmProcessAlterReq(SMnodeMgmt *pMgmt, SNodeMsg *pMsg) {
}
if
(
alterReq
.
dnodeId
!=
pDnode
->
dnodeId
)
{
terrno
=
TSDB_CODE_DND_MNODE_INVALID_OPTION
;
dError
(
"failed to alter mnode since %s"
,
terrstr
());
return
-
1
;
}
SMnodeOpt
option
=
{
0
};
if
(
mmBuildOptionFromReq
(
pMgmt
,
&
option
,
&
alterReq
)
!=
0
)
{
terrno
=
TSDB_CODE_DND_MNODE_INVALID_OPTION
;
terrno
=
TSDB_CODE_NODE_INVALID_OPTION
;
dError
(
"failed to alter mnode since %s"
,
terrstr
());
return
-
1
;
}
else
{
return
mmAlter
(
pMgmt
,
&
alterReq
);
}
SMnode
*
pMnode
=
mmAcquire
(
pMgmt
);
if
(
pMnode
==
NULL
)
{
terrno
=
TSDB_CODE_DND_MNODE_NOT_DEPLOYED
;
dError
(
"failed to alter mnode since %s"
,
terrstr
());
return
-
1
;
}
dDebug
(
"start to alter mnode"
);
int32_t
code
=
mmAlter
(
pMgmt
,
&
option
);
mmRelease
(
pMgmt
,
pMnode
);
return
code
;
}
int32_t
mmProcessDropReq
(
SMnodeMgmt
*
pMgmt
,
SNodeMsg
*
pMsg
)
{
SDnode
*
pDnode
=
pMgmt
->
pDnode
;
SRpcMsg
*
pReq
=
&
pMsg
->
rpcMsg
;
SDDropMnodeReq
dropReq
=
{
0
};
if
(
tDeserializeSMCreateDropMnodeReq
(
pReq
->
pCont
,
pReq
->
contLen
,
&
dropReq
)
!=
0
)
{
terrno
=
TSDB_CODE_INVALID_MSG
;
return
-
1
;
}
if
(
dropReq
.
dnodeId
!=
pDnode
->
dnodeId
)
{
terrno
=
TSDB_CODE_DND_MNODE_INVALID_OPTION
;
dError
(
"failed to drop mnode since %s"
,
terrstr
());
return
-
1
;
}
SMnode
*
pMnode
=
mmAcquire
(
pMgmt
);
if
(
pMnode
==
NULL
)
{
terrno
=
TSDB_CODE_DND_MNODE_NOT_DEPLOYED
;
dError
(
"failed to drop mnode since %s"
,
terrstr
());
return
-
1
;
}
dDebug
(
"start to drop mnode"
);
int32_t
code
=
mmDrop
(
pMgmt
);
mmRelease
(
pMgmt
,
pMnode
);
return
code
;
}
void
mmInitMsgHandles
(
SMgmtWrapper
*
pWrapper
)
{
...
...
source/dnode/mgmt/mnode/src/mmWorker.c
浏览文件 @
4131fbea
...
...
@@ -18,27 +18,22 @@
static
void
mmProcessQueue
(
SMnodeMgmt
*
pMgmt
,
SNodeMsg
*
pMsg
)
{
dTrace
(
"msg:%p, will be processed in mnode queue"
,
pMsg
);
SMnode
*
pMnode
=
mmAcquire
(
pMgmt
);
SRpcMsg
*
pRpc
=
&
pMsg
->
rpcMsg
;
bool
isReq
=
(
pRpc
->
msgType
&
1U
);
int32_t
code
=
-
1
;
if
(
pM
node
!=
NULL
)
{
pMsg
->
pNode
=
pMnode
;
if
(
pM
sg
->
rpcMsg
.
msgType
!=
TDMT_DND_ALTER_MNODE
)
{
pMsg
->
pNode
=
pM
gmt
->
pM
node
;
code
=
mndProcessMsg
(
pMsg
);
mmRelease
(
pMgmt
,
pMnode
);
}
else
{
code
=
mmProcessAlterReq
(
pMgmt
,
pMsg
);
}
if
(
isReq
)
{
if
(
pMsg
->
rpcMsg
.
handle
==
NULL
)
return
;
if
(
code
==
0
)
{
if
(
pRpc
->
msgType
&
1U
)
{
if
(
pRpc
->
handle
==
NULL
)
return
;
if
(
code
!=
TSDB_CODE_MND_ACTION_IN_PROGRESS
)
{
if
(
code
!=
0
)
code
=
terrno
;
SRpcMsg
rsp
=
{.
handle
=
pRpc
->
handle
,
.
contLen
=
pMsg
->
rspLen
,
.
pCont
=
pMsg
->
pRsp
};
dndSendRsp
(
pMgmt
->
pWrapper
,
&
rsp
);
}
else
{
if
(
terrno
!=
TSDB_CODE_MND_ACTION_IN_PROGRESS
)
{
SRpcMsg
rsp
=
{.
handle
=
pRpc
->
handle
,
.
contLen
=
pMsg
->
rspLen
,
.
pCont
=
pMsg
->
pRsp
,
.
code
=
terrno
};
dndSendRsp
(
pMgmt
->
pWrapper
,
&
rsp
);
}
}
}
...
...
@@ -67,27 +62,14 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
}
void
mmStopWorker
(
SMnodeMgmt
*
pMgmt
)
{
taosWLockLatch
(
&
pMgmt
->
latch
);
pMgmt
->
deployed
=
0
;
taosWUnLockLatch
(
&
pMgmt
->
latch
);
while
(
pMgmt
->
refCount
>
1
)
{
taosMsleep
(
10
);
}
dndCleanupWorker
(
&
pMgmt
->
readWorker
);
dndCleanupWorker
(
&
pMgmt
->
writeWorker
);
dndCleanupWorker
(
&
pMgmt
->
syncWorker
);
}
static
int32_t
mmPutMsgToWorker
(
SMnodeMgmt
*
pMgmt
,
SDnodeWorker
*
pWorker
,
SNodeMsg
*
pMsg
)
{
SMnode
*
pMnode
=
mmAcquire
(
pMgmt
);
if
(
pMnode
==
NULL
)
return
-
1
;
dTrace
(
"msg:%p, put into worker %s"
,
pMsg
,
pWorker
->
name
);
int32_t
code
=
dndWriteMsgToWorker
(
pWorker
,
pMsg
);
mmRelease
(
pMgmt
,
pMnode
);
return
code
;
return
dndWriteMsgToWorker
(
pWorker
,
pMsg
);
}
int32_t
mmProcessWriteMsg
(
SMnodeMgmt
*
pMgmt
,
SNodeMsg
*
pMsg
)
{
...
...
source/dnode/mgmt/qnode/inc/qmInt.h
浏览文件 @
4131fbea
...
...
@@ -17,6 +17,7 @@
#define _TD_DND_QNODE_INT_H_
#include "qm.h"
#include "qnode.h"
#ifdef __cplusplus
extern
"C"
{
...
...
source/dnode/mgmt/qnode/src/qmInt.c
浏览文件 @
4131fbea
...
...
@@ -23,6 +23,7 @@ static void qmInitOption(SQnodeMgmt *pMgmt, SQnodeOpt *pOption) {
pOption
->
pWrapper
=
pMgmt
->
pWrapper
;
pOption
->
sendReqFp
=
dndSendReqToDnode
;
pOption
->
sendMnodeReqFp
=
dndSendReqToMnode
;
pOption
->
sendRspFp
=
dndSendRsp
;
pOption
->
dnodeId
=
pDnode
->
dnodeId
;
pOption
->
clusterId
=
pDnode
->
clusterId
;
}
...
...
source/dnode/mgmt/snode/inc/smInt.h
浏览文件 @
4131fbea
...
...
@@ -17,6 +17,7 @@
#define _TD_DND_SNODE_INT_H_
#include "sm.h"
#include "snode.h"
#ifdef __cplusplus
extern
"C"
{
...
...
source/dnode/mgmt/snode/src/smInt.c
浏览文件 @
4131fbea
...
...
@@ -23,6 +23,7 @@ static void smInitOption(SSnodeMgmt *pMgmt, SSnodeOpt *pOption) {
pOption
->
pWrapper
=
pMgmt
->
pWrapper
;
pOption
->
sendReqFp
=
dndSendReqToDnode
;
pOption
->
sendMnodeReqFp
=
dndSendReqToMnode
;
pOption
->
sendRspFp
=
dndSendRsp
;
pOption
->
dnodeId
=
pDnode
->
dnodeId
;
pOption
->
clusterId
=
pDnode
->
clusterId
;
}
...
...
source/dnode/mgmt/test/mnode/dmnode.cpp
浏览文件 @
4131fbea
...
...
@@ -40,7 +40,7 @@ TEST_F(DndTestMnode, 01_Create_Mnode) {
SRpcMsg
*
pRsp
=
test
.
SendReq
(
TDMT_DND_CREATE_MNODE
,
pReq
,
contLen
);
ASSERT_NE
(
pRsp
,
nullptr
);
ASSERT_EQ
(
pRsp
->
code
,
TSDB_CODE_
DND_M
NODE_INVALID_OPTION
);
ASSERT_EQ
(
pRsp
->
code
,
TSDB_CODE_NODE_INVALID_OPTION
);
}
{
...
...
@@ -57,7 +57,7 @@ TEST_F(DndTestMnode, 01_Create_Mnode) {
SRpcMsg
*
pRsp
=
test
.
SendReq
(
TDMT_DND_CREATE_MNODE
,
pReq
,
contLen
);
ASSERT_NE
(
pRsp
,
nullptr
);
ASSERT_EQ
(
pRsp
->
code
,
TSDB_CODE_
DND_M
NODE_INVALID_OPTION
);
ASSERT_EQ
(
pRsp
->
code
,
TSDB_CODE_NODE_INVALID_OPTION
);
}
{
...
...
@@ -77,7 +77,7 @@ TEST_F(DndTestMnode, 01_Create_Mnode) {
SRpcMsg
*
pRsp
=
test
.
SendReq
(
TDMT_DND_CREATE_MNODE
,
pReq
,
contLen
);
ASSERT_NE
(
pRsp
,
nullptr
);
ASSERT_EQ
(
pRsp
->
code
,
TSDB_CODE_
DND_M
NODE_ALREADY_DEPLOYED
);
ASSERT_EQ
(
pRsp
->
code
,
TSDB_CODE_NODE_ALREADY_DEPLOYED
);
}
}
...
...
@@ -96,7 +96,7 @@ TEST_F(DndTestMnode, 02_Alter_Mnode) {
SRpcMsg
*
pRsp
=
test
.
SendReq
(
TDMT_DND_ALTER_MNODE
,
pReq
,
contLen
);
ASSERT_NE
(
pRsp
,
nullptr
);
ASSERT_EQ
(
pRsp
->
code
,
TSDB_CODE_
DND_M
NODE_INVALID_OPTION
);
ASSERT_EQ
(
pRsp
->
code
,
TSDB_CODE_NODE_INVALID_OPTION
);
}
{
...
...
@@ -113,7 +113,7 @@ TEST_F(DndTestMnode, 02_Alter_Mnode) {
SRpcMsg
*
pRsp
=
test
.
SendReq
(
TDMT_DND_ALTER_MNODE
,
pReq
,
contLen
);
ASSERT_NE
(
pRsp
,
nullptr
);
ASSERT_EQ
(
pRsp
->
code
,
TSDB_CODE_
DND_M
NODE_INVALID_OPTION
);
ASSERT_EQ
(
pRsp
->
code
,
TSDB_CODE_NODE_INVALID_OPTION
);
}
{
...
...
@@ -145,7 +145,7 @@ TEST_F(DndTestMnode, 03_Drop_Mnode) {
SRpcMsg
*
pRsp
=
test
.
SendReq
(
TDMT_DND_DROP_MNODE
,
pReq
,
contLen
);
ASSERT_NE
(
pRsp
,
nullptr
);
ASSERT_EQ
(
pRsp
->
code
,
TSDB_CODE_
DND_M
NODE_INVALID_OPTION
);
ASSERT_EQ
(
pRsp
->
code
,
TSDB_CODE_NODE_INVALID_OPTION
);
}
{
...
...
@@ -171,7 +171,7 @@ TEST_F(DndTestMnode, 03_Drop_Mnode) {
SRpcMsg
*
pRsp
=
test
.
SendReq
(
TDMT_DND_DROP_MNODE
,
pReq
,
contLen
);
ASSERT_NE
(
pRsp
,
nullptr
);
ASSERT_EQ
(
pRsp
->
code
,
TSDB_CODE_
DND_M
NODE_NOT_DEPLOYED
);
ASSERT_EQ
(
pRsp
->
code
,
TSDB_CODE_NODE_NOT_DEPLOYED
);
}
{
...
...
@@ -188,7 +188,7 @@ TEST_F(DndTestMnode, 03_Drop_Mnode) {
SRpcMsg
*
pRsp
=
test
.
SendReq
(
TDMT_DND_ALTER_MNODE
,
pReq
,
contLen
);
ASSERT_NE
(
pRsp
,
nullptr
);
ASSERT_EQ
(
pRsp
->
code
,
TSDB_CODE_
DND_M
NODE_NOT_DEPLOYED
);
ASSERT_EQ
(
pRsp
->
code
,
TSDB_CODE_NODE_NOT_DEPLOYED
);
}
{
...
...
source/dnode/mgmt/vnode/inc/vmInt.h
浏览文件 @
4131fbea
...
...
@@ -16,9 +16,9 @@
#ifndef _TD_DND_VNODES_INT_H_
#define _TD_DND_VNODES_INT_H_
#include "dm.h"
#include "sync.h"
#include "vm.h"
#include "vnode.h"
#ifdef __cplusplus
extern
"C"
{
...
...
source/dnode/mgmt/vnode/src/vmFile.c
浏览文件 @
4131fbea
...
...
@@ -45,7 +45,7 @@ SVnodeObj **vmGetVnodesFromHash(SVnodesMgmt *pMgmt, int32_t *numOfVnodes) {
}
int32_t
vmGetVnodesFromFile
(
SVnodesMgmt
*
pMgmt
,
SWrapperCfg
**
ppCfgs
,
int32_t
*
numOfVnodes
)
{
int32_t
code
=
TSDB_CODE_
DND_VNODE_READ
_FILE_ERROR
;
int32_t
code
=
TSDB_CODE_
NODE_PARSE
_FILE_ERROR
;
int32_t
len
=
0
;
int32_t
maxLen
=
30000
;
char
*
content
=
calloc
(
1
,
maxLen
+
1
);
...
...
source/dnode/mnode/impl/src/mndMnode.c
浏览文件 @
4131fbea
...
...
@@ -320,7 +320,7 @@ static int32_t mndSetCreateMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDno
action
.
pCont
=
pReq
;
action
.
contLen
=
contLen
;
action
.
msgType
=
TDMT_DND_ALTER_MNODE
;
action
.
acceptableCode
=
TSDB_CODE_
DND_M
NODE_ALREADY_DEPLOYED
;
action
.
acceptableCode
=
TSDB_CODE_NODE_ALREADY_DEPLOYED
;
if
(
mndTransAppendRedoAction
(
pTrans
,
&
action
)
!=
0
)
{
free
(
pReq
);
...
...
@@ -345,7 +345,7 @@ static int32_t mndSetCreateMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDno
action
.
pCont
=
pReq
;
action
.
contLen
=
contLen
;
action
.
msgType
=
TDMT_DND_CREATE_MNODE
;
action
.
acceptableCode
=
TSDB_CODE_
DND_M
NODE_ALREADY_DEPLOYED
;
action
.
acceptableCode
=
TSDB_CODE_NODE_ALREADY_DEPLOYED
;
if
(
mndTransAppendRedoAction
(
pTrans
,
&
action
)
!=
0
)
{
free
(
pReq
);
return
-
1
;
...
...
@@ -490,7 +490,7 @@ static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnode
action
.
pCont
=
pReq
;
action
.
contLen
=
contLen
;
action
.
msgType
=
TDMT_DND_ALTER_MNODE
;
action
.
acceptableCode
=
TSDB_CODE_
DND_M
NODE_ALREADY_DEPLOYED
;
action
.
acceptableCode
=
TSDB_CODE_NODE_ALREADY_DEPLOYED
;
if
(
mndTransAppendRedoAction
(
pTrans
,
&
action
)
!=
0
)
{
free
(
pReq
);
...
...
@@ -517,7 +517,7 @@ static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnode
action
.
pCont
=
pReq
;
action
.
contLen
=
contLen
;
action
.
msgType
=
TDMT_DND_DROP_MNODE
;
action
.
acceptableCode
=
TSDB_CODE_
DND_M
NODE_NOT_DEPLOYED
;
action
.
acceptableCode
=
TSDB_CODE_NODE_NOT_DEPLOYED
;
if
(
mndTransAppendRedoAction
(
pTrans
,
&
action
)
!=
0
)
{
free
(
pReq
);
return
-
1
;
...
...
source/util/src/terror.c
浏览文件 @
4131fbea
...
...
@@ -69,10 +69,6 @@ TAOS_DEFINE_ERROR(TSDB_CODE_APP_NOT_READY, "Database not ready")
TAOS_DEFINE_ERROR
(
TSDB_CODE_RPC_FQDN_ERROR
,
"Unable to resolve FQDN"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_RPC_INVALID_VERSION
,
"Invalid app version"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_COMPRESS_ERROR
,
"Failed to compress msg"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_NODE_ALREADY_DEPLOYED
,
"Node already deployed"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_NODE_NOT_DEPLOYED
,
"Node not deployed"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_NODE_PARSE_FILE_ERROR
,
"Invalid json format"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_NODE_INVALID_OPTION
,
"Invalid node option"
)
//common & util
TAOS_DEFINE_ERROR
(
TSDB_CODE_OPS_NOT_SUPPORT
,
"Operation not supported"
)
...
...
@@ -278,18 +274,13 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_UNSUPPORTED_TOPIC, "Topic with aggregatio
TAOS_DEFINE_ERROR
(
TSDB_CODE_DND_ACTION_IN_PROGRESS
,
"Action in progress"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_DND_OFFLINE
,
"Dnode is offline"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_DND_INVALID_MSG_LEN
,
"Invalid message length"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_DND_DNODE_READ_FILE_ERROR
,
"Read dnode.json error"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_DND_DNODE_WRITE_FILE_ERROR
,
"Write dnode.json error"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED
,
"Mnode already deployed"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_DND_MNODE_NOT_DEPLOYED
,
"Mnode not deployed"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_DND_MNODE_INVALID_OPTION
,
"Mnode option invalid"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_DND_MNODE_READ_FILE_ERROR
,
"Read mnode.json error"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_DND_MNODE_WRITE_FILE_ERROR
,
"Write mnode.json error"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_NODE_ALREADY_DEPLOYED
,
"Node already deployed"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_NODE_NOT_DEPLOYED
,
"Node not deployed"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_NODE_PARSE_FILE_ERROR
,
"Invalid json format"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_NODE_INVALID_OPTION
,
"Invalid node option"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_DND_VNODE_ALREADY_DEPLOYED
,
"Vnode already deployed"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_DND_VNODE_NOT_DEPLOYED
,
"Vnode not deployed"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_DND_VNODE_INVALID_OPTION
,
"Vnode option invalid"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_DND_VNODE_READ_FILE_ERROR
,
"Read vnodes.json error"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_DND_VNODE_WRITE_FILE_ERROR
,
"Write vnodes.json error"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_DND_VNODE_TOO_MANY_VNODES
,
"Too many vnodes"
)
// vnode
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录