Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
Cherenkov effect
TDengine
提交
dc3c33a8
T
TDengine
项目概览
Cherenkov effect
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
2
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,发现更多精彩内容 >>
提交
dc3c33a8
编写于
4月 08, 2020
作者:
J
Jeff Tao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
integrate WAL module
上级
4f6cb328
变更
8
隐藏空白更改
内联
并排
Showing
8 changed file
with
70 addition
and
27 deletion
+70
-27
src/inc/dnode.h
src/inc/dnode.h
+1
-1
src/inc/tsync.h
src/inc/tsync.h
+4
-2
src/inc/twal.h
src/inc/twal.h
+2
-2
src/vnode/main/inc/vnodeInt.h
src/vnode/main/inc/vnodeInt.h
+2
-2
src/vnode/main/src/vnodeMain.c
src/vnode/main/src/vnodeMain.c
+45
-11
src/vnode/main/src/vnodeWrite.c
src/vnode/main/src/vnodeWrite.c
+2
-1
src/vnode/wal/src/walMain.c
src/vnode/wal/src/walMain.c
+6
-6
src/vnode/wal/test/waltest.c
src/vnode/wal/test/waltest.c
+8
-2
未找到文件。
src/inc/dnode.h
浏览文件 @
dc3c33a8
...
...
@@ -42,7 +42,7 @@ void *dnodeAllocateWqueue(void *pVnode);
void
dnodeFreeWqueue
(
void
*
queue
);
void
*
dnodeAllocateRqueue
(
void
*
pVnode
);
void
dnodeFreeRqueue
(
void
*
rqueue
);
void
dnodeSend
WriteResponse
(
void
*
pVnode
,
void
*
param
,
int32_t
code
);
void
dnodeSend
RpcWriteRsp
(
void
*
pVnode
,
void
*
param
,
int32_t
code
);
#ifdef __cplusplus
}
...
...
src/inc/tsync.h
浏览文件 @
dc3c33a8
...
...
@@ -66,11 +66,11 @@ typedef struct {
// if name is null, get the file from index or after, used by master
// if name is provided, get the named file at the specified index, used by unsynced node
// it returns the file magic number and size, if file not there, magic shall be 0.
uint32_t
(
*
getFileInfo
)(
char
*
name
,
int
*
index
,
in
t
*
size
);
uint32_t
(
*
getFileInfo
)(
void
*
ahandle
,
char
*
name
,
uint32_t
*
index
,
int32_
t
*
size
);
// get the wal file from index or after
// return value, -1: error, 1:more wal files, 0:last WAL. if name[0]==0, no WAL file
int
(
*
getWalInfo
)(
char
*
name
,
in
t
*
index
);
int
(
*
getWalInfo
)(
void
*
ahandle
,
char
*
name
,
uint32_
t
*
index
);
// when a forward pkt is received, call this to handle data
int
(
*
writeToCache
)(
void
*
ahandle
,
void
*
pHead
,
int
type
);
...
...
@@ -94,11 +94,13 @@ int syncGetNodesRole(tsync_h shandle, SNodesRole *);
extern
char
*
syncRole
[];
//global configurable parameters
extern
int
tsMaxSyncNum
;
extern
int
tsSyncTcpThreads
;
extern
int
tsMaxWatchFiles
;
extern
short
tsSyncPort
;
extern
int
tsMaxFwdInfo
;
extern
int
sDebugFlag
;
#ifdef __cplusplus
}
...
...
src/inc/twal.h
浏览文件 @
dc3c33a8
...
...
@@ -40,12 +40,12 @@ typedef struct {
typedef
void
*
twal_h
;
// WAL HANDLE
twal_h
walOpen
(
char
*
path
,
int
max
,
int
level
);
twal_h
walOpen
(
char
*
path
,
SWalCfg
*
pCfg
);
void
walClose
(
twal_h
);
int
walRenew
(
twal_h
);
int
walWrite
(
twal_h
,
SWalHead
*
);
void
walFsync
(
twal_h
);
int
walRestore
(
twal_h
,
void
*
pVnode
,
int
(
*
writeFp
)(
void
*
ahandle
,
SWalHea
d
*
pHead
,
int
type
));
int
walRestore
(
twal_h
,
void
*
pVnode
,
int
(
*
writeFp
)(
void
*
ahandle
,
voi
d
*
pHead
,
int
type
));
int
walGetWalFile
(
twal_h
,
char
*
name
,
uint32_t
*
index
);
extern
int
wDebugFlag
;
...
...
src/vnode/main/inc/vnodeInt.h
浏览文件 @
dc3c33a8
...
...
@@ -35,7 +35,7 @@ typedef struct {
int32_t
vgId
;
// global vnode group ID
int32_t
refCount
;
// reference count
EVnStatus
status
;
int
role
;
int
8_t
role
;
int64_t
version
;
void
*
wqueue
;
void
*
rqueue
;
...
...
@@ -49,7 +49,7 @@ typedef struct {
SWalCfg
walCfg
;
}
SVnodeObj
;
int
vnodeWriteToQueue
(
void
*
param
,
SWalHea
d
*
pHead
,
int
type
);
int
vnodeWriteToQueue
(
void
*
param
,
voi
d
*
pHead
,
int
type
);
void
vnodeInitWriteFp
(
void
);
void
vnodeInitReadFp
(
void
);
...
...
src/vnode/main/src/vnodeMain.c
浏览文件 @
dc3c33a8
...
...
@@ -29,13 +29,18 @@
#include "vnode.h"
#include "vnodeInt.h"
static
void
*
tsDnodeVnodesHash
;
static
void
vnodeCleanUp
(
SVnodeObj
*
pVnode
);
static
void
vnodeBuildVloadMsg
(
char
*
pNode
,
void
*
param
);
static
int
vnodeWALCallback
(
void
*
arg
);
static
int32_t
vnodeSaveCfg
(
SMDCreateVnodeMsg
*
pVnodeCfg
);
static
int32_t
vnodeReadCfg
(
SVnodeObj
*
pVnode
);
static
void
*
tsDnodeVnodesHash
;
static
void
vnodeCleanUp
(
SVnodeObj
*
pVnode
);
static
void
vnodeBuildVloadMsg
(
char
*
pNode
,
void
*
param
);
static
int
vnodeWalCallback
(
void
*
arg
);
static
int32_t
vnodeSaveCfg
(
SMDCreateVnodeMsg
*
pVnodeCfg
);
static
int32_t
vnodeReadCfg
(
SVnodeObj
*
pVnode
);
static
int
vnodeWalCallback
(
void
*
arg
);
static
uint32_t
vnodeGetFileInfo
(
void
*
ahandle
,
char
*
name
,
uint32_t
*
index
,
int32_t
*
size
);
static
int
vnodeGetWalInfo
(
void
*
ahandle
,
char
*
name
,
uint32_t
*
index
);
static
void
vnodeNotifyRole
(
void
*
ahandle
,
int8_t
role
);
// module global
static
int
tsOpennedVnodes
;
static
pthread_once_t
vnodeModuleInit
=
PTHREAD_ONCE_INIT
;
...
...
@@ -140,14 +145,27 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
pVnode
->
rqueue
=
dnodeAllocateRqueue
(
pVnode
);
sprintf
(
temp
,
"%s/wal"
,
rootDir
);
pVnode
->
wal
=
walOpen
(
temp
,
pVnode
->
walCfg
.
wals
,
pVnode
->
walCfg
.
commitLog
);
pVnode
->
sync
=
NULL
;
pVnode
->
wal
=
walOpen
(
temp
,
&
pVnode
->
walCfg
);
SSyncInfo
syncInfo
;
syncInfo
.
vgId
=
pVnode
->
vgId
;
syncInfo
.
vgId
=
pVnode
->
version
;
syncInfo
.
syncCfg
=
pVnode
->
syncCfg
;
sprintf
(
syncInfo
.
path
,
"%s/tsdb/"
,
rootDir
);
syncInfo
.
ahandle
=
pVnode
;
syncInfo
.
getWalInfo
=
vnodeGetWalInfo
;
syncInfo
.
getFileInfo
=
vnodeGetFileInfo
;
syncInfo
.
writeToCache
=
vnodeWriteToQueue
;
syncInfo
.
confirmForward
=
dnodeSendRpcWriteRsp
;
syncInfo
.
notifyRole
=
vnodeNotifyRole
;
// pVnode->sync = syncStart(&syncInfo);;
pVnode
->
events
=
NULL
;
pVnode
->
cq
=
NULL
;
STsdbAppH
appH
=
{
0
};
appH
.
appH
=
(
void
*
)
pVnode
;
appH
.
walCallBack
=
vnodeW
AL
Callback
;
appH
.
walCallBack
=
vnodeW
al
Callback
;
sprintf
(
temp
,
"%s/tsdb"
,
rootDir
);
void
*
pTsdb
=
tsdbOpenRepo
(
temp
,
&
appH
);
...
...
@@ -281,11 +299,27 @@ static void vnodeCleanUp(SVnodeObj *pVnode) {
}
// TODO: this is a simple implement
static
int
vnodeW
AL
Callback
(
void
*
arg
)
{
static
int
vnodeW
al
Callback
(
void
*
arg
)
{
SVnodeObj
*
pVnode
=
arg
;
return
walRenew
(
pVnode
->
wal
);
}
static
uint32_t
vnodeGetFileInfo
(
void
*
ahandle
,
char
*
name
,
uint32_t
*
index
,
int32_t
*
size
)
{
// SVnodeObj *pVnode = ahandle;
//tsdbGetFileInfo(pVnode->tsdb, name, index, size);
return
0
;
}
static
int
vnodeGetWalInfo
(
void
*
ahandle
,
char
*
name
,
uint32_t
*
index
)
{
SVnodeObj
*
pVnode
=
ahandle
;
return
walGetWalFile
(
pVnode
->
wal
,
name
,
index
);
}
static
void
vnodeNotifyRole
(
void
*
ahandle
,
int8_t
role
)
{
SVnodeObj
*
pVnode
=
ahandle
;
pVnode
->
role
=
role
;
}
static
int32_t
vnodeSaveCfg
(
SMDCreateVnodeMsg
*
pVnodeCfg
)
{
char
cfgFile
[
TSDB_FILENAME_LEN
*
2
]
=
{
0
};
sprintf
(
cfgFile
,
"%s/vnode%d/config"
,
tsVnodeDir
,
pVnodeCfg
->
cfg
.
vgId
);
...
...
src/vnode/main/src/vnodeWrite.c
浏览文件 @
dc3c33a8
...
...
@@ -248,8 +248,9 @@ static int32_t vnodeProcessDropStableMsg(SVnodeObj *pVnode, void *pCont, SRspRet
return
code
;
}
int
vnodeWriteToQueue
(
void
*
param
,
SWalHead
*
pHead
,
int
type
)
{
int
vnodeWriteToQueue
(
void
*
param
,
void
*
data
,
int
type
)
{
SVnodeObj
*
pVnode
=
param
;
SWalHead
*
pHead
=
data
;
int
size
=
sizeof
(
SWalHead
)
+
pHead
->
len
;
SWalHead
*
pWal
=
(
SWalHead
*
)
taosAllocateQitem
(
size
);
...
...
src/vnode/wal/src/walMain.c
浏览文件 @
dc3c33a8
...
...
@@ -49,18 +49,18 @@ int wDebugFlag = 135;
static
uint32_t
walSignature
=
0xFAFBFDFE
;
static
int
walHandleExistingFiles
(
char
*
path
);
static
int
walRestoreWalFile
(
char
*
name
,
void
*
pVnode
,
int
(
*
writeFp
)(
void
*
,
SWalHea
d
*
,
int
));
static
int
walRestoreWalFile
(
char
*
name
,
void
*
pVnode
,
int
(
*
writeFp
)(
void
*
,
voi
d
*
,
int
));
static
int
walRemoveWalFiles
(
char
*
path
);
void
*
walOpen
(
char
*
path
,
int
max
,
int
level
)
{
void
*
walOpen
(
char
*
path
,
SWalCfg
*
pCfg
)
{
SWal
*
pWal
=
calloc
(
sizeof
(
SWal
),
1
);
if
(
pWal
==
NULL
)
return
NULL
;
pWal
->
fd
=
-
1
;
pWal
->
max
=
max
;
pWal
->
max
=
pCfg
->
wals
;
pWal
->
id
=
0
;
pWal
->
num
=
0
;
pWal
->
level
=
level
;
pWal
->
level
=
pCfg
->
commitLog
;
strcpy
(
pWal
->
path
,
path
);
pthread_mutex_init
(
&
pWal
->
mutex
,
NULL
);
...
...
@@ -169,7 +169,7 @@ void walFsync(void *handle) {
fsync
(
pWal
->
fd
);
}
int
walRestore
(
void
*
handle
,
void
*
pVnode
,
int
(
*
writeFp
)(
void
*
,
SWalHea
d
*
,
int
))
{
int
walRestore
(
void
*
handle
,
void
*
pVnode
,
int
(
*
writeFp
)(
void
*
,
voi
d
*
,
int
))
{
SWal
*
pWal
=
(
SWal
*
)
handle
;
int
code
=
0
;
struct
dirent
*
ent
;
...
...
@@ -246,7 +246,7 @@ int walGetWalFile(void *handle, char *name, uint32_t *index) {
return
code
;
}
static
int
walRestoreWalFile
(
char
*
name
,
void
*
pVnode
,
int
(
*
writeFp
)(
void
*
,
SWalHea
d
*
,
int
))
{
static
int
walRestoreWalFile
(
char
*
name
,
void
*
pVnode
,
int
(
*
writeFp
)(
void
*
,
voi
d
*
,
int
))
{
int
code
=
0
;
char
*
buffer
=
malloc
(
1024000
);
// size for one record
...
...
src/vnode/wal/test/waltest.c
浏览文件 @
dc3c33a8
...
...
@@ -21,8 +21,10 @@
int64_t
ver
=
0
;
void
*
pWal
=
NULL
;
int
writeToQueue
(
void
*
pVnode
,
SWalHead
*
pHead
,
int
type
)
{
int
writeToQueue
(
void
*
pVnode
,
void
*
data
,
int
type
)
{
// do nothing
SWalHead
*
pHead
=
data
;
if
(
pHead
->
version
>
ver
)
ver
=
pHead
->
version
;
...
...
@@ -74,7 +76,11 @@ int main(int argc, char *argv[]) {
taosInitLog
(
"wal.log"
,
100000
,
10
);
pWal
=
walOpen
(
path
,
max
,
level
);
SWalCfg
walCfg
;
walCfg
.
commitLog
=
level
;
walCfg
.
wals
=
max
;
pWal
=
walOpen
(
path
,
&
walCfg
);
if
(
pWal
==
NULL
)
{
printf
(
"failed to open wal
\n
"
);
exit
(
-
1
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录