Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
ce79f915
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1185
Star
22015
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看板
体验新版 GitCode,发现更多精彩内容 >>
You need to sign in or sign up before continuing.
提交
ce79f915
编写于
9月 22, 2022
作者:
S
Shengliang Guan
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
fix: restore vnodes in multi-threads
上级
3c65b2d8
变更
2
隐藏空白更改
内联
并排
Showing
2 changed file
with
84 addition
and
13 deletion
+84
-13
source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
+1
-0
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
+83
-13
未找到文件。
source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
浏览文件 @
ce79f915
...
...
@@ -74,6 +74,7 @@ typedef struct {
TdThread
thread
;
SVnodeMgmt
*
pMgmt
;
SWrapperCfg
*
pCfgs
;
SVnodeObj
**
ppVnodes
;
}
SVnodeThread
;
// vmInt.c
...
...
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
浏览文件 @
ce79f915
...
...
@@ -218,14 +218,14 @@ static void vmCloseVnodes(SVnodeMgmt *pMgmt) {
dInfo
(
"start to close all vnodes"
);
int32_t
numOfVnodes
=
0
;
SVnodeObj
**
pVnodes
=
vmGetVnodeListFromHash
(
pMgmt
,
&
numOfVnodes
);
SVnodeObj
**
p
p
Vnodes
=
vmGetVnodeListFromHash
(
pMgmt
,
&
numOfVnodes
);
for
(
int32_t
i
=
0
;
i
<
numOfVnodes
;
++
i
)
{
vmCloseVnode
(
pMgmt
,
pVnodes
[
i
]);
vmCloseVnode
(
pMgmt
,
p
p
Vnodes
[
i
]);
}
if
(
pVnodes
!=
NULL
)
{
taosMemoryFree
(
pVnodes
);
if
(
p
p
Vnodes
!=
NULL
)
{
taosMemoryFree
(
p
p
Vnodes
);
}
if
(
pMgmt
->
hash
!=
NULL
)
{
...
...
@@ -331,22 +331,92 @@ static int32_t vmRequire(const SMgmtInputOpt *pInput, bool *required) {
return
0
;
}
static
int32_t
vmStart
(
SVnodeMgmt
*
pMgmt
)
{
static
void
*
vmRestoreVnodeInThread
(
void
*
param
)
{
SVnodeThread
*
pThread
=
param
;
SVnodeMgmt
*
pMgmt
=
pThread
->
pMgmt
;
dInfo
(
"thread:%d, start to restore %d vnodes"
,
pThread
->
threadIndex
,
pThread
->
vnodeNum
);
setThreadName
(
"restore-vnodes"
);
for
(
int32_t
v
=
0
;
v
<
pThread
->
vnodeNum
;
++
v
)
{
SVnodeObj
*
pVnode
=
pThread
->
ppVnodes
[
v
];
char
stepDesc
[
TSDB_STEP_DESC_LEN
]
=
{
0
};
snprintf
(
stepDesc
,
TSDB_STEP_DESC_LEN
,
"vgId:%d, start to restore, %d of %d have been restored"
,
pVnode
->
vgId
,
pMgmt
->
state
.
openVnodes
,
pMgmt
->
state
.
totalVnodes
);
tmsgReportStartup
(
"vnode-restore"
,
stepDesc
);
int32_t
code
=
vnodeStart
(
pVnode
->
pImpl
);
if
(
code
!=
0
)
{
dError
(
"vgId:%d, failed to restore vnode by thread:%d"
,
pVnode
->
vgId
,
pThread
->
threadIndex
);
pThread
->
failed
++
;
}
else
{
dDebug
(
"vgId:%d, is restored by thread:%d"
,
pVnode
->
vgId
,
pThread
->
threadIndex
);
pThread
->
opened
++
;
atomic_add_fetch_32
(
&
pMgmt
->
state
.
openVnodes
,
1
);
}
}
dInfo
(
"thread:%d, numOfVnodes:%d, restored:%d failed:%d"
,
pThread
->
threadIndex
,
pThread
->
vnodeNum
,
pThread
->
opened
,
pThread
->
failed
);
return
NULL
;
}
static
int32_t
vmStartVnodes
(
SVnodeMgmt
*
pMgmt
)
{
int32_t
numOfVnodes
=
0
;
SVnodeObj
**
pVnodes
=
vmGetVnodeListFromHash
(
pMgmt
,
&
numOfVnodes
);
SVnodeObj
**
p
p
Vnodes
=
vmGetVnodeListFromHash
(
pMgmt
,
&
numOfVnodes
);
for
(
int32_t
i
=
0
;
i
<
numOfVnodes
;
++
i
)
{
SVnodeObj
*
pVnode
=
pVnodes
[
i
];
vnodeStart
(
pVnode
->
pImpl
);
int32_t
threadNum
=
tsNumOfCores
/
2
;
if
(
threadNum
<
1
)
threadNum
=
1
;
int32_t
vnodesPerThread
=
numOfVnodes
/
threadNum
+
1
;
SVnodeThread
*
threads
=
taosMemoryCalloc
(
threadNum
,
sizeof
(
SVnodeThread
));
for
(
int32_t
t
=
0
;
t
<
threadNum
;
++
t
)
{
threads
[
t
].
threadIndex
=
t
;
threads
[
t
].
pMgmt
=
pMgmt
;
threads
[
t
].
ppVnodes
=
taosMemoryCalloc
(
vnodesPerThread
,
sizeof
(
SVnode
*
));
}
for
(
int32_t
v
=
0
;
v
<
numOfVnodes
;
++
v
)
{
int32_t
t
=
v
%
threadNum
;
SVnodeThread
*
pThread
=
&
threads
[
t
];
pThread
->
ppVnodes
[
pThread
->
vnodeNum
++
]
=
ppVnodes
[
v
];
}
pMgmt
->
state
.
openVnodes
=
0
;
dInfo
(
"restore %d vnodes with %d threads"
,
numOfVnodes
,
threadNum
);
for
(
int32_t
t
=
0
;
t
<
threadNum
;
++
t
)
{
SVnodeThread
*
pThread
=
&
threads
[
t
];
if
(
pThread
->
vnodeNum
==
0
)
continue
;
TdThreadAttr
thAttr
;
taosThreadAttrInit
(
&
thAttr
);
taosThreadAttrSetDetachState
(
&
thAttr
,
PTHREAD_CREATE_JOINABLE
);
if
(
taosThreadCreate
(
&
pThread
->
thread
,
&
thAttr
,
vmRestoreVnodeInThread
,
pThread
)
!=
0
)
{
dError
(
"thread:%d, failed to create thread to restore vnode since %s"
,
pThread
->
threadIndex
,
strerror
(
errno
));
}
taosThreadAttrDestroy
(
&
thAttr
);
}
for
(
int32_t
t
=
0
;
t
<
threadNum
;
++
t
)
{
SVnodeThread
*
pThread
=
&
threads
[
t
];
if
(
pThread
->
vnodeNum
>
0
&&
taosCheckPthreadValid
(
pThread
->
thread
))
{
taosThreadJoin
(
pThread
->
thread
,
NULL
);
taosThreadClear
(
&
pThread
->
thread
);
}
taosMemoryFree
(
pThread
->
pCfgs
);
}
taosMemoryFree
(
threads
);
for
(
int32_t
i
=
0
;
i
<
numOfVnodes
;
++
i
)
{
SVnodeObj
*
pVnode
=
pVnodes
[
i
];
SVnodeObj
*
pVnode
=
p
p
Vnodes
[
i
];
vmReleaseVnode
(
pMgmt
,
pVnode
);
}
if
(
pVnodes
!=
NULL
)
{
taosMemoryFree
(
pVnodes
);
if
(
p
p
Vnodes
!=
NULL
)
{
taosMemoryFree
(
p
p
Vnodes
);
}
return
0
;
...
...
@@ -360,7 +430,7 @@ SMgmtFunc vmGetMgmtFunc() {
SMgmtFunc
mgmtFunc
=
{
0
};
mgmtFunc
.
openFp
=
vmInit
;
mgmtFunc
.
closeFp
=
(
NodeCloseFp
)
vmCleanup
;
mgmtFunc
.
startFp
=
(
NodeStartFp
)
vmStart
;
mgmtFunc
.
startFp
=
(
NodeStartFp
)
vmStart
Vnodes
;
mgmtFunc
.
stopFp
=
(
NodeStopFp
)
vmStop
;
mgmtFunc
.
requiredFp
=
vmRequire
;
mgmtFunc
.
getHandlesFp
=
vmGetMsgHandles
;
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录