Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
793eb114
T
TDengine
项目概览
taosdata
/
TDengine
12 个月 前同步成功
通知
1180
Star
22014
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,发现更多精彩内容 >>
提交
793eb114
编写于
12月 13, 2021
作者:
L
Liu Jicong
浏览文件
操作
浏览文件
下载
差异文件
Merge branch 'feature/tq' into 3.0
上级
6031721f
dc245929
变更
11
隐藏空白更改
内联
并排
Showing
11 changed file
with
292 addition
and
146 deletion
+292
-146
include/dnode/vnode/tq/tq.h
include/dnode/vnode/tq/tq.h
+1
-1
include/libs/wal/wal.h
include/libs/wal/wal.h
+29
-17
source/libs/wal/inc/walInt.h
source/libs/wal/inc/walInt.h
+1
-1
source/libs/wal/src/walMeta.c
source/libs/wal/src/walMeta.c
+14
-14
source/libs/wal/src/walMgmt.c
source/libs/wal/src/walMgmt.c
+23
-29
source/libs/wal/src/walRead.c
source/libs/wal/src/walRead.c
+142
-1
source/libs/wal/src/walSeek.c
source/libs/wal/src/walSeek.c
+7
-10
source/libs/wal/src/walUtil.c
source/libs/wal/src/walUtil.c
+2
-0
source/libs/wal/src/walWrite.c
source/libs/wal/src/walWrite.c
+54
-54
source/libs/wal/test/walMetaTest.cpp
source/libs/wal/test/walMetaTest.cpp
+18
-18
source/util/src/tarray.c
source/util/src/tarray.c
+1
-1
未找到文件。
include/dnode/vnode/tq/tq.h
浏览文件 @
793eb114
...
...
@@ -256,7 +256,7 @@ typedef struct STQ {
// the collection of group handle
// the handle of kvstore
char
*
path
;
STqCfg
*
tqConfig
;
STqCfg
*
tqConfig
;
TqLogReader
*
tqLogReader
;
TqMemRef
tqMemRef
;
TqMetaStore
*
tqMeta
;
...
...
include/libs/wal/wal.h
浏览文件 @
793eb114
...
...
@@ -44,7 +44,7 @@ typedef struct SWalReadHead {
int8_t
reserved
[
2
];
int32_t
len
;
int64_t
version
;
char
cont
[];
char
body
[];
}
SWalReadHead
;
typedef
struct
{
...
...
@@ -52,9 +52,9 @@ typedef struct {
int32_t
fsyncPeriod
;
// millisecond
int32_t
retentionPeriod
;
// secs
int32_t
rollPeriod
;
// secs
int
32_t
retentionSize
;
// secs
int
64_t
retentionSize
;
int64_t
segSize
;
EWalType
walL
evel
;
// wal level
EWalType
l
evel
;
// wal level
}
SWalCfg
;
typedef
struct
{
...
...
@@ -90,15 +90,17 @@ typedef struct {
#define WAL_CUR_FILE_WRITABLE 2
#define WAL_CUR_FAILED 4
typedef
struct
SWalVer
{
int64_t
firstVer
;
int64_t
verInSnapshotting
;
int64_t
snapshotVer
;
int64_t
commitVer
;
int64_t
lastVer
;
}
SWalVer
;
typedef
struct
SWal
{
// cfg
int32_t
vgId
;
int32_t
fsyncPeriod
;
// millisecond
int32_t
rollPeriod
;
// second
int64_t
segSize
;
int64_t
retentionSize
;
int32_t
retentionPeriod
;
EWalType
level
;
SWalCfg
cfg
;
//total size
int64_t
totSize
;
//fsync seq
...
...
@@ -109,12 +111,7 @@ typedef struct SWal {
int64_t
writeLogTfd
;
int64_t
writeIdxTfd
;
//wal lifecycle
int64_t
firstVersion
;
int64_t
snapshotVersion
;
int64_t
commitVersion
;
int64_t
lastVersion
;
//snapshotting version
int64_t
snapshottingVer
;
SWalVer
vers
;
//roll status
int64_t
lastRollSeq
;
//file set
...
...
@@ -126,9 +123,20 @@ typedef struct SWal {
//path
char
path
[
WAL_PATH_LEN
];
//reusable write head
SWalHead
h
ead
;
SWalHead
writeH
ead
;
}
SWal
;
// WAL HANDLE
typedef
struct
SWalReadHandle
{
SWal
*
pWal
;
int64_t
readLogTfd
;
int64_t
readIdxTfd
;
int64_t
curFileFirstVer
;
int64_t
curVersion
;
int64_t
capacity
;
int64_t
status
;
//if cursor valid
SWalHead
head
;
}
SWalReadHandle
;
typedef
int32_t
(
*
FWalWrite
)(
void
*
ahandle
,
void
*
pHead
);
// module initialization
...
...
@@ -154,6 +162,10 @@ int32_t walEndTakeSnapshot(SWal *);
//int32_t walDataCorrupted(SWal*);
// read
SWalReadHandle
*
walOpenReadHandle
(
SWal
*
);
void
walCloseReadHandle
(
SWalReadHandle
*
);
int32_t
walReadWithHandle
(
SWalReadHandle
*
pRead
,
int64_t
ver
);
int32_t
walRead
(
SWal
*
,
SWalHead
**
,
int64_t
ver
);
int32_t
walReadWithFp
(
SWal
*
,
FWalWrite
writeFp
,
int64_t
verStart
,
int32_t
readNum
);
...
...
source/libs/wal/inc/walInt.h
浏览文件 @
793eb114
...
...
@@ -90,7 +90,7 @@ static inline int walValidHeadCksum(SWalHead* pHead) {
}
static
inline
int
walValidBodyCksum
(
SWalHead
*
pHead
)
{
return
taosCheckChecksum
((
uint8_t
*
)
pHead
->
head
.
cont
,
pHead
->
head
.
len
,
pHead
->
cksumBody
);
return
taosCheckChecksum
((
uint8_t
*
)
pHead
->
head
.
body
,
pHead
->
head
.
len
,
pHead
->
cksumBody
);
}
static
inline
int
walValidCksum
(
SWalHead
*
pHead
,
void
*
body
,
int64_t
bodyLen
)
{
...
...
source/libs/wal/src/walMeta.c
浏览文件 @
793eb114
...
...
@@ -25,15 +25,15 @@
#include <regex.h>
int64_t
walGetFirstVer
(
SWal
*
pWal
)
{
return
pWal
->
firstVersion
;
return
pWal
->
vers
.
firstVer
;
}
int64_t
walGetSnaphostVer
(
SWal
*
pWal
)
{
return
pWal
->
snapshotVersion
;
return
pWal
->
vers
.
snapshotVer
;
}
int64_t
walGetLastVer
(
SWal
*
pWal
)
{
return
pWal
->
lastVersion
;
return
pWal
->
vers
.
lastVer
;
}
int
walRollFileInfo
(
SWal
*
pWal
)
{
...
...
@@ -42,7 +42,7 @@ int walRollFileInfo(SWal* pWal) {
SArray
*
pArray
=
pWal
->
fileInfoSet
;
if
(
taosArrayGetSize
(
pArray
)
!=
0
)
{
WalFileInfo
*
pInfo
=
taosArrayGetLast
(
pArray
);
pInfo
->
lastVer
=
pWal
->
lastVersion
;
pInfo
->
lastVer
=
pWal
->
vers
.
lastVer
;
pInfo
->
closeTs
=
ts
;
}
...
...
@@ -51,7 +51,7 @@ int walRollFileInfo(SWal* pWal) {
if
(
pNewInfo
==
NULL
)
{
return
-
1
;
}
pNewInfo
->
firstVer
=
pWal
->
lastVersion
+
1
;
pNewInfo
->
firstVer
=
pWal
->
vers
.
lastVer
+
1
;
pNewInfo
->
lastVer
=
-
1
;
pNewInfo
->
createTs
=
ts
;
pNewInfo
->
closeTs
=
-
1
;
...
...
@@ -74,13 +74,13 @@ char* walMetaSerialize(SWal* pWal) {
return
NULL
;
}
cJSON_AddItemToObject
(
pRoot
,
"meta"
,
pMeta
);
sprintf
(
buf
,
"%"
PRId64
,
pWal
->
firstVersion
);
sprintf
(
buf
,
"%"
PRId64
,
pWal
->
vers
.
firstVer
);
cJSON_AddStringToObject
(
pMeta
,
"firstVer"
,
buf
);
sprintf
(
buf
,
"%"
PRId64
,
pWal
->
snapshotVersion
);
sprintf
(
buf
,
"%"
PRId64
,
pWal
->
vers
.
snapshotVer
);
cJSON_AddStringToObject
(
pMeta
,
"snapshotVer"
,
buf
);
sprintf
(
buf
,
"%"
PRId64
,
pWal
->
commitVersion
);
sprintf
(
buf
,
"%"
PRId64
,
pWal
->
vers
.
commitVer
);
cJSON_AddStringToObject
(
pMeta
,
"commitVer"
,
buf
);
sprintf
(
buf
,
"%"
PRId64
,
pWal
->
lastVersion
);
sprintf
(
buf
,
"%"
PRId64
,
pWal
->
vers
.
lastVer
);
cJSON_AddStringToObject
(
pMeta
,
"lastVer"
,
buf
);
cJSON_AddItemToObject
(
pRoot
,
"files"
,
pFiles
);
...
...
@@ -116,13 +116,13 @@ int walMetaDeserialize(SWal* pWal, const char* bytes) {
pRoot
=
cJSON_Parse
(
bytes
);
pMeta
=
cJSON_GetObjectItem
(
pRoot
,
"meta"
);
pField
=
cJSON_GetObjectItem
(
pMeta
,
"firstVer"
);
pWal
->
firstVersion
=
atoll
(
cJSON_GetStringValue
(
pField
));
pWal
->
vers
.
firstVer
=
atoll
(
cJSON_GetStringValue
(
pField
));
pField
=
cJSON_GetObjectItem
(
pMeta
,
"snapshotVer"
);
pWal
->
snapshotVersion
=
atoll
(
cJSON_GetStringValue
(
pField
));
pWal
->
vers
.
snapshotVer
=
atoll
(
cJSON_GetStringValue
(
pField
));
pField
=
cJSON_GetObjectItem
(
pMeta
,
"commitVer"
);
pWal
->
commitVersion
=
atoll
(
cJSON_GetStringValue
(
pField
));
pWal
->
vers
.
commitVer
=
atoll
(
cJSON_GetStringValue
(
pField
));
pField
=
cJSON_GetObjectItem
(
pMeta
,
"lastVer"
);
pWal
->
lastVersion
=
atoll
(
cJSON_GetStringValue
(
pField
));
pWal
->
vers
.
lastVer
=
atoll
(
cJSON_GetStringValue
(
pField
));
pFiles
=
cJSON_GetObjectItem
(
pRoot
,
"files"
);
int
sz
=
cJSON_GetArraySize
(
pFiles
);
...
...
@@ -161,7 +161,7 @@ static int walFindCurMetaVer(SWal* pWal) {
DIR
*
dir
=
opendir
(
pWal
->
path
);
if
(
dir
==
NULL
)
{
wError
(
"vgId:%d, path:%s, failed to open since %s"
,
pWal
->
vgId
,
pWal
->
path
,
strerror
(
errno
));
wError
(
"vgId:%d, path:%s, failed to open since %s"
,
pWal
->
cfg
.
vgId
,
pWal
->
path
,
strerror
(
errno
));
return
-
1
;
}
...
...
source/libs/wal/src/walMgmt.c
浏览文件 @
793eb114
...
...
@@ -86,21 +86,15 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
pWal
->
writeCur
=
-
1
;
//set config
pWal
->
vgId
=
pCfg
->
vgId
;
pWal
->
fsyncPeriod
=
pCfg
->
fsyncPeriod
;
pWal
->
rollPeriod
=
pCfg
->
rollPeriod
;
pWal
->
segSize
=
pCfg
->
segSize
;
pWal
->
retentionSize
=
pCfg
->
retentionSize
;
pWal
->
retentionPeriod
=
pCfg
->
retentionPeriod
;
pWal
->
level
=
pCfg
->
walLevel
;
memcpy
(
&
pWal
->
cfg
,
pCfg
,
sizeof
(
SWalCfg
));
//init version info
pWal
->
firstVersion
=
-
1
;
pWal
->
commitVersion
=
-
1
;
pWal
->
snapshotVersion
=
-
1
;
pWal
->
lastVersion
=
-
1
;
pWal
->
vers
.
firstVer
=
-
1
;
pWal
->
vers
.
commitVer
=
-
1
;
pWal
->
vers
.
snapshotVer
=
-
1
;
pWal
->
vers
.
lastVer
=
-
1
;
pWal
->
snapshottingVer
=
-
1
;
pWal
->
vers
.
verInSnapshotting
=
-
1
;
pWal
->
totSize
=
0
;
...
...
@@ -108,8 +102,8 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
pWal
->
lastRollSeq
=
-
1
;
//init write buffer
memset
(
&
pWal
->
h
ead
,
0
,
sizeof
(
SWalHead
));
pWal
->
h
ead
.
head
.
sver
=
0
;
memset
(
&
pWal
->
writeH
ead
,
0
,
sizeof
(
SWalHead
));
pWal
->
writeH
ead
.
head
.
sver
=
0
;
tstrncpy
(
pWal
->
path
,
path
,
sizeof
(
pWal
->
path
));
pthread_mutex_init
(
&
pWal
->
mutex
,
NULL
);
...
...
@@ -129,7 +123,7 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
}
walReadMeta
(
pWal
);
wDebug
(
"vgId:%d, wal:%p is opened, level:%d fsyncPeriod:%d"
,
pWal
->
vgId
,
pWal
,
pWal
->
level
,
pWal
->
fsyncPeriod
);
wDebug
(
"vgId:%d, wal:%p is opened, level:%d fsyncPeriod:%d"
,
pWal
->
cfg
.
vgId
,
pWal
,
pWal
->
cfg
.
level
,
pWal
->
cfg
.
fsyncPeriod
);
return
pWal
;
}
...
...
@@ -137,17 +131,17 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
int32_t
walAlter
(
SWal
*
pWal
,
SWalCfg
*
pCfg
)
{
if
(
pWal
==
NULL
)
return
TSDB_CODE_WAL_APP_ERROR
;
if
(
pWal
->
level
==
pCfg
->
walLevel
&&
pWal
->
fsyncPeriod
==
pCfg
->
fsyncPeriod
)
{
wDebug
(
"vgId:%d, old walLevel:%d fsync:%d, new walLevel:%d fsync:%d not change"
,
pWal
->
vgId
,
pWal
->
level
,
pWal
->
fsyncPeriod
,
pCfg
->
walL
evel
,
pCfg
->
fsyncPeriod
);
if
(
pWal
->
cfg
.
level
==
pCfg
->
level
&&
pWal
->
cfg
.
fsyncPeriod
==
pCfg
->
fsyncPeriod
)
{
wDebug
(
"vgId:%d, old walLevel:%d fsync:%d, new walLevel:%d fsync:%d not change"
,
pWal
->
cfg
.
vgId
,
pWal
->
cfg
.
level
,
pWal
->
cfg
.
fsyncPeriod
,
pCfg
->
l
evel
,
pCfg
->
fsyncPeriod
);
return
0
;
}
wInfo
(
"vgId:%d, change old walLevel:%d fsync:%d, new walLevel:%d fsync:%d"
,
pWal
->
vgId
,
pWal
->
level
,
pWal
->
fsyncPeriod
,
pCfg
->
walL
evel
,
pCfg
->
fsyncPeriod
);
wInfo
(
"vgId:%d, change old walLevel:%d fsync:%d, new walLevel:%d fsync:%d"
,
pWal
->
cfg
.
vgId
,
pWal
->
cfg
.
level
,
pWal
->
cfg
.
fsyncPeriod
,
pCfg
->
l
evel
,
pCfg
->
fsyncPeriod
);
pWal
->
level
=
pCfg
->
walL
evel
;
pWal
->
fsyncPeriod
=
pCfg
->
fsyncPeriod
;
pWal
->
cfg
.
level
=
pCfg
->
l
evel
;
pWal
->
cfg
.
fsyncPeriod
=
pCfg
->
fsyncPeriod
;
pWal
->
fsyncSeq
=
pCfg
->
fsyncPeriod
/
1000
;
if
(
pWal
->
fsyncSeq
<=
0
)
pWal
->
fsyncSeq
=
1
;
...
...
@@ -171,22 +165,22 @@ void walClose(SWal *pWal) {
static
int32_t
walInitObj
(
SWal
*
pWal
)
{
if
(
taosMkDir
(
pWal
->
path
)
!=
0
)
{
wError
(
"vgId:%d, path:%s, failed to create directory since %s"
,
pWal
->
vgId
,
pWal
->
path
,
strerror
(
errno
));
wError
(
"vgId:%d, path:%s, failed to create directory since %s"
,
pWal
->
cfg
.
vgId
,
pWal
->
path
,
strerror
(
errno
));
return
TAOS_SYSTEM_ERROR
(
errno
);
}
pWal
->
fileInfoSet
=
taosArrayInit
(
8
,
sizeof
(
WalFileInfo
));
if
(
pWal
->
fileInfoSet
==
NULL
)
{
wError
(
"vgId:%d, path:%s, failed to init taosArray %s"
,
pWal
->
vgId
,
pWal
->
path
,
strerror
(
errno
));
wError
(
"vgId:%d, path:%s, failed to init taosArray %s"
,
pWal
->
cfg
.
vgId
,
pWal
->
path
,
strerror
(
errno
));
return
TAOS_SYSTEM_ERROR
(
errno
);
}
wDebug
(
"vgId:%d, object is initialized"
,
pWal
->
vgId
);
wDebug
(
"vgId:%d, object is initialized"
,
pWal
->
cfg
.
vgId
);
return
0
;
}
static
void
walFreeObj
(
void
*
wal
)
{
SWal
*
pWal
=
wal
;
wDebug
(
"vgId:%d, wal:%p is freed"
,
pWal
->
vgId
,
pWal
);
wDebug
(
"vgId:%d, wal:%p is freed"
,
pWal
->
cfg
.
vgId
,
pWal
);
tfClose
(
pWal
->
writeLogTfd
);
tfClose
(
pWal
->
writeIdxTfd
);
...
...
@@ -197,7 +191,7 @@ static void walFreeObj(void *wal) {
}
static
bool
walNeedFsync
(
SWal
*
pWal
)
{
if
(
pWal
->
fsyncPeriod
<=
0
||
pWal
->
level
!=
TAOS_WAL_FSYNC
)
{
if
(
pWal
->
cfg
.
fsyncPeriod
<=
0
||
pWal
->
cfg
.
level
!=
TAOS_WAL_FSYNC
)
{
return
false
;
}
...
...
@@ -217,10 +211,10 @@ static void walFsyncAll() {
SWal
*
pWal
=
taosIterateRef
(
tsWal
.
refSetId
,
0
);
while
(
pWal
)
{
if
(
walNeedFsync
(
pWal
))
{
wTrace
(
"vgId:%d, do fsync, level:%d seq:%d rseq:%d"
,
pWal
->
vgId
,
pWal
->
level
,
pWal
->
fsyncSeq
,
atomic_load_32
(
&
tsWal
.
seq
));
wTrace
(
"vgId:%d, do fsync, level:%d seq:%d rseq:%d"
,
pWal
->
cfg
.
vgId
,
pWal
->
cfg
.
level
,
pWal
->
fsyncSeq
,
atomic_load_32
(
&
tsWal
.
seq
));
int32_t
code
=
tfFsync
(
pWal
->
writeLogTfd
);
if
(
code
!=
0
)
{
wError
(
"vgId:%d, file:%"
PRId64
".log, failed to fsync since %s"
,
pWal
->
vgId
,
walGetLastFileFirstVer
(
pWal
),
strerror
(
code
));
wError
(
"vgId:%d, file:%"
PRId64
".log, failed to fsync since %s"
,
pWal
->
cfg
.
vgId
,
walGetLastFileFirstVer
(
pWal
),
strerror
(
code
));
}
}
pWal
=
taosIterateRef
(
tsWal
.
refSetId
,
pWal
->
refId
);
...
...
source/libs/wal/src/walRead.c
浏览文件 @
793eb114
...
...
@@ -16,6 +16,147 @@
#include "walInt.h"
#include "tfile.h"
SWalReadHandle
*
walOpenReadHandle
(
SWal
*
pWal
)
{
SWalReadHandle
*
pRead
=
malloc
(
sizeof
(
SWalReadHandle
));
if
(
pRead
==
NULL
)
{
return
NULL
;
}
memset
(
pRead
,
0
,
sizeof
(
SWalReadHandle
));
pRead
->
pWal
=
pWal
;
pRead
->
readIdxTfd
=
-
1
;
pRead
->
readLogTfd
=
-
1
;
return
NULL
;
}
void
walCloseReadHandle
(
SWalReadHandle
*
pRead
)
{
tfClose
(
pRead
->
readIdxTfd
);
tfClose
(
pRead
->
readLogTfd
);
free
(
pRead
);
}
int32_t
walRegisterRead
(
SWalReadHandle
*
pRead
,
int64_t
ver
)
{
return
0
;
}
static
int32_t
walReadSeekFilePos
(
SWalReadHandle
*
pRead
,
int64_t
fileFirstVer
,
int64_t
ver
)
{
int
code
=
0
;
int64_t
idxTfd
=
pRead
->
readIdxTfd
;
int64_t
logTfd
=
pRead
->
readLogTfd
;
//seek position
int64_t
offset
=
(
ver
-
fileFirstVer
)
*
WAL_IDX_ENTRY_SIZE
;
code
=
tfLseek
(
idxTfd
,
offset
,
SEEK_SET
);
if
(
code
!=
0
)
{
return
-
1
;
}
WalIdxEntry
entry
;
code
=
tfRead
(
idxTfd
,
&
entry
,
sizeof
(
WalIdxEntry
));
if
(
code
!=
0
)
{
return
-
1
;
}
//TODO:deserialize
ASSERT
(
entry
.
ver
==
ver
);
code
=
tfLseek
(
logTfd
,
entry
.
offset
,
SEEK_SET
);
if
(
code
!=
0
)
{
return
-
1
;
}
return
code
;
}
static
int32_t
walReadChangeFile
(
SWalReadHandle
*
pRead
,
int64_t
fileFirstVer
)
{
char
fnameStr
[
WAL_FILE_LEN
];
tfClose
(
pRead
->
readIdxTfd
);
tfClose
(
pRead
->
readLogTfd
);
walBuildLogName
(
pRead
->
pWal
,
fileFirstVer
,
fnameStr
);
int
logTfd
=
tfOpenRead
(
fnameStr
);
if
(
logTfd
<
0
)
{
return
-
1
;
}
walBuildIdxName
(
pRead
->
pWal
,
fileFirstVer
,
fnameStr
);
int
idxTfd
=
tfOpenRead
(
fnameStr
);
if
(
idxTfd
<
0
)
{
return
-
1
;
}
pRead
->
readLogTfd
=
logTfd
;
pRead
->
readIdxTfd
=
idxTfd
;
return
0
;
}
static
int32_t
walReadSeekVer
(
SWalReadHandle
*
pRead
,
int64_t
ver
)
{
int
code
;
SWal
*
pWal
=
pRead
->
pWal
;
if
(
ver
==
pWal
->
vers
.
lastVer
)
{
return
0
;
}
if
(
ver
>
pWal
->
vers
.
lastVer
||
ver
<
pWal
->
vers
.
firstVer
)
{
return
-
1
;
}
if
(
ver
<
pWal
->
vers
.
snapshotVer
)
{
}
WalFileInfo
tmpInfo
;
tmpInfo
.
firstVer
=
ver
;
//bsearch in fileSet
WalFileInfo
*
pRet
=
taosArraySearch
(
pWal
->
fileInfoSet
,
&
tmpInfo
,
compareWalFileInfo
,
TD_LE
);
ASSERT
(
pRet
!=
NULL
);
if
(
pRead
->
curFileFirstVer
!=
pRet
->
firstVer
)
{
code
=
walReadChangeFile
(
pRead
,
pRet
->
firstVer
);
if
(
code
<
0
)
{
//TODO: set error flag
return
-
1
;
}
}
code
=
walReadSeekFilePos
(
pRead
,
pRet
->
firstVer
,
ver
);
if
(
code
<
0
)
{
return
-
1
;
}
pRead
->
curVersion
=
ver
;
return
0
;
}
int32_t
walReadWithHandle
(
SWalReadHandle
*
pRead
,
int64_t
ver
)
{
int
code
;
//TODO: check wal life
if
(
pRead
->
curVersion
!=
ver
)
{
walReadSeekVer
(
pRead
,
ver
);
}
if
(
!
tfValid
(
pRead
->
readLogTfd
))
return
-
1
;
if
(
sizeof
(
SWalHead
)
!=
tfRead
(
pRead
->
readLogTfd
,
&
pRead
->
head
,
sizeof
(
SWalHead
)))
{
return
-
1
;
}
code
=
walValidHeadCksum
(
&
pRead
->
head
);
if
(
code
!=
0
)
{
return
-
1
;
}
if
(
pRead
->
capacity
<
pRead
->
head
.
head
.
len
)
{
void
*
ptr
=
realloc
(
pRead
,
pRead
->
head
.
head
.
len
);
if
(
ptr
==
NULL
)
{
return
-
1
;
}
pRead
=
ptr
;
pRead
->
capacity
=
pRead
->
head
.
head
.
len
;
}
if
(
pRead
->
head
.
head
.
len
!=
tfRead
(
pRead
->
readLogTfd
,
&
pRead
->
head
.
head
.
body
,
pRead
->
head
.
head
.
len
))
{
return
-
1
;
}
code
=
walValidBodyCksum
(
&
pRead
->
head
);
if
(
code
!=
0
)
{
return
-
1
;
}
return
0
;
}
int32_t
walRead
(
SWal
*
pWal
,
SWalHead
**
ppHead
,
int64_t
ver
)
{
int
code
;
code
=
walSeekVer
(
pWal
,
ver
);
...
...
@@ -42,7 +183,7 @@ int32_t walRead(SWal *pWal, SWalHead **ppHead, int64_t ver) {
*
ppHead
=
NULL
;
return
-
1
;
}
if
(
tfRead
(
pWal
->
writeLogTfd
,
(
*
ppHead
)
->
head
.
cont
,
(
*
ppHead
)
->
head
.
len
)
!=
(
*
ppHead
)
->
head
.
len
)
{
if
(
tfRead
(
pWal
->
writeLogTfd
,
(
*
ppHead
)
->
head
.
body
,
(
*
ppHead
)
->
head
.
len
)
!=
(
*
ppHead
)
->
head
.
len
)
{
return
-
1
;
}
//TODO: endian compatibility processing after read
...
...
source/libs/wal/src/walSeek.c
浏览文件 @
793eb114
...
...
@@ -78,10 +78,12 @@ int walChangeFile(SWal *pWal, int64_t ver) {
code
=
tfClose
(
pWal
->
writeLogTfd
);
if
(
code
!=
0
)
{
//TODO
return
-
1
;
}
code
=
tfClose
(
pWal
->
writeIdxTfd
);
if
(
code
!=
0
)
{
//TODO
return
-
1
;
}
WalFileInfo
tmpInfo
;
tmpInfo
.
firstVer
=
ver
;
...
...
@@ -106,24 +108,19 @@ int walChangeFile(SWal *pWal, int64_t ver) {
pWal
->
writeLogTfd
=
logTfd
;
pWal
->
writeIdxTfd
=
idxTfd
;
return
code
;
}
int
walGetVerOffset
(
SWal
*
pWal
,
int64_t
ver
)
{
int
code
;
return
0
;
return
fileFirstVer
;
}
int
walSeekVer
(
SWal
*
pWal
,
int64_t
ver
)
{
int
code
;
if
(
ver
==
pWal
->
lastVersion
)
{
if
(
ver
==
pWal
->
vers
.
lastVer
)
{
return
0
;
}
if
(
ver
>
pWal
->
lastVersion
||
ver
<
pWal
->
firstVersion
)
{
if
(
ver
>
pWal
->
vers
.
lastVer
||
ver
<
pWal
->
vers
.
firstVer
)
{
return
-
1
;
}
if
(
ver
<
pWal
->
snapshotVersion
)
{
//TODO: set flag to prevent roll back
if
(
ver
<
pWal
->
vers
.
snapshotVer
)
{
}
if
(
ver
<
walGetCurFileFirstVer
(
pWal
)
||
(
ver
>
walGetCurFileLastVer
(
pWal
)))
{
code
=
walChangeFile
(
pWal
,
ver
);
...
...
source/libs/wal/src/walUtil.c
浏览文件 @
793eb114
...
...
@@ -17,6 +17,7 @@
#include "os.h"
#include "walInt.h"
#if 0
int32_t walGetNextFile(SWal *pWal, int64_t *nextFileId) {
int64_t curFileId = *nextFileId;
int64_t minFileId = INT64_MAX;
...
...
@@ -116,3 +117,4 @@ int32_t walGetNewFile(SWal *pWal, int64_t *newFileId) {
return 0;
}
#endif
source/libs/wal/src/walWrite.c
浏览文件 @
793eb114
...
...
@@ -114,22 +114,22 @@ void walRemoveAllOldFiles(void *handle) {
#endif
int32_t
walCommit
(
SWal
*
pWal
,
int64_t
ver
)
{
ASSERT
(
pWal
->
commitVersion
>=
pWal
->
snapshotVersion
);
ASSERT
(
pWal
->
commitVersion
<=
pWal
->
lastVersion
);
if
(
ver
<
pWal
->
commitVersion
||
ver
>
pWal
->
lastVersion
)
{
ASSERT
(
pWal
->
vers
.
commitVer
>=
pWal
->
vers
.
snapshotVer
);
ASSERT
(
pWal
->
vers
.
commitVer
<=
pWal
->
vers
.
lastVer
);
if
(
ver
<
pWal
->
vers
.
commitVer
||
ver
>
pWal
->
vers
.
lastVer
)
{
return
-
1
;
}
pWal
->
commitVersion
=
ver
;
pWal
->
vers
.
commitVer
=
ver
;
return
0
;
}
int32_t
walRollback
(
SWal
*
pWal
,
int64_t
ver
)
{
int
code
;
char
fnameStr
[
WAL_FILE_LEN
];
if
(
ver
==
pWal
->
lastVersion
)
{
if
(
ver
==
pWal
->
vers
.
lastVer
)
{
return
0
;
}
if
(
ver
>
pWal
->
lastVersion
||
ver
<
pWal
->
commitVersion
)
{
if
(
ver
>
pWal
->
vers
.
lastVer
||
ver
<
pWal
->
vers
.
commitVer
)
{
return
-
1
;
}
pthread_mutex_lock
(
&
pWal
->
mutex
);
...
...
@@ -220,7 +220,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
if
(
code
<
0
)
{
return
-
1
;
}
pWal
->
lastVersion
=
ver
-
1
;
pWal
->
vers
.
lastVer
=
ver
-
1
;
((
WalFileInfo
*
)
taosArrayGetLast
(
pWal
->
fileInfoSet
))
->
lastVer
=
ver
-
1
;
((
WalFileInfo
*
)
taosArrayGetLast
(
pWal
->
fileInfoSet
))
->
fileSize
=
entry
.
offset
;
...
...
@@ -230,9 +230,9 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
}
int32_t
walBeginTakeSnapshot
(
SWal
*
pWal
,
int64_t
ver
)
{
pWal
->
snapshottingVer
=
ver
;
pWal
->
vers
.
verInSnapshotting
=
ver
;
//check file rolling
if
(
pWal
->
retentionPeriod
==
0
)
{
if
(
pWal
->
cfg
.
retentionPeriod
==
0
)
{
walRoll
(
pWal
);
}
...
...
@@ -240,10 +240,10 @@ int32_t walBeginTakeSnapshot(SWal* pWal, int64_t ver) {
}
int32_t
walEndTakeSnapshot
(
SWal
*
pWal
)
{
int64_t
ver
=
pWal
->
snapshottingVer
;
int64_t
ver
=
pWal
->
vers
.
verInSnapshotting
;
if
(
ver
==
-
1
)
return
-
1
;
pWal
->
snapshotVersion
=
ver
;
pWal
->
vers
.
snapshotVer
=
ver
;
int
ts
=
taosGetTimestampSec
();
int
deleteCnt
=
0
;
...
...
@@ -257,8 +257,8 @@ int32_t walEndTakeSnapshot(SWal *pWal) {
}
//iterate files, until the searched result
for
(
WalFileInfo
*
iter
=
pWal
->
fileInfoSet
->
pData
;
iter
<
pInfo
;
iter
++
)
{
if
(
pWal
->
totSize
>
pWal
->
retentionSize
||
iter
->
closeTs
+
pWal
->
retentionPeriod
>
ts
)
{
if
(
pWal
->
totSize
>
pWal
->
cfg
.
retentionSize
||
iter
->
closeTs
+
pWal
->
cfg
.
retentionPeriod
>
ts
)
{
//delete according to file size or close time
deleteCnt
++
;
newTotSize
-=
iter
->
fileSize
;
...
...
@@ -278,13 +278,13 @@ int32_t walEndTakeSnapshot(SWal *pWal) {
taosArrayPopFrontBatch
(
pWal
->
fileInfoSet
,
deleteCnt
);
if
(
taosArrayGetSize
(
pWal
->
fileInfoSet
)
==
0
)
{
pWal
->
writeCur
=
-
1
;
pWal
->
firstVersion
=
-
1
;
pWal
->
vers
.
firstVer
=
-
1
;
}
else
{
pWal
->
firstVersion
=
((
WalFileInfo
*
)
taosArrayGet
(
pWal
->
fileInfoSet
,
0
))
->
firstVer
;
pWal
->
vers
.
firstVer
=
((
WalFileInfo
*
)
taosArrayGet
(
pWal
->
fileInfoSet
,
0
))
->
firstVer
;
}
pWal
->
writeCur
=
taosArrayGetSize
(
pWal
->
fileInfoSet
)
-
1
;;
pWal
->
totSize
=
newTotSize
;
pWal
->
snapshottingVer
=
-
1
;
pWal
->
vers
.
verInSnapshotting
=
-
1
;
//save snapshot ver, commit ver
int
code
=
walWriteMeta
(
pWal
);
...
...
@@ -311,7 +311,7 @@ int walRoll(SWal *pWal) {
}
int64_t
idxTfd
,
logTfd
;
//create new file
int64_t
newFileFirstVersion
=
pWal
->
lastVersion
+
1
;
int64_t
newFileFirstVersion
=
pWal
->
vers
.
lastVer
+
1
;
char
fnameStr
[
WAL_FILE_LEN
];
walBuildIdxName
(
pWal
,
newFileFirstVersion
,
fnameStr
);
idxTfd
=
tfOpenCreateWrite
(
fnameStr
);
...
...
@@ -357,18 +357,18 @@ int64_t walWrite(SWal *pWal, int64_t index, uint8_t msgType, const void *body, i
int
code
=
0
;
// no wal
if
(
pWal
->
level
==
TAOS_WAL_NOLOG
)
return
0
;
if
(
pWal
->
cfg
.
level
==
TAOS_WAL_NOLOG
)
return
0
;
if
(
index
==
pWal
->
lastVersion
+
1
)
{
if
(
index
==
pWal
->
vers
.
lastVer
+
1
)
{
if
(
taosArrayGetSize
(
pWal
->
fileInfoSet
)
==
0
)
{
pWal
->
firstVersion
=
index
;
pWal
->
vers
.
firstVer
=
index
;
code
=
walRoll
(
pWal
);
ASSERT
(
code
==
0
);
}
else
{
int64_t
passed
=
walGetSeq
()
-
pWal
->
lastRollSeq
;
if
(
pWal
->
rollPeriod
!=
-
1
&&
pWal
->
rollPeriod
!=
0
&&
passed
>
pWal
->
rollPeriod
)
{
if
(
pWal
->
cfg
.
rollPeriod
!=
-
1
&&
pWal
->
cfg
.
rollPeriod
!=
0
&&
passed
>
pWal
->
cfg
.
rollPeriod
)
{
walRoll
(
pWal
);
}
else
if
(
pWal
->
segSize
!=
-
1
&&
pWal
->
segSize
!=
0
&&
walGetLastFileSize
(
pWal
)
>
pWal
->
segSize
)
{
}
else
if
(
pWal
->
cfg
.
segSize
!=
-
1
&&
pWal
->
cfg
.
segSize
!=
0
&&
walGetLastFileSize
(
pWal
)
>
pWal
->
cfg
.
segSize
)
{
walRoll
(
pWal
);
}
}
...
...
@@ -380,23 +380,23 @@ int64_t walWrite(SWal *pWal, int64_t index, uint8_t msgType, const void *body, i
/*if (!tfValid(pWal->curLogTfd)) return 0;*/
pthread_mutex_lock
(
&
pWal
->
mutex
);
pWal
->
h
ead
.
head
.
version
=
index
;
pWal
->
writeH
ead
.
head
.
version
=
index
;
pWal
->
h
ead
.
head
.
len
=
bodyLen
;
pWal
->
h
ead
.
head
.
msgType
=
msgType
;
pWal
->
head
.
cksumHead
=
walCalcHeadCksum
(
&
pWal
->
h
ead
);
pWal
->
h
ead
.
cksumBody
=
walCalcBodyCksum
(
body
,
bodyLen
);
pWal
->
writeH
ead
.
head
.
len
=
bodyLen
;
pWal
->
writeH
ead
.
head
.
msgType
=
msgType
;
pWal
->
writeHead
.
cksumHead
=
walCalcHeadCksum
(
&
pWal
->
writeH
ead
);
pWal
->
writeH
ead
.
cksumBody
=
walCalcBodyCksum
(
body
,
bodyLen
);
if
(
tfWrite
(
pWal
->
writeLogTfd
,
&
pWal
->
h
ead
,
sizeof
(
SWalHead
))
!=
sizeof
(
SWalHead
))
{
if
(
tfWrite
(
pWal
->
writeLogTfd
,
&
pWal
->
writeH
ead
,
sizeof
(
SWalHead
))
!=
sizeof
(
SWalHead
))
{
//ftruncate
code
=
TAOS_SYSTEM_ERROR
(
errno
);
wError
(
"vgId:%d, file:%"
PRId64
".log, failed to write since %s"
,
pWal
->
vgId
,
walGetLastFileFirstVer
(
pWal
),
strerror
(
errno
));
wError
(
"vgId:%d, file:%"
PRId64
".log, failed to write since %s"
,
pWal
->
cfg
.
vgId
,
walGetLastFileFirstVer
(
pWal
),
strerror
(
errno
));
}
if
(
tfWrite
(
pWal
->
writeLogTfd
,
&
body
,
bodyLen
)
!=
bodyLen
)
{
//ftruncate
code
=
TAOS_SYSTEM_ERROR
(
errno
);
wError
(
"vgId:%d, file:%"
PRId64
".log, failed to write since %s"
,
pWal
->
vgId
,
walGetLastFileFirstVer
(
pWal
),
strerror
(
errno
));
wError
(
"vgId:%d, file:%"
PRId64
".log, failed to write since %s"
,
pWal
->
cfg
.
vgId
,
walGetLastFileFirstVer
(
pWal
),
strerror
(
errno
));
}
code
=
walWriteIndex
(
pWal
,
index
,
walGetCurFileOffset
(
pWal
));
if
(
code
!=
0
)
{
...
...
@@ -405,7 +405,7 @@ int64_t walWrite(SWal *pWal, int64_t index, uint8_t msgType, const void *body, i
}
//set status
pWal
->
lastVersion
=
index
;
pWal
->
vers
.
lastVer
=
index
;
pWal
->
totSize
+=
sizeof
(
SWalHead
)
+
bodyLen
;
walGetCurFileInfo
(
pWal
)
->
lastVer
=
index
;
walGetCurFileInfo
(
pWal
)
->
fileSize
+=
sizeof
(
SWalHead
)
+
bodyLen
;
...
...
@@ -416,10 +416,10 @@ int64_t walWrite(SWal *pWal, int64_t index, uint8_t msgType, const void *body, i
}
void
walFsync
(
SWal
*
pWal
,
bool
forceFsync
)
{
if
(
forceFsync
||
(
pWal
->
level
==
TAOS_WAL_FSYNC
&&
pWal
->
fsyncPeriod
==
0
))
{
wTrace
(
"vgId:%d, fileId:%"
PRId64
".log, do fsync"
,
pWal
->
vgId
,
walGetCurFileFirstVer
(
pWal
));
if
(
forceFsync
||
(
pWal
->
cfg
.
level
==
TAOS_WAL_FSYNC
&&
pWal
->
cfg
.
fsyncPeriod
==
0
))
{
wTrace
(
"vgId:%d, fileId:%"
PRId64
".log, do fsync"
,
pWal
->
cfg
.
vgId
,
walGetCurFileFirstVer
(
pWal
));
if
(
tfFsync
(
pWal
->
writeLogTfd
)
<
0
)
{
wError
(
"vgId:%d, file:%"
PRId64
".log, fsync failed since %s"
,
pWal
->
vgId
,
walGetCurFileFirstVer
(
pWal
),
strerror
(
errno
));
wError
(
"vgId:%d, file:%"
PRId64
".log, fsync failed since %s"
,
pWal
->
cfg
.
vgId
,
walGetCurFileFirstVer
(
pWal
),
strerror
(
errno
));
}
}
}
...
...
@@ -492,29 +492,29 @@ int32_t walGetWalFile(void *handle, char *fileName, int64_t *fileId) {
}
#endif
static
int
walValidateOffset
(
SWal
*
pWal
,
int64_t
ver
)
{
int
code
=
0
;
SWalHead
*
pHead
=
NULL
;
code
=
(
int
)
walRead
(
pWal
,
&
pHead
,
ver
);
if
(
pHead
->
head
.
version
!=
ver
)
{
return
-
1
;
}
return
0
;
}
/*static int walValidateOffset(SWal* pWal, int64_t ver) {*/
/*int code = 0;*/
/*SWalHead *pHead = NULL;*/
/*code = (int)walRead(pWal, &pHead, ver);*/
/*if(pHead->head.version != ver) {*/
/*return -1;*/
/*}*/
/*return 0;*/
/*}*/
static
int64_t
walGetOffset
(
SWal
*
pWal
,
int64_t
ver
)
{
int
code
=
walSeekVer
(
pWal
,
ver
);
if
(
code
!=
0
)
{
return
-
1
;
}
/*static int64_t walGetOffset(SWal* pWal, int64_t ver) {*/
/*int code = walSeekVer(pWal, ver);*/
/*if(code != 0) {*/
/*return -1;*/
/*}*/
code
=
walValidateOffset
(
pWal
,
ver
);
if
(
code
!=
0
)
{
return
-
1
;
}
/*code = walValidateOffset(pWal, ver);*/
/*if(code != 0) {*/
/*return -1;*/
/*}*/
return
0
;
}
/*return 0;*/
/*}*/
#if 0
static int32_t walSkipCorruptedRecord(SWal *pWal, SWalHead *pHead, int64_t tfd, int64_t *offset) {
...
...
source/libs/wal/test/walMetaTest.cpp
浏览文件 @
793eb114
...
...
@@ -24,7 +24,7 @@ class WalCleanEnv : public ::testing::Test {
pCfg
->
segSize
=
-
1
;
pCfg
->
retentionPeriod
=
0
;
pCfg
->
retentionSize
=
0
;
pCfg
->
walL
evel
=
TAOS_WAL_FSYNC
;
pCfg
->
l
evel
=
TAOS_WAL_FSYNC
;
pWal
=
walOpen
(
pathName
,
pCfg
);
free
(
pCfg
);
ASSERT
(
pWal
!=
NULL
);
...
...
@@ -56,7 +56,7 @@ class WalCleanDeleteEnv : public ::testing::Test {
memset
(
pCfg
,
0
,
sizeof
(
SWalCfg
));
pCfg
->
retentionPeriod
=
0
;
pCfg
->
retentionSize
=
0
;
pCfg
->
walL
evel
=
TAOS_WAL_FSYNC
;
pCfg
->
l
evel
=
TAOS_WAL_FSYNC
;
pWal
=
walOpen
(
pathName
,
pCfg
);
free
(
pCfg
);
ASSERT
(
pWal
!=
NULL
);
...
...
@@ -95,7 +95,7 @@ class WalKeepEnv : public ::testing::Test {
pCfg
->
segSize
=
-
1
;
pCfg
->
retentionPeriod
=
0
;
pCfg
->
retentionSize
=
0
;
pCfg
->
walL
evel
=
TAOS_WAL_FSYNC
;
pCfg
->
l
evel
=
TAOS_WAL_FSYNC
;
pWal
=
walOpen
(
pathName
,
pCfg
);
free
(
pCfg
);
ASSERT
(
pWal
!=
NULL
);
...
...
@@ -164,18 +164,18 @@ TEST_F(WalKeepEnv, readOldMeta) {
for
(
int
i
=
0
;
i
<
10
;
i
++
)
{
code
=
walWrite
(
pWal
,
i
,
i
+
1
,
(
void
*
)
ranStr
,
len
);
ASSERT_EQ
(
code
,
0
);
ASSERT_EQ
(
pWal
->
lastVersion
,
i
);
ASSERT_EQ
(
pWal
->
vers
.
lastVer
,
i
);
code
=
walWrite
(
pWal
,
i
+
2
,
i
,
(
void
*
)
ranStr
,
len
);
ASSERT_EQ
(
code
,
-
1
);
ASSERT_EQ
(
pWal
->
lastVersion
,
i
);
ASSERT_EQ
(
pWal
->
vers
.
lastVer
,
i
);
}
char
*
oldss
=
walMetaSerialize
(
pWal
);
TearDown
();
SetUp
();
ASSERT_EQ
(
pWal
->
firstVersion
,
0
);
ASSERT_EQ
(
pWal
->
lastVersion
,
9
);
ASSERT_EQ
(
pWal
->
vers
.
firstVer
,
0
);
ASSERT_EQ
(
pWal
->
vers
.
lastVer
,
9
);
char
*
newss
=
walMetaSerialize
(
pWal
);
...
...
@@ -195,10 +195,10 @@ TEST_F(WalCleanEnv, write) {
for
(
int
i
=
0
;
i
<
10
;
i
++
)
{
code
=
walWrite
(
pWal
,
i
,
i
+
1
,
(
void
*
)
ranStr
,
len
);
ASSERT_EQ
(
code
,
0
);
ASSERT_EQ
(
pWal
->
lastVersion
,
i
);
ASSERT_EQ
(
pWal
->
vers
.
lastVer
,
i
);
code
=
walWrite
(
pWal
,
i
+
2
,
i
,
(
void
*
)
ranStr
,
len
);
ASSERT_EQ
(
code
,
-
1
);
ASSERT_EQ
(
pWal
->
lastVersion
,
i
);
ASSERT_EQ
(
pWal
->
vers
.
lastVer
,
i
);
}
code
=
walWriteMeta
(
pWal
);
ASSERT_EQ
(
code
,
0
);
...
...
@@ -211,14 +211,14 @@ TEST_F(WalCleanEnv, rollback) {
for
(
int
i
=
0
;
i
<
10
;
i
++
)
{
code
=
walWrite
(
pWal
,
i
,
i
+
1
,
(
void
*
)
ranStr
,
len
);
ASSERT_EQ
(
code
,
0
);
ASSERT_EQ
(
pWal
->
lastVersion
,
i
);
ASSERT_EQ
(
pWal
->
vers
.
lastVer
,
i
);
}
code
=
walRollback
(
pWal
,
5
);
ASSERT_EQ
(
code
,
0
);
ASSERT_EQ
(
pWal
->
lastVersion
,
4
);
ASSERT_EQ
(
pWal
->
vers
.
lastVer
,
4
);
code
=
walRollback
(
pWal
,
3
);
ASSERT_EQ
(
code
,
0
);
ASSERT_EQ
(
pWal
->
lastVersion
,
2
);
ASSERT_EQ
(
pWal
->
vers
.
lastVer
,
2
);
code
=
walWriteMeta
(
pWal
);
ASSERT_EQ
(
code
,
0
);
}
...
...
@@ -231,16 +231,16 @@ TEST_F(WalCleanDeleteEnv, roll) {
for
(
i
=
0
;
i
<
100
;
i
++
)
{
code
=
walWrite
(
pWal
,
i
,
0
,
(
void
*
)
ranStr
,
len
);
ASSERT_EQ
(
code
,
0
);
ASSERT_EQ
(
pWal
->
lastVersion
,
i
);
ASSERT_EQ
(
pWal
->
vers
.
lastVer
,
i
);
code
=
walCommit
(
pWal
,
i
);
ASSERT_EQ
(
pWal
->
commitVersion
,
i
);
ASSERT_EQ
(
pWal
->
vers
.
commitVer
,
i
);
}
walBeginTakeSnapshot
(
pWal
,
i
-
1
);
ASSERT_EQ
(
pWal
->
snapshottingVer
,
i
-
1
);
ASSERT_EQ
(
pWal
->
vers
.
verInSnapshotting
,
i
-
1
);
walEndTakeSnapshot
(
pWal
);
ASSERT_EQ
(
pWal
->
snapshotVersion
,
i
-
1
);
ASSERT_EQ
(
pWal
->
snapshottingVer
,
-
1
);
ASSERT_EQ
(
pWal
->
vers
.
snapshotVer
,
i
-
1
);
ASSERT_EQ
(
pWal
->
vers
.
verInSnapshotting
,
-
1
);
code
=
walWrite
(
pWal
,
5
,
0
,
(
void
*
)
ranStr
,
len
);
ASSERT_NE
(
code
,
0
);
...
...
@@ -249,7 +249,7 @@ TEST_F(WalCleanDeleteEnv, roll) {
code
=
walWrite
(
pWal
,
i
,
0
,
(
void
*
)
ranStr
,
len
);
ASSERT_EQ
(
code
,
0
);
code
=
walCommit
(
pWal
,
i
);
ASSERT_EQ
(
pWal
->
commitVersion
,
i
);
ASSERT_EQ
(
pWal
->
vers
.
commitVer
,
i
);
}
//code = walWriteMeta(pWal);
...
...
source/util/src/tarray.c
浏览文件 @
793eb114
...
...
@@ -250,7 +250,7 @@ void taosArrayPopFrontBatch(SArray* pArray, size_t cnt) {
if
(
pArray
->
size
==
0
)
{
return
;
}
memmove
(
pArray
->
pData
,
(
char
*
)
pArray
->
pData
+
cnt
*
pArray
->
elemSize
,
pArray
->
size
);
memmove
(
pArray
->
pData
,
(
char
*
)
pArray
->
pData
+
cnt
*
pArray
->
elemSize
,
pArray
->
size
*
pArray
->
elemSize
);
}
void
taosArrayPopTailBatch
(
SArray
*
pArray
,
size_t
cnt
)
{
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录