Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
930b8500
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1184
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,发现更多精彩内容 >>
提交
930b8500
编写于
5月 23, 2022
作者:
L
Liu Jicong
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
refacotr: remove tq meta store
上级
4d290984
变更
5
展开全部
显示空白变更内容
内联
并排
Showing
5 changed file
with
1 addition
and
1194 deletion
+1
-1194
source/dnode/vnode/CMakeLists.txt
source/dnode/vnode/CMakeLists.txt
+0
-1
source/dnode/vnode/src/inc/tq.h
source/dnode/vnode/src/inc/tq.h
+0
-163
source/dnode/vnode/src/tq/tq.c
source/dnode/vnode/src/tq/tq.c
+1
-129
source/dnode/vnode/src/tq/tqMetaStore.c
source/dnode/vnode/src/tq/tqMetaStore.c
+0
-622
source/dnode/vnode/test/tqMetaTest.cpp
source/dnode/vnode/test/tqMetaTest.cpp
+0
-279
未找到文件。
source/dnode/vnode/CMakeLists.txt
浏览文件 @
930b8500
...
...
@@ -48,7 +48,6 @@ target_sources(
# tq
"src/tq/tq.c"
"src/tq/tqCommit.c"
"src/tq/tqMetaStore.c"
"src/tq/tqOffset.c"
"src/tq/tqPush.c"
"src/tq/tqRead.c"
...
...
source/dnode/vnode/src/inc/tq.h
浏览文件 @
930b8500
...
...
@@ -41,45 +41,6 @@ extern "C" {
#define tqTrace(...) do { if (tqDebugFlag & DEBUG_TRACE) { taosPrintLog("TQ ", DEBUG_TRACE, tqDebugFlag, __VA_ARGS__); }} while(0)
// clang-format on
#define TQ_BUFFER_SIZE 4
#define TQ_BUCKET_MASK 0xFF
#define TQ_BUCKET_SIZE 256
#define TQ_PAGE_SIZE 4096
// key + offset + size
#define TQ_IDX_SIZE 24
// 4096 / 24
#define TQ_MAX_IDX_ONE_PAGE 170
// 24 * 170
#define TQ_IDX_PAGE_BODY_SIZE 4080
// 4096 - 4080
#define TQ_IDX_PAGE_HEAD_SIZE 16
#define TQ_ACTION_CONST 0
#define TQ_ACTION_INUSE 1
#define TQ_ACTION_INUSE_CONT 2
#define TQ_ACTION_INTXN 3
#define TQ_SVER 0
// TODO: inplace mode is not implemented
#define TQ_UPDATE_INPLACE 0
#define TQ_UPDATE_APPEND 1
#define TQ_DUP_INTXN_REWRITE 0
#define TQ_DUP_INTXN_REJECT 2
static
inline
bool
tqUpdateAppend
(
int32_t
tqConfigFlag
)
{
return
tqConfigFlag
&
TQ_UPDATE_APPEND
;
}
static
inline
bool
tqDupIntxnReject
(
int32_t
tqConfigFlag
)
{
return
tqConfigFlag
&
TQ_DUP_INTXN_REJECT
;
}
static
const
int8_t
TQ_CONST_DELETE
=
TQ_ACTION_CONST
;
#define TQ_DELETE_TOKEN (void*)&TQ_CONST_DELETE
typedef
enum
{
TQ_ITEM_READY
,
TQ_ITEM_PROCESS
,
TQ_ITEM_EMPTY
}
STqItemStatus
;
typedef
struct
STqOffsetCfg
STqOffsetCfg
;
typedef
struct
STqOffsetStore
STqOffsetStore
;
...
...
@@ -98,53 +59,6 @@ struct STqReadHandle {
STSchema
*
pSchema
;
};
typedef
struct
{
int16_t
ver
;
int16_t
action
;
int32_t
checksum
;
int64_t
ssize
;
char
content
[];
}
STqSerializedHead
;
typedef
int32_t
(
*
FTqSerialize
)(
const
void
*
pObj
,
STqSerializedHead
**
ppHead
);
typedef
int32_t
(
*
FTqDeserialize
)(
void
*
self
,
const
STqSerializedHead
*
pHead
,
void
**
ppObj
);
typedef
void
(
*
FTqDelete
)(
void
*
);
typedef
struct
{
int64_t
key
;
int64_t
offset
;
int64_t
serializedSize
;
void
*
valueInUse
;
void
*
valueInTxn
;
}
STqMetaHandle
;
typedef
struct
STqMetaList
{
STqMetaHandle
handle
;
struct
STqMetaList
*
next
;
// struct STqMetaList* inTxnPrev;
// struct STqMetaList* inTxnNext;
struct
STqMetaList
*
unpersistPrev
;
struct
STqMetaList
*
unpersistNext
;
}
STqMetaList
;
typedef
struct
{
STQ
*
pTq
;
STqMetaList
*
bucket
[
TQ_BUCKET_SIZE
];
// a table head
STqMetaList
*
unpersistHead
;
// topics that are not connectted
STqMetaList
*
unconnectTopic
;
TdFilePtr
pFile
;
TdFilePtr
pIdxFile
;
char
*
dirPath
;
int32_t
tqConfigFlag
;
FTqSerialize
pSerializer
;
FTqDeserialize
pDeserializer
;
FTqDelete
pDeleter
;
}
STqMetaStore
;
typedef
struct
{
int64_t
consumerId
;
int32_t
epoch
;
...
...
@@ -189,87 +103,10 @@ typedef struct {
static
STqMgmt
tqMgmt
;
typedef
struct
{
int8_t
status
;
int64_t
offset
;
qTaskInfo_t
task
;
STqReadHandle
*
pReadHandle
;
}
STqTaskItem
;
// new version
typedef
struct
{
int64_t
firstOffset
;
int64_t
lastOffset
;
STqTaskItem
output
[
TQ_BUFFER_SIZE
];
}
STqBuffer
;
typedef
struct
{
char
topicName
[
TSDB_TOPIC_FNAME_LEN
];
char
*
sql
;
char
*
logicalPlan
;
char
*
physicalPlan
;
char
*
qmsg
;
STqBuffer
buffer
;
SWalReadHandle
*
pReadhandle
;
}
STqTopic
;
typedef
struct
{
int64_t
consumerId
;
int32_t
epoch
;
char
cgroup
[
TSDB_TOPIC_FNAME_LEN
];
SArray
*
topics
;
// SArray<STqTopic>
}
STqConsumer
;
typedef
struct
{
int8_t
type
;
int8_t
nodeType
;
int8_t
reserved
[
6
];
int64_t
streamId
;
qTaskInfo_t
task
;
// TODO sync function
}
STqStreamPusher
;
typedef
struct
{
int8_t
inited
;
tmr_h
timer
;
}
STqPushMgmt
;
static
STqPushMgmt
tqPushMgmt
;
// init once
int
tqInit
();
void
tqCleanUp
();
// open in each vnode
// required by vnode
int32_t
tqSerializeConsumer
(
const
STqConsumer
*
,
STqSerializedHead
**
);
int32_t
tqDeserializeConsumer
(
STQ
*
,
const
STqSerializedHead
*
,
STqConsumer
**
);
static
int
FORCE_INLINE
tqQueryExecuting
(
int32_t
status
)
{
return
status
;
}
// tqMetaStore.h
STqMetaStore
*
tqStoreOpen
(
STQ
*
pTq
,
const
char
*
path
,
FTqSerialize
pSerializer
,
FTqDeserialize
pDeserializer
,
FTqDelete
pDeleter
,
int32_t
tqConfigFlag
);
int32_t
tqStoreClose
(
STqMetaStore
*
);
// int32_t tqStoreDelete(TqMetaStore*);
// int32_t tqStoreCommitAll(TqMetaStore*);
int32_t
tqStorePersist
(
STqMetaStore
*
);
// clean deleted idx and data from persistent file
int32_t
tqStoreCompact
(
STqMetaStore
*
);
void
*
tqHandleGet
(
STqMetaStore
*
,
int64_t
key
);
// make it unpersist
void
*
tqHandleTouchGet
(
STqMetaStore
*
,
int64_t
key
);
int32_t
tqHandleMovePut
(
STqMetaStore
*
,
int64_t
key
,
void
*
value
);
int32_t
tqHandleCopyPut
(
STqMetaStore
*
,
int64_t
key
,
void
*
value
,
size_t
vsize
);
// delete committed kv pair
// notice that a delete action still needs to be committed
int32_t
tqHandleDel
(
STqMetaStore
*
,
int64_t
key
);
int32_t
tqHandlePurge
(
STqMetaStore
*
,
int64_t
key
);
int32_t
tqHandleCommit
(
STqMetaStore
*
,
int64_t
key
);
int32_t
tqHandleAbort
(
STqMetaStore
*
,
int64_t
key
);
// tqOffset
STqOffsetStore
*
STqOffsetOpen
(
STqOffsetCfg
*
);
void
STqOffsetClose
(
STqOffsetStore
*
);
...
...
source/dnode/vnode/src/tq/tq.c
浏览文件 @
930b8500
...
...
@@ -214,7 +214,7 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
if
(
taosHashGetSize
(
pTq
->
pStreamTasks
)
==
0
)
return
0
;
if
(
tdUpdateExpireWindow
(
pTq
->
pVnode
->
pSma
,
msg
,
ver
)
!=
0
)
{
// TODO
error handle
// TODO
handle sma error
}
void
*
data
=
taosMemoryMalloc
(
msgLen
);
if
(
data
==
NULL
)
{
...
...
@@ -230,134 +230,6 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
int
tqCommit
(
STQ
*
pTq
)
{
// do nothing
/*return tqStorePersist(pTq->tqMeta);*/
return
0
;
}
int32_t
tqGetTopicHandleSize
(
const
STqTopic
*
pTopic
)
{
return
strlen
(
pTopic
->
topicName
)
+
strlen
(
pTopic
->
sql
)
+
strlen
(
pTopic
->
physicalPlan
)
+
strlen
(
pTopic
->
qmsg
)
+
sizeof
(
int64_t
)
*
3
;
}
int32_t
tqGetConsumerHandleSize
(
const
STqConsumer
*
pConsumer
)
{
int
num
=
taosArrayGetSize
(
pConsumer
->
topics
);
int32_t
sz
=
0
;
for
(
int
i
=
0
;
i
<
num
;
i
++
)
{
STqTopic
*
pTopic
=
taosArrayGet
(
pConsumer
->
topics
,
i
);
sz
+=
tqGetTopicHandleSize
(
pTopic
);
}
return
sz
;
}
static
FORCE_INLINE
int32_t
tEncodeSTqTopic
(
void
**
buf
,
const
STqTopic
*
pTopic
)
{
int32_t
tlen
=
0
;
tlen
+=
taosEncodeString
(
buf
,
pTopic
->
topicName
);
/*tlen += taosEncodeString(buf, pTopic->sql);*/
/*tlen += taosEncodeString(buf, pTopic->physicalPlan);*/
tlen
+=
taosEncodeString
(
buf
,
pTopic
->
qmsg
);
/*tlen += taosEncodeFixedI64(buf, pTopic->persistedOffset);*/
/*tlen += taosEncodeFixedI64(buf, pTopic->committedOffset);*/
/*tlen += taosEncodeFixedI64(buf, pTopic->currentOffset);*/
return
tlen
;
}
static
FORCE_INLINE
const
void
*
tDecodeSTqTopic
(
const
void
*
buf
,
STqTopic
*
pTopic
)
{
buf
=
taosDecodeStringTo
(
buf
,
pTopic
->
topicName
);
/*buf = taosDecodeString(buf, &pTopic->sql);*/
/*buf = taosDecodeString(buf, &pTopic->physicalPlan);*/
buf
=
taosDecodeString
(
buf
,
&
pTopic
->
qmsg
);
/*buf = taosDecodeFixedI64(buf, &pTopic->persistedOffset);*/
/*buf = taosDecodeFixedI64(buf, &pTopic->committedOffset);*/
/*buf = taosDecodeFixedI64(buf, &pTopic->currentOffset);*/
return
buf
;
}
static
FORCE_INLINE
int32_t
tEncodeSTqConsumer
(
void
**
buf
,
const
STqConsumer
*
pConsumer
)
{
int32_t
sz
;
int32_t
tlen
=
0
;
tlen
+=
taosEncodeFixedI64
(
buf
,
pConsumer
->
consumerId
);
tlen
+=
taosEncodeFixedI32
(
buf
,
pConsumer
->
epoch
);
tlen
+=
taosEncodeString
(
buf
,
pConsumer
->
cgroup
);
sz
=
taosArrayGetSize
(
pConsumer
->
topics
);
tlen
+=
taosEncodeFixedI32
(
buf
,
sz
);
for
(
int32_t
i
=
0
;
i
<
sz
;
i
++
)
{
STqTopic
*
pTopic
=
taosArrayGet
(
pConsumer
->
topics
,
i
);
tlen
+=
tEncodeSTqTopic
(
buf
,
pTopic
);
}
return
tlen
;
}
static
FORCE_INLINE
const
void
*
tDecodeSTqConsumer
(
const
void
*
buf
,
STqConsumer
*
pConsumer
)
{
int32_t
sz
;
buf
=
taosDecodeFixedI64
(
buf
,
&
pConsumer
->
consumerId
);
buf
=
taosDecodeFixedI32
(
buf
,
&
pConsumer
->
epoch
);
buf
=
taosDecodeStringTo
(
buf
,
pConsumer
->
cgroup
);
buf
=
taosDecodeFixedI32
(
buf
,
&
sz
);
pConsumer
->
topics
=
taosArrayInit
(
sz
,
sizeof
(
STqTopic
));
if
(
pConsumer
->
topics
==
NULL
)
return
NULL
;
for
(
int32_t
i
=
0
;
i
<
sz
;
i
++
)
{
STqTopic
pTopic
;
buf
=
tDecodeSTqTopic
(
buf
,
&
pTopic
);
taosArrayPush
(
pConsumer
->
topics
,
&
pTopic
);
}
return
buf
;
}
int
tqSerializeConsumer
(
const
STqConsumer
*
pConsumer
,
STqSerializedHead
**
ppHead
)
{
int32_t
sz
=
tEncodeSTqConsumer
(
NULL
,
pConsumer
);
if
(
sz
>
(
*
ppHead
)
->
ssize
)
{
void
*
tmpPtr
=
taosMemoryRealloc
(
*
ppHead
,
sizeof
(
STqSerializedHead
)
+
sz
);
if
(
tmpPtr
==
NULL
)
{
taosMemoryFree
(
*
ppHead
);
terrno
=
TSDB_CODE_TQ_OUT_OF_MEMORY
;
return
-
1
;
}
*
ppHead
=
tmpPtr
;
(
*
ppHead
)
->
ssize
=
sz
;
}
void
*
ptr
=
(
*
ppHead
)
->
content
;
void
*
abuf
=
ptr
;
tEncodeSTqConsumer
(
&
abuf
,
pConsumer
);
return
0
;
}
int32_t
tqDeserializeConsumer
(
STQ
*
pTq
,
const
STqSerializedHead
*
pHead
,
STqConsumer
**
ppConsumer
)
{
const
void
*
str
=
pHead
->
content
;
*
ppConsumer
=
taosMemoryCalloc
(
1
,
sizeof
(
STqConsumer
));
if
(
*
ppConsumer
==
NULL
)
{
terrno
=
TSDB_CODE_TQ_OUT_OF_MEMORY
;
return
-
1
;
}
if
(
tDecodeSTqConsumer
(
str
,
*
ppConsumer
)
==
NULL
)
{
terrno
=
TSDB_CODE_TQ_OUT_OF_MEMORY
;
return
-
1
;
}
STqConsumer
*
pConsumer
=
*
ppConsumer
;
int32_t
sz
=
taosArrayGetSize
(
pConsumer
->
topics
);
for
(
int32_t
i
=
0
;
i
<
sz
;
i
++
)
{
STqTopic
*
pTopic
=
taosArrayGet
(
pConsumer
->
topics
,
i
);
pTopic
->
pReadhandle
=
walOpenReadHandle
(
pTq
->
pWal
);
if
(
pTopic
->
pReadhandle
==
NULL
)
{
ASSERT
(
false
);
}
for
(
int
j
=
0
;
j
<
TQ_BUFFER_SIZE
;
j
++
)
{
pTopic
->
buffer
.
output
[
j
].
status
=
0
;
STqReadHandle
*
pReadHandle
=
tqInitSubmitMsgScanner
(
pTq
->
pVnode
->
pMeta
);
SReadHandle
handle
=
{
.
reader
=
pReadHandle
,
.
meta
=
pTq
->
pVnode
->
pMeta
,
.
pMsgCb
=
&
pTq
->
pVnode
->
msgCb
,
};
pTopic
->
buffer
.
output
[
j
].
pReadHandle
=
pReadHandle
;
pTopic
->
buffer
.
output
[
j
].
task
=
qCreateStreamExecTaskInfo
(
pTopic
->
qmsg
,
&
handle
);
}
}
return
0
;
}
...
...
source/dnode/vnode/src/tq/tqMetaStore.c
已删除
100644 → 0
浏览文件 @
4d290984
此差异已折叠。
点击以展开。
source/dnode/vnode/test/tqMetaTest.cpp
已删除
100644 → 0
浏览文件 @
4d290984
#include <gtest/gtest.h>
#include <cstring>
#include <iostream>
#include <queue>
#include "tqMetaStore.h"
struct
Foo
{
int32_t
a
;
};
int
FooSerializer
(
const
void
*
pObj
,
STqSerializedHead
**
ppHead
)
{
Foo
*
foo
=
(
Foo
*
)
pObj
;
if
((
*
ppHead
)
==
NULL
||
(
*
ppHead
)
->
ssize
<
sizeof
(
STqSerializedHead
)
+
sizeof
(
int32_t
))
{
*
ppHead
=
(
STqSerializedHead
*
)
taosMemoryRealloc
(
*
ppHead
,
sizeof
(
STqSerializedHead
)
+
sizeof
(
int32_t
));
(
*
ppHead
)
->
ssize
=
sizeof
(
STqSerializedHead
)
+
sizeof
(
int32_t
);
}
*
(
int32_t
*
)(
*
ppHead
)
->
content
=
foo
->
a
;
return
(
*
ppHead
)
->
ssize
;
}
const
void
*
FooDeserializer
(
const
STqSerializedHead
*
pHead
,
void
**
ppObj
)
{
if
(
*
ppObj
==
NULL
)
{
*
ppObj
=
taosMemoryRealloc
(
*
ppObj
,
sizeof
(
int32_t
));
}
Foo
*
pFoo
=
*
(
Foo
**
)
ppObj
;
pFoo
->
a
=
*
(
int32_t
*
)
pHead
->
content
;
return
NULL
;
}
void
FooDeleter
(
void
*
pObj
)
{
taosMemoryFree
(
pObj
);
}
class
TqMetaUpdateAppendTest
:
public
::
testing
::
Test
{
protected:
void
SetUp
()
override
{
taosRemoveDir
(
pathName
);
pMeta
=
tqStoreOpen
(
pathName
,
FooSerializer
,
FooDeserializer
,
FooDeleter
,
TQ_UPDATE_APPEND
);
ASSERT
(
pMeta
);
}
void
TearDown
()
override
{
tqStoreClose
(
pMeta
);
}
STqMetaStore
*
pMeta
;
const
char
*
pathName
=
TD_TMP_DIR_PATH
"tq_test"
;
};
TEST_F
(
TqMetaUpdateAppendTest
,
copyPutTest
)
{
Foo
foo
;
foo
.
a
=
3
;
tqHandleCopyPut
(
pMeta
,
1
,
&
foo
,
sizeof
(
Foo
));
Foo
*
pFoo
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
1
);
EXPECT_EQ
(
pFoo
==
NULL
,
true
);
tqHandleCommit
(
pMeta
,
1
);
pFoo
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
1
);
EXPECT_EQ
(
pFoo
->
a
,
3
);
}
TEST_F
(
TqMetaUpdateAppendTest
,
persistTest
)
{
Foo
*
pFoo
=
(
Foo
*
)
taosMemoryMalloc
(
sizeof
(
Foo
));
pFoo
->
a
=
2
;
tqHandleMovePut
(
pMeta
,
1
,
pFoo
);
Foo
*
pBar
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
1
);
EXPECT_EQ
(
pBar
==
NULL
,
true
);
tqHandleCommit
(
pMeta
,
1
);
pBar
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
1
);
EXPECT_EQ
(
pBar
->
a
,
pFoo
->
a
);
pBar
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
2
);
EXPECT_EQ
(
pBar
==
NULL
,
true
);
tqStoreClose
(
pMeta
);
pMeta
=
tqStoreOpen
(
pathName
,
FooSerializer
,
FooDeserializer
,
FooDeleter
,
TQ_UPDATE_APPEND
);
ASSERT
(
pMeta
);
pBar
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
1
);
ASSERT_EQ
(
pBar
!=
NULL
,
true
);
EXPECT_EQ
(
pBar
->
a
,
2
);
pBar
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
2
);
EXPECT_EQ
(
pBar
==
NULL
,
true
);
}
TEST_F
(
TqMetaUpdateAppendTest
,
uncommittedTest
)
{
Foo
*
pFoo
=
(
Foo
*
)
taosMemoryMalloc
(
sizeof
(
Foo
));
pFoo
->
a
=
3
;
tqHandleMovePut
(
pMeta
,
1
,
pFoo
);
pFoo
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
1
);
EXPECT_EQ
(
pFoo
==
NULL
,
true
);
}
TEST_F
(
TqMetaUpdateAppendTest
,
abortTest
)
{
Foo
*
pFoo
=
(
Foo
*
)
taosMemoryMalloc
(
sizeof
(
Foo
));
pFoo
->
a
=
3
;
tqHandleMovePut
(
pMeta
,
1
,
pFoo
);
pFoo
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
1
);
EXPECT_EQ
(
pFoo
==
NULL
,
true
);
tqHandleAbort
(
pMeta
,
1
);
pFoo
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
1
);
EXPECT_EQ
(
pFoo
==
NULL
,
true
);
}
TEST_F
(
TqMetaUpdateAppendTest
,
deleteTest
)
{
Foo
*
pFoo
=
(
Foo
*
)
taosMemoryMalloc
(
sizeof
(
Foo
));
pFoo
->
a
=
3
;
tqHandleMovePut
(
pMeta
,
1
,
pFoo
);
pFoo
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
1
);
EXPECT_EQ
(
pFoo
==
NULL
,
true
);
tqHandleCommit
(
pMeta
,
1
);
pFoo
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
1
);
ASSERT_EQ
(
pFoo
!=
NULL
,
true
);
EXPECT_EQ
(
pFoo
->
a
,
3
);
tqHandleDel
(
pMeta
,
1
);
pFoo
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
1
);
ASSERT_EQ
(
pFoo
!=
NULL
,
true
);
EXPECT_EQ
(
pFoo
->
a
,
3
);
tqHandleCommit
(
pMeta
,
1
);
pFoo
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
1
);
EXPECT_EQ
(
pFoo
==
NULL
,
true
);
tqStoreClose
(
pMeta
);
pMeta
=
tqStoreOpen
(
pathName
,
FooSerializer
,
FooDeserializer
,
FooDeleter
,
TQ_UPDATE_APPEND
);
ASSERT
(
pMeta
);
pFoo
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
1
);
EXPECT_EQ
(
pFoo
==
NULL
,
true
);
}
TEST_F
(
TqMetaUpdateAppendTest
,
intxnPersist
)
{
Foo
*
pFoo
=
(
Foo
*
)
taosMemoryMalloc
(
sizeof
(
Foo
));
pFoo
->
a
=
3
;
tqHandleMovePut
(
pMeta
,
1
,
pFoo
);
tqHandleCommit
(
pMeta
,
1
);
Foo
*
pBar
=
(
Foo
*
)
taosMemoryMalloc
(
sizeof
(
Foo
));
pBar
->
a
=
4
;
tqHandleMovePut
(
pMeta
,
1
,
pBar
);
Foo
*
pFoo1
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
1
);
EXPECT_EQ
(
pFoo1
->
a
,
3
);
tqStoreClose
(
pMeta
);
pMeta
=
tqStoreOpen
(
pathName
,
FooSerializer
,
FooDeserializer
,
FooDeleter
,
TQ_UPDATE_APPEND
);
ASSERT
(
pMeta
);
pFoo1
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
1
);
EXPECT_EQ
(
pFoo1
->
a
,
3
);
tqHandleCommit
(
pMeta
,
1
);
pFoo1
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
1
);
EXPECT_EQ
(
pFoo1
->
a
,
4
);
tqStoreClose
(
pMeta
);
pMeta
=
tqStoreOpen
(
pathName
,
FooSerializer
,
FooDeserializer
,
FooDeleter
,
TQ_UPDATE_APPEND
);
ASSERT
(
pMeta
);
pFoo1
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
1
);
EXPECT_EQ
(
pFoo1
->
a
,
4
);
}
TEST_F
(
TqMetaUpdateAppendTest
,
multiplePage
)
{
taosSeedRand
(
0
);
std
::
vector
<
int
>
v
;
for
(
int
i
=
0
;
i
<
1000
;
i
++
)
{
v
.
push_back
(
taosRand
());
Foo
foo
;
foo
.
a
=
v
[
i
];
tqHandleCopyPut
(
pMeta
,
i
,
&
foo
,
sizeof
(
Foo
));
}
for
(
int
i
=
0
;
i
<
500
;
i
++
)
{
tqHandleCommit
(
pMeta
,
i
);
Foo
*
pFoo
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
i
);
ASSERT_EQ
(
pFoo
!=
NULL
,
true
)
<<
" at idx "
<<
i
<<
"
\n
"
;
EXPECT_EQ
(
pFoo
->
a
,
v
[
i
]);
}
tqStoreClose
(
pMeta
);
pMeta
=
tqStoreOpen
(
pathName
,
FooSerializer
,
FooDeserializer
,
FooDeleter
,
TQ_UPDATE_APPEND
);
ASSERT
(
pMeta
);
for
(
int
i
=
500
;
i
<
1000
;
i
++
)
{
tqHandleCommit
(
pMeta
,
i
);
Foo
*
pFoo
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
i
);
ASSERT_EQ
(
pFoo
!=
NULL
,
true
)
<<
" at idx "
<<
i
<<
"
\n
"
;
EXPECT_EQ
(
pFoo
->
a
,
v
[
i
]);
}
for
(
int
i
=
0
;
i
<
1000
;
i
++
)
{
Foo
*
pFoo
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
i
);
ASSERT_EQ
(
pFoo
!=
NULL
,
true
)
<<
" at idx "
<<
i
<<
"
\n
"
;
EXPECT_EQ
(
pFoo
->
a
,
v
[
i
]);
}
}
TEST_F
(
TqMetaUpdateAppendTest
,
multipleRewrite
)
{
taosSeedRand
(
0
);
std
::
vector
<
int
>
v
;
for
(
int
i
=
0
;
i
<
1000
;
i
++
)
{
v
.
push_back
(
taosRand
());
Foo
foo
;
foo
.
a
=
v
[
i
];
tqHandleCopyPut
(
pMeta
,
i
,
&
foo
,
sizeof
(
Foo
));
}
for
(
int
i
=
0
;
i
<
500
;
i
++
)
{
tqHandleCommit
(
pMeta
,
i
);
v
[
i
]
=
taosRand
();
Foo
foo
;
foo
.
a
=
v
[
i
];
tqHandleCopyPut
(
pMeta
,
i
,
&
foo
,
sizeof
(
Foo
));
}
for
(
int
i
=
500
;
i
<
1000
;
i
++
)
{
v
[
i
]
=
taosRand
();
Foo
foo
;
foo
.
a
=
v
[
i
];
tqHandleCopyPut
(
pMeta
,
i
,
&
foo
,
sizeof
(
Foo
));
}
for
(
int
i
=
0
;
i
<
1000
;
i
++
)
{
tqHandleCommit
(
pMeta
,
i
);
}
tqStoreClose
(
pMeta
);
pMeta
=
tqStoreOpen
(
pathName
,
FooSerializer
,
FooDeserializer
,
FooDeleter
,
TQ_UPDATE_APPEND
);
ASSERT
(
pMeta
);
for
(
int
i
=
500
;
i
<
1000
;
i
++
)
{
v
[
i
]
=
taosRand
();
Foo
foo
;
foo
.
a
=
v
[
i
];
tqHandleCopyPut
(
pMeta
,
i
,
&
foo
,
sizeof
(
Foo
));
tqHandleCommit
(
pMeta
,
i
);
}
for
(
int
i
=
0
;
i
<
1000
;
i
++
)
{
Foo
*
pFoo
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
i
);
ASSERT_EQ
(
pFoo
!=
NULL
,
true
)
<<
" at idx "
<<
i
<<
"
\n
"
;
EXPECT_EQ
(
pFoo
->
a
,
v
[
i
]);
}
}
TEST_F
(
TqMetaUpdateAppendTest
,
dupCommit
)
{
taosSeedRand
(
0
);
std
::
vector
<
int
>
v
;
for
(
int
i
=
0
;
i
<
1000
;
i
++
)
{
v
.
push_back
(
taosRand
());
Foo
foo
;
foo
.
a
=
v
[
i
];
tqHandleCopyPut
(
pMeta
,
i
,
&
foo
,
sizeof
(
Foo
));
}
for
(
int
i
=
0
;
i
<
1000
;
i
++
)
{
int
ret
=
tqHandleCommit
(
pMeta
,
i
);
EXPECT_EQ
(
ret
,
0
);
ret
=
tqHandleCommit
(
pMeta
,
i
);
EXPECT_EQ
(
ret
,
-
1
);
}
for
(
int
i
=
0
;
i
<
1000
;
i
++
)
{
int
ret
=
tqHandleCommit
(
pMeta
,
i
);
EXPECT_EQ
(
ret
,
-
1
);
}
for
(
int
i
=
0
;
i
<
1000
;
i
++
)
{
Foo
*
pFoo
=
(
Foo
*
)
tqHandleGet
(
pMeta
,
i
);
ASSERT_EQ
(
pFoo
!=
NULL
,
true
)
<<
" at idx "
<<
i
<<
"
\n
"
;
EXPECT_EQ
(
pFoo
->
a
,
v
[
i
]);
}
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录