Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
哥们要飞
TDengine
提交
2a4feceb
T
TDengine
项目概览
哥们要飞
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
1
Star
1
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,发现更多精彩内容 >>
提交
2a4feceb
编写于
5月 23, 2020
作者:
陶建辉(Jeff)
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
tune the code to make the vnode cleanup process more clear
上级
c733b76b
变更
5
隐藏空白更改
内联
并排
Showing
5 changed file
with
115 addition
and
63 deletion
+115
-63
src/cq/src/cqMain.c
src/cq/src/cqMain.c
+7
-2
src/inc/tsync.h
src/inc/tsync.h
+2
-2
src/util/src/tqueue.c
src/util/src/tqueue.c
+2
-0
src/vnode/src/vnodeMain.c
src/vnode/src/vnodeMain.c
+94
-57
src/wal/src/walMain.c
src/wal/src/walMain.c
+10
-2
未找到文件。
src/cq/src/cqMain.c
浏览文件 @
2a4feceb
...
@@ -18,6 +18,7 @@
...
@@ -18,6 +18,7 @@
#include <stdlib.h>
#include <stdlib.h>
#include <string.h>
#include <string.h>
#include <pthread.h>
#include <pthread.h>
#include <errno.h>
#include "taosdef.h"
#include "taosdef.h"
#include "taosmsg.h"
#include "taosmsg.h"
#include "tglobal.h"
#include "tglobal.h"
...
@@ -64,7 +65,10 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj);
...
@@ -64,7 +65,10 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj);
void
*
cqOpen
(
void
*
ahandle
,
const
SCqCfg
*
pCfg
)
{
void
*
cqOpen
(
void
*
ahandle
,
const
SCqCfg
*
pCfg
)
{
SCqContext
*
pContext
=
calloc
(
sizeof
(
SCqContext
),
1
);
SCqContext
*
pContext
=
calloc
(
sizeof
(
SCqContext
),
1
);
if
(
pContext
==
NULL
)
return
NULL
;
if
(
pContext
==
NULL
)
{
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
return
NULL
;
}
strcpy
(
pContext
->
user
,
pCfg
->
user
);
strcpy
(
pContext
->
user
,
pCfg
->
user
);
strcpy
(
pContext
->
pass
,
pCfg
->
pass
);
strcpy
(
pContext
->
pass
,
pCfg
->
pass
);
...
@@ -82,6 +86,7 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
...
@@ -82,6 +86,7 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
void
cqClose
(
void
*
handle
)
{
void
cqClose
(
void
*
handle
)
{
SCqContext
*
pContext
=
handle
;
SCqContext
*
pContext
=
handle
;
if
(
handle
==
NULL
)
return
;
// stop all CQs
// stop all CQs
cqStop
(
pContext
);
cqStop
(
pContext
);
...
@@ -106,9 +111,9 @@ void cqClose(void *handle) {
...
@@ -106,9 +111,9 @@ void cqClose(void *handle) {
void
cqStart
(
void
*
handle
)
{
void
cqStart
(
void
*
handle
)
{
SCqContext
*
pContext
=
handle
;
SCqContext
*
pContext
=
handle
;
cTrace
(
"vgId:%d, start all CQs"
,
pContext
->
vgId
);
if
(
pContext
->
dbConn
||
pContext
->
master
)
return
;
if
(
pContext
->
dbConn
||
pContext
->
master
)
return
;
cTrace
(
"vgId:%d, start all CQs"
,
pContext
->
vgId
);
pthread_mutex_lock
(
&
pContext
->
mutex
);
pthread_mutex_lock
(
&
pContext
->
mutex
);
pContext
->
master
=
1
;
pContext
->
master
=
1
;
...
...
src/inc/tsync.h
浏览文件 @
2a4feceb
...
@@ -94,8 +94,8 @@ typedef void* tsync_h;
...
@@ -94,8 +94,8 @@ typedef void* tsync_h;
tsync_h
syncStart
(
const
SSyncInfo
*
);
tsync_h
syncStart
(
const
SSyncInfo
*
);
void
syncStop
(
tsync_h
shandle
);
void
syncStop
(
tsync_h
shandle
);
int
syncReconfig
(
tsync_h
shandle
,
const
SSyncCfg
*
);
int
32_t
syncReconfig
(
tsync_h
shandle
,
const
SSyncCfg
*
);
int
syncForwardToPeer
(
tsync_h
shandle
,
void
*
pHead
,
void
*
mhandle
,
int
qtype
);
int
32_t
syncForwardToPeer
(
tsync_h
shandle
,
void
*
pHead
,
void
*
mhandle
,
int
qtype
);
void
syncConfirmForward
(
tsync_h
shandle
,
uint64_t
version
,
int32_t
code
);
void
syncConfirmForward
(
tsync_h
shandle
,
uint64_t
version
,
int32_t
code
);
void
syncRecover
(
tsync_h
shandle
);
// recover from other nodes:
void
syncRecover
(
tsync_h
shandle
);
// recover from other nodes:
int
syncGetNodesRole
(
tsync_h
shandle
,
SNodesRole
*
);
int
syncGetNodesRole
(
tsync_h
shandle
,
SNodesRole
*
);
...
...
src/util/src/tqueue.c
浏览文件 @
2a4feceb
...
@@ -65,6 +65,7 @@ taos_queue taosOpenQueue() {
...
@@ -65,6 +65,7 @@ taos_queue taosOpenQueue() {
}
}
void
taosCloseQueue
(
taos_queue
param
)
{
void
taosCloseQueue
(
taos_queue
param
)
{
if
(
param
==
NULL
)
return
;
STaosQueue
*
queue
=
(
STaosQueue
*
)
param
;
STaosQueue
*
queue
=
(
STaosQueue
*
)
param
;
STaosQnode
*
pTemp
;
STaosQnode
*
pTemp
;
STaosQnode
*
pNode
=
queue
->
head
;
STaosQnode
*
pNode
=
queue
->
head
;
...
@@ -224,6 +225,7 @@ taos_qset taosOpenQset() {
...
@@ -224,6 +225,7 @@ taos_qset taosOpenQset() {
}
}
void
taosCloseQset
(
taos_qset
param
)
{
void
taosCloseQset
(
taos_qset
param
)
{
if
(
param
==
NULL
)
return
;
STaosQset
*
qset
=
(
STaosQset
*
)
param
;
STaosQset
*
qset
=
(
STaosQset
*
)
param
;
pthread_mutex_destroy
(
&
qset
->
mutex
);
pthread_mutex_destroy
(
&
qset
->
mutex
);
tsem_destroy
(
&
qset
->
sem
);
tsem_destroy
(
&
qset
->
sem
);
...
...
src/vnode/src/vnodeMain.c
浏览文件 @
2a4feceb
...
@@ -35,7 +35,7 @@ static void vnodeCleanUp(SVnodeObj *pVnode);
...
@@ -35,7 +35,7 @@ static void vnodeCleanUp(SVnodeObj *pVnode);
static
int32_t
vnodeSaveCfg
(
SMDCreateVnodeMsg
*
pVnodeCfg
);
static
int32_t
vnodeSaveCfg
(
SMDCreateVnodeMsg
*
pVnodeCfg
);
static
int32_t
vnodeReadCfg
(
SVnodeObj
*
pVnode
);
static
int32_t
vnodeReadCfg
(
SVnodeObj
*
pVnode
);
static
int32_t
vnodeSaveVersion
(
SVnodeObj
*
pVnode
);
static
int32_t
vnodeSaveVersion
(
SVnodeObj
*
pVnode
);
static
bool
vnodeReadVersion
(
SVnodeObj
*
pVnode
);
static
int32_t
vnodeReadVersion
(
SVnodeObj
*
pVnode
);
static
int
vnodeProcessTsdbStatus
(
void
*
arg
,
int
status
);
static
int
vnodeProcessTsdbStatus
(
void
*
arg
,
int
status
);
static
uint32_t
vnodeGetFileInfo
(
void
*
ahandle
,
char
*
name
,
uint32_t
*
index
,
int32_t
*
size
,
uint64_t
*
fversion
);
static
uint32_t
vnodeGetFileInfo
(
void
*
ahandle
,
char
*
name
,
uint32_t
*
index
,
int32_t
*
size
,
uint64_t
*
fversion
);
static
int
vnodeGetWalInfo
(
void
*
ahandle
,
char
*
name
,
uint32_t
*
index
);
static
int
vnodeGetWalInfo
(
void
*
ahandle
,
char
*
name
,
uint32_t
*
index
);
...
@@ -46,9 +46,9 @@ static pthread_once_t vnodeModuleInit = PTHREAD_ONCE_INIT;
...
@@ -46,9 +46,9 @@ static pthread_once_t vnodeModuleInit = PTHREAD_ONCE_INIT;
#ifndef _SYNC
#ifndef _SYNC
tsync_h
syncStart
(
const
SSyncInfo
*
info
)
{
return
NULL
;
}
tsync_h
syncStart
(
const
SSyncInfo
*
info
)
{
return
NULL
;
}
int
syncForwardToPeer
(
tsync_h
shandle
,
void
*
pHead
,
void
*
mhandle
,
int
qtype
)
{
return
0
;
}
int
32_t
syncForwardToPeer
(
tsync_h
shandle
,
void
*
pHead
,
void
*
mhandle
,
int
qtype
)
{
return
0
;
}
void
syncStop
(
tsync_h
shandle
)
{}
void
syncStop
(
tsync_h
shandle
)
{}
int
syncReconfig
(
tsync_h
shandle
,
const
SSyncCfg
*
cfg
)
{
return
0
;
}
int
32_t
syncReconfig
(
tsync_h
shandle
,
const
SSyncCfg
*
cfg
)
{
return
0
;
}
int
syncGetNodesRole
(
tsync_h
shandle
,
SNodesRole
*
cfg
)
{
return
0
;
}
int
syncGetNodesRole
(
tsync_h
shandle
,
SNodesRole
*
cfg
)
{
return
0
;
}
void
syncConfirmForward
(
tsync_h
shandle
,
uint64_t
version
,
int32_t
code
)
{}
void
syncConfirmForward
(
tsync_h
shandle
,
uint64_t
version
,
int32_t
code
)
{}
#endif
#endif
...
@@ -185,26 +185,40 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
...
@@ -185,26 +185,40 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
pthread_once
(
&
vnodeModuleInit
,
vnodeInit
);
pthread_once
(
&
vnodeModuleInit
,
vnodeInit
);
SVnodeObj
*
pVnode
=
calloc
(
sizeof
(
SVnodeObj
),
1
);
SVnodeObj
*
pVnode
=
calloc
(
sizeof
(
SVnodeObj
),
1
);
if
(
pVnode
==
NULL
)
{
vError
(
"vgId:%d, failed to open vnode since no enough memory"
,
vnode
);
return
TAOS_SYSTEM_ERROR
(
errno
);
}
atomic_add_fetch_32
(
&
tsOpennedVnodes
,
1
);
atomic_add_fetch_32
(
&
pVnode
->
refCount
,
1
);
pVnode
->
vgId
=
vnode
;
pVnode
->
vgId
=
vnode
;
pVnode
->
status
=
TAOS_VN_STATUS_INIT
;
pVnode
->
status
=
TAOS_VN_STATUS_INIT
;
pVnode
->
refCount
=
1
;
pVnode
->
version
=
0
;
pVnode
->
version
=
0
;
pVnode
->
tsdbCfg
.
tsdbId
=
pVnode
->
vgId
;
pVnode
->
tsdbCfg
.
tsdbId
=
pVnode
->
vgId
;
pVnode
->
rootDir
=
strdup
(
rootDir
);
pVnode
->
rootDir
=
strdup
(
rootDir
);
taosHashPut
(
tsDnodeVnodesHash
,
(
const
char
*
)
&
pVnode
->
vgId
,
sizeof
(
int32_t
),
(
char
*
)(
&
pVnode
),
sizeof
(
SVnodeObj
*
));
int32_t
code
=
vnodeReadCfg
(
pVnode
);
int32_t
code
=
vnodeReadCfg
(
pVnode
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
vError
(
"vgId:%d, failed to read cfg file"
,
pVnode
->
vgId
);
vnodeCleanUp
(
pVnode
);
taosHashRemove
(
tsDnodeVnodesHash
,
(
const
char
*
)
&
pVnode
->
vgId
,
sizeof
(
int32_t
));
return
code
;
}
code
=
vnodeReadVersion
(
pVnode
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
vnodeCleanUp
(
pVnode
);
return
code
;
return
code
;
}
}
vnodeReadVersion
(
pVnode
);
pVnode
->
fversion
=
pVnode
->
version
;
pVnode
->
fversion
=
pVnode
->
version
;
pVnode
->
wqueue
=
dnodeAllocateWqueue
(
pVnode
);
pVnode
->
wqueue
=
dnodeAllocateWqueue
(
pVnode
);
pVnode
->
rqueue
=
dnodeAllocateRqueue
(
pVnode
);
pVnode
->
rqueue
=
dnodeAllocateRqueue
(
pVnode
);
if
(
pVnode
->
wqueue
==
NULL
||
pVnode
->
rqueue
==
NULL
)
{
vnodeCleanUp
(
pVnode
);
return
terrno
;
}
SCqCfg
cqCfg
=
{
0
};
SCqCfg
cqCfg
=
{
0
};
sprintf
(
cqCfg
.
user
,
"root"
);
sprintf
(
cqCfg
.
user
,
"root"
);
...
@@ -212,22 +226,29 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
...
@@ -212,22 +226,29 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
cqCfg
.
vgId
=
vnode
;
cqCfg
.
vgId
=
vnode
;
cqCfg
.
cqWrite
=
vnodeWriteToQueue
;
cqCfg
.
cqWrite
=
vnodeWriteToQueue
;
pVnode
->
cq
=
cqOpen
(
pVnode
,
&
cqCfg
);
pVnode
->
cq
=
cqOpen
(
pVnode
,
&
cqCfg
);
if
(
pVnode
->
cq
==
NULL
)
{
vnodeCleanUp
(
pVnode
);
return
terrno
;
}
STsdbAppH
appH
=
{
0
};
STsdbAppH
appH
=
{
0
};
appH
.
appH
=
(
void
*
)
pVnode
;
appH
.
appH
=
(
void
*
)
pVnode
;
appH
.
notifyStatus
=
vnodeProcessTsdbStatus
;
appH
.
notifyStatus
=
vnodeProcessTsdbStatus
;
appH
.
cqH
=
pVnode
->
cq
;
appH
.
cqH
=
pVnode
->
cq
;
sprintf
(
temp
,
"%s/tsdb"
,
rootDir
);
sprintf
(
temp
,
"%s/tsdb"
,
rootDir
);
pVnode
->
tsdb
=
tsdbOpenRepo
(
temp
,
&
appH
);
pVnode
->
tsdb
=
tsdbOpenRepo
(
temp
,
&
appH
);
if
(
pVnode
->
tsdb
==
NULL
)
{
if
(
pVnode
->
tsdb
==
NULL
)
{
vError
(
"vgId:%d, failed to open tsdb at %s(%s)"
,
pVnode
->
vgId
,
temp
,
tstrerror
(
terrno
));
vnodeCleanUp
(
pVnode
);
taosHashRemove
(
tsDnodeVnodesHash
,
(
const
char
*
)
&
pVnode
->
vgId
,
sizeof
(
int32_t
));
return
terrno
;
return
terrno
;
}
}
sprintf
(
temp
,
"%s/wal"
,
rootDir
);
sprintf
(
temp
,
"%s/wal"
,
rootDir
);
pVnode
->
wal
=
walOpen
(
temp
,
&
pVnode
->
walCfg
);
pVnode
->
wal
=
walOpen
(
temp
,
&
pVnode
->
walCfg
);
if
(
pVnode
->
wal
==
NULL
)
{
vnodeCleanUp
(
pVnode
);
return
terrno
;
}
walRestore
(
pVnode
->
wal
,
pVnode
,
vnodeWriteToQueue
);
walRestore
(
pVnode
->
wal
,
pVnode
,
vnodeWriteToQueue
);
SSyncInfo
syncInfo
;
SSyncInfo
syncInfo
;
...
@@ -243,6 +264,10 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
...
@@ -243,6 +264,10 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
syncInfo
.
notifyRole
=
vnodeNotifyRole
;
syncInfo
.
notifyRole
=
vnodeNotifyRole
;
syncInfo
.
notifyFileSynced
=
vnodeNotifyFileSynced
;
syncInfo
.
notifyFileSynced
=
vnodeNotifyFileSynced
;
pVnode
->
sync
=
syncStart
(
&
syncInfo
);
pVnode
->
sync
=
syncStart
(
&
syncInfo
);
if
(
pVnode
->
sync
==
NULL
)
{
vnodeCleanUp
(
pVnode
);
return
terrno
;
}
#ifndef _SYNC
#ifndef _SYNC
pVnode
->
role
=
TAOS_SYNC_ROLE_MASTER
;
pVnode
->
role
=
TAOS_SYNC_ROLE_MASTER
;
...
@@ -253,11 +278,11 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
...
@@ -253,11 +278,11 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
cqStart
(
pVnode
->
cq
);
cqStart
(
pVnode
->
cq
);
pVnode
->
events
=
NULL
;
pVnode
->
events
=
NULL
;
pVnode
->
status
=
TAOS_VN_STATUS_READY
;
pVnode
->
status
=
TAOS_VN_STATUS_READY
;
vTrace
(
"vgId:%d, vnode is opened in %s, pVnode:%p"
,
pVnode
->
vgId
,
rootDir
,
pVnode
);
vTrace
(
"vgId:%d, vnode is opened in %s, pVnode:%p"
,
pVnode
->
vgId
,
rootDir
,
pVnode
);
atomic_add_fetch_32
(
&
tsOpennedVnodes
,
1
);
taosHashPut
(
tsDnodeVnodesHash
,
(
const
char
*
)
&
pVnode
->
vgId
,
sizeof
(
int32_t
),
(
char
*
)(
&
pVnode
),
sizeof
(
SVnodeObj
*
));
return
TSDB_CODE_SUCCESS
;
return
TSDB_CODE_SUCCESS
;
}
}
...
@@ -286,13 +311,6 @@ void vnodeRelease(void *pVnodeRaw) {
...
@@ -286,13 +311,6 @@ void vnodeRelease(void *pVnodeRaw) {
}
}
tfree
(
pVnode
->
rootDir
);
tfree
(
pVnode
->
rootDir
);
// remove read queue
dnodeFreeRqueue
(
pVnode
->
rqueue
);
pVnode
->
rqueue
=
NULL
;
// remove write queue
dnodeFreeWqueue
(
pVnode
->
wqueue
);
pVnode
->
wqueue
=
NULL
;
if
(
pVnode
->
status
==
TAOS_VN_STATUS_DELETING
)
{
if
(
pVnode
->
status
==
TAOS_VN_STATUS_DELETING
)
{
char
rootDir
[
TSDB_FILENAME_LEN
]
=
{
0
};
char
rootDir
[
TSDB_FILENAME_LEN
]
=
{
0
};
...
@@ -387,15 +405,26 @@ static void vnodeCleanUp(SVnodeObj *pVnode) {
...
@@ -387,15 +405,26 @@ static void vnodeCleanUp(SVnodeObj *pVnode) {
pVnode
->
sync
=
NULL
;
pVnode
->
sync
=
NULL
;
}
}
cqClose
(
pVnode
->
cq
);
if
(
pVnode
->
wal
)
pVnode
->
cq
=
NULL
;
walClose
(
pVnode
->
wal
);
pVnode
->
wal
=
NULL
;
tsdbCloseRepo
(
pVnode
->
tsdb
,
1
);
if
(
pVnode
->
tsdb
)
tsdbCloseRepo
(
pVnode
->
tsdb
,
1
);
pVnode
->
tsdb
=
NULL
;
pVnode
->
tsdb
=
NULL
;
walClose
(
pVnode
->
wal
);
if
(
pVnode
->
cq
)
pVnode
->
wal
=
NULL
;
cqClose
(
pVnode
->
cq
);
pVnode
->
cq
=
NULL
;
if
(
pVnode
->
wqueue
)
dnodeFreeWqueue
(
pVnode
->
wqueue
);
pVnode
->
wqueue
=
NULL
;
if
(
pVnode
->
rqueue
)
dnodeFreeRqueue
(
pVnode
->
rqueue
);
pVnode
->
rqueue
=
NULL
;
vnodeRelease
(
pVnode
);
vnodeRelease
(
pVnode
);
}
}
...
@@ -512,27 +541,31 @@ static int32_t vnodeSaveCfg(SMDCreateVnodeMsg *pVnodeCfg) {
...
@@ -512,27 +541,31 @@ static int32_t vnodeSaveCfg(SMDCreateVnodeMsg *pVnodeCfg) {
}
}
static
int32_t
vnodeReadCfg
(
SVnodeObj
*
pVnode
)
{
static
int32_t
vnodeReadCfg
(
SVnodeObj
*
pVnode
)
{
char
cfgFile
[
TSDB_FILENAME_LEN
+
30
]
=
{
0
};
cJSON
*
root
=
NULL
;
char
*
content
=
NULL
;
char
cfgFile
[
TSDB_FILENAME_LEN
+
30
]
=
{
0
};
int
maxLen
=
1000
;
int32_t
code
=
TSDB_CODE_OTHERS
;
terrno
=
TSDB_CODE_OTHERS
;
sprintf
(
cfgFile
,
"%s/vnode%d/config.json"
,
tsVnodeDir
,
pVnode
->
vgId
);
sprintf
(
cfgFile
,
"%s/vnode%d/config.json"
,
tsVnodeDir
,
pVnode
->
vgId
);
FILE
*
fp
=
fopen
(
cfgFile
,
"r"
);
FILE
*
fp
=
fopen
(
cfgFile
,
"r"
);
if
(
!
fp
)
{
if
(
!
fp
)
{
vError
(
"vgId:%d, failed to open vnode cfg file
for read, file:%s
, error:%s"
,
pVnode
->
vgId
,
vError
(
"vgId:%d, failed to open vnode cfg file
:%s to read
, error:%s"
,
pVnode
->
vgId
,
cfgFile
,
strerror
(
errno
));
cfgFile
,
strerror
(
errno
));
return
errno
;
code
=
TAOS_SYSTEM_ERROR
(
errno
);
goto
PARSE_OVER
;
}
}
int
ret
=
TSDB_CODE_OTHERS
;
content
=
calloc
(
1
,
maxLen
+
1
);
int
maxLen
=
1000
;
if
(
content
==
NULL
)
goto
PARSE_OVER
;
char
*
content
=
calloc
(
1
,
maxLen
+
1
);
int
len
=
fread
(
content
,
1
,
maxLen
,
fp
);
int
len
=
fread
(
content
,
1
,
maxLen
,
fp
);
if
(
len
<=
0
)
{
if
(
len
<=
0
)
{
free
(
content
);
fclose
(
fp
);
vError
(
"vgId:%d, failed to read vnode cfg, content is null"
,
pVnode
->
vgId
);
vError
(
"vgId:%d, failed to read vnode cfg, content is null"
,
pVnode
->
vgId
);
return
false
;
return
errno
;
}
}
cJSON
*
root
=
cJSON_Parse
(
content
);
root
=
cJSON_Parse
(
content
);
if
(
root
==
NULL
)
{
if
(
root
==
NULL
)
{
vError
(
"vgId:%d, failed to read vnode cfg, invalid json format"
,
pVnode
->
vgId
);
vError
(
"vgId:%d, failed to read vnode cfg, invalid json format"
,
pVnode
->
vgId
);
goto
PARSE_OVER
;
goto
PARSE_OVER
;
...
@@ -691,19 +724,19 @@ static int32_t vnodeReadCfg(SVnodeObj *pVnode) {
...
@@ -691,19 +724,19 @@ static int32_t vnodeReadCfg(SVnodeObj *pVnode) {
pVnode
->
syncCfg
.
nodeInfo
[
i
].
nodePort
+=
TSDB_PORT_SYNC
;
pVnode
->
syncCfg
.
nodeInfo
[
i
].
nodePort
+=
TSDB_PORT_SYNC
;
}
}
ret
=
0
;
code
=
TSDB_CODE_SUCCESS
;
vPrint
(
"vgId:%d, read vnode cfg success
ed
, replcia:%d"
,
pVnode
->
vgId
,
pVnode
->
syncCfg
.
replica
);
vPrint
(
"vgId:%d, read vnode cfg success
fully
, replcia:%d"
,
pVnode
->
vgId
,
pVnode
->
syncCfg
.
replica
);
for
(
int32_t
i
=
0
;
i
<
pVnode
->
syncCfg
.
replica
;
i
++
)
{
for
(
int32_t
i
=
0
;
i
<
pVnode
->
syncCfg
.
replica
;
i
++
)
{
vPrint
(
"vgId:%d, dnode:%d, %s:%d"
,
pVnode
->
vgId
,
pVnode
->
syncCfg
.
nodeInfo
[
i
].
nodeId
,
vPrint
(
"vgId:%d, dnode:%d, %s:%d"
,
pVnode
->
vgId
,
pVnode
->
syncCfg
.
nodeInfo
[
i
].
nodeId
,
pVnode
->
syncCfg
.
nodeInfo
[
i
].
nodeFqdn
,
pVnode
->
syncCfg
.
nodeInfo
[
i
].
nodePort
);
pVnode
->
syncCfg
.
nodeInfo
[
i
].
nodeFqdn
,
pVnode
->
syncCfg
.
nodeInfo
[
i
].
nodePort
);
}
}
PARSE_OVER:
PARSE_OVER:
free
(
content
);
t
free
(
content
);
cJSON_Delete
(
root
);
cJSON_Delete
(
root
);
fclose
(
fp
);
if
(
fp
)
fclose
(
fp
);
return
ret
;
return
code
;
}
}
static
int32_t
vnodeSaveVersion
(
SVnodeObj
*
pVnode
)
{
static
int32_t
vnodeSaveVersion
(
SVnodeObj
*
pVnode
)
{
...
@@ -713,7 +746,7 @@ static int32_t vnodeSaveVersion(SVnodeObj *pVnode) {
...
@@ -713,7 +746,7 @@ static int32_t vnodeSaveVersion(SVnodeObj *pVnode) {
if
(
!
fp
)
{
if
(
!
fp
)
{
vError
(
"vgId:%d, failed to open vnode version file for write, file:%s error:%s"
,
pVnode
->
vgId
,
vError
(
"vgId:%d, failed to open vnode version file for write, file:%s error:%s"
,
pVnode
->
vgId
,
versionFile
,
strerror
(
errno
));
versionFile
,
strerror
(
errno
));
return
errno
;
return
TAOS_SYSTEM_ERROR
(
errno
);
}
}
int32_t
len
=
0
;
int32_t
len
=
0
;
...
@@ -733,29 +766,33 @@ static int32_t vnodeSaveVersion(SVnodeObj *pVnode) {
...
@@ -733,29 +766,33 @@ static int32_t vnodeSaveVersion(SVnodeObj *pVnode) {
return
0
;
return
0
;
}
}
static
bool
vnodeReadVersion
(
SVnodeObj
*
pVnode
)
{
static
int32_t
vnodeReadVersion
(
SVnodeObj
*
pVnode
)
{
char
versionFile
[
TSDB_FILENAME_LEN
+
30
]
=
{
0
};
char
versionFile
[
TSDB_FILENAME_LEN
+
30
]
=
{
0
};
char
*
content
=
NULL
;
cJSON
*
root
=
NULL
;
int
maxLen
=
100
;
int32_t
code
=
TSDB_CODE_OTHERS
;
sprintf
(
versionFile
,
"%s/vnode%d/version.json"
,
tsVnodeDir
,
pVnode
->
vgId
);
sprintf
(
versionFile
,
"%s/vnode%d/version.json"
,
tsVnodeDir
,
pVnode
->
vgId
);
FILE
*
fp
=
fopen
(
versionFile
,
"r"
);
FILE
*
fp
=
fopen
(
versionFile
,
"r"
);
if
(
!
fp
)
{
if
(
!
fp
)
{
if
(
errno
!=
ENOENT
)
{
if
(
errno
!=
ENOENT
)
{
vError
(
"vgId:%d, failed to open version file:%s error:%s"
,
pVnode
->
vgId
,
versionFile
,
strerror
(
errno
));
vError
(
"vgId:%d, failed to open version file:%s error:%s"
,
pVnode
->
vgId
,
versionFile
,
strerror
(
errno
));
code
=
TAOS_SYSTEM_ERROR
(
errno
);
}
else
{
code
=
TSDB_CODE_SUCCESS
;
}
}
return
false
;
goto
PARSE_OVER
;
}
}
bool
ret
=
false
;
content
=
calloc
(
1
,
maxLen
+
1
);
int
maxLen
=
100
;
char
*
content
=
calloc
(
1
,
maxLen
+
1
);
int
len
=
fread
(
content
,
1
,
maxLen
,
fp
);
int
len
=
fread
(
content
,
1
,
maxLen
,
fp
);
if
(
len
<=
0
)
{
if
(
len
<=
0
)
{
free
(
content
);
vError
(
"vgId:%d, failed to read vnode version, content is null"
,
pVnode
->
vgId
);
fclose
(
fp
);
goto
PARSE_OVER
;
vPrint
(
"vgId:%d, failed to read vnode version, content is null"
,
pVnode
->
vgId
);
return
false
;
}
}
cJSON
*
root
=
cJSON_Parse
(
content
);
root
=
cJSON_Parse
(
content
);
if
(
root
==
NULL
)
{
if
(
root
==
NULL
)
{
vError
(
"vgId:%d, failed to read vnode version, invalid json format"
,
pVnode
->
vgId
);
vError
(
"vgId:%d, failed to read vnode version, invalid json format"
,
pVnode
->
vgId
);
goto
PARSE_OVER
;
goto
PARSE_OVER
;
...
@@ -768,13 +805,13 @@ static bool vnodeReadVersion(SVnodeObj *pVnode) {
...
@@ -768,13 +805,13 @@ static bool vnodeReadVersion(SVnodeObj *pVnode) {
}
}
pVnode
->
version
=
version
->
valueint
;
pVnode
->
version
=
version
->
valueint
;
ret
=
true
;
code
=
TSDB_CODE_SUCCESS
;
vPrint
(
"vgId:%d, read vnode version succe
ed
, version:%"
PRId64
,
pVnode
->
vgId
,
pVnode
->
version
);
vPrint
(
"vgId:%d, read vnode version succe
ssfully
, version:%"
PRId64
,
pVnode
->
vgId
,
pVnode
->
version
);
PARSE_OVER:
PARSE_OVER:
free
(
content
);
t
free
(
content
);
cJSON_Delete
(
root
);
cJSON_Delete
(
root
);
fclose
(
fp
);
if
(
fp
)
fclose
(
fp
);
return
ret
;
return
code
;
}
}
src/wal/src/walMain.c
浏览文件 @
2a4feceb
...
@@ -25,6 +25,7 @@
...
@@ -25,6 +25,7 @@
#include "tlog.h"
#include "tlog.h"
#include "tchecksum.h"
#include "tchecksum.h"
#include "tutil.h"
#include "tutil.h"
#include "taoserror.h"
#include "twal.h"
#include "twal.h"
#include "tqueue.h"
#include "tqueue.h"
...
@@ -56,7 +57,10 @@ static int walRemoveWalFiles(const char *path);
...
@@ -56,7 +57,10 @@ static int walRemoveWalFiles(const char *path);
void
*
walOpen
(
const
char
*
path
,
const
SWalCfg
*
pCfg
)
{
void
*
walOpen
(
const
char
*
path
,
const
SWalCfg
*
pCfg
)
{
SWal
*
pWal
=
calloc
(
sizeof
(
SWal
),
1
);
SWal
*
pWal
=
calloc
(
sizeof
(
SWal
),
1
);
if
(
pWal
==
NULL
)
return
NULL
;
if
(
pWal
==
NULL
)
{
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
return
NULL
;
}
pWal
->
fd
=
-
1
;
pWal
->
fd
=
-
1
;
pWal
->
max
=
pCfg
->
wals
;
pWal
->
max
=
pCfg
->
wals
;
...
@@ -75,6 +79,7 @@ void *walOpen(const char *path, const SWalCfg *pCfg) {
...
@@ -75,6 +79,7 @@ void *walOpen(const char *path, const SWalCfg *pCfg) {
walRenew
(
pWal
);
walRenew
(
pWal
);
if
(
pWal
->
fd
<
0
)
{
if
(
pWal
->
fd
<
0
)
{
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
wError
(
"wal:%s, failed to open"
,
path
);
wError
(
"wal:%s, failed to open"
,
path
);
pthread_mutex_destroy
(
&
pWal
->
mutex
);
pthread_mutex_destroy
(
&
pWal
->
mutex
);
free
(
pWal
);
free
(
pWal
);
...
@@ -112,9 +117,10 @@ void walClose(void *handle) {
...
@@ -112,9 +117,10 @@ void walClose(void *handle) {
}
}
int
walRenew
(
void
*
handle
)
{
int
walRenew
(
void
*
handle
)
{
if
(
handle
==
NULL
)
return
0
;
SWal
*
pWal
=
handle
;
SWal
*
pWal
=
handle
;
int
code
=
0
;
int
code
=
0
;
pthread_mutex_lock
(
&
pWal
->
mutex
);
pthread_mutex_lock
(
&
pWal
->
mutex
);
if
(
pWal
->
fd
>=
0
)
{
if
(
pWal
->
fd
>=
0
)
{
...
@@ -156,6 +162,7 @@ int walRenew(void *handle) {
...
@@ -156,6 +162,7 @@ int walRenew(void *handle) {
int
walWrite
(
void
*
handle
,
SWalHead
*
pHead
)
{
int
walWrite
(
void
*
handle
,
SWalHead
*
pHead
)
{
SWal
*
pWal
=
handle
;
SWal
*
pWal
=
handle
;
int
code
=
0
;
int
code
=
0
;
if
(
pWal
==
NULL
)
return
-
1
;
// no wal
// no wal
if
(
pWal
->
level
==
TAOS_WAL_NOLOG
)
return
0
;
if
(
pWal
->
level
==
TAOS_WAL_NOLOG
)
return
0
;
...
@@ -178,6 +185,7 @@ int walWrite(void *handle, SWalHead *pHead) {
...
@@ -178,6 +185,7 @@ int walWrite(void *handle, SWalHead *pHead) {
void
walFsync
(
void
*
handle
)
{
void
walFsync
(
void
*
handle
)
{
SWal
*
pWal
=
handle
;
SWal
*
pWal
=
handle
;
if
(
pWal
==
NULL
)
return
;
if
(
pWal
->
level
==
TAOS_WAL_FSYNC
&&
pWal
->
fd
>=
0
)
{
if
(
pWal
->
level
==
TAOS_WAL_FSYNC
&&
pWal
->
fd
>=
0
)
{
if
(
fsync
(
pWal
->
fd
)
<
0
)
{
if
(
fsync
(
pWal
->
fd
)
<
0
)
{
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录