Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
98899a12
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,发现更多精彩内容 >>
提交
98899a12
编写于
11月 02, 2021
作者:
S
Shengliang Guan
浏览文件
操作
浏览文件
下载
差异文件
Merge remote-tracking branch 'origin/3.0' into feature/dnode3
上级
cfae3b94
06f55a36
变更
4
显示空白变更内容
内联
并排
Showing
4 changed file
with
355 addition
and
61 deletion
+355
-61
include/server/vnode/tq/tq.h
include/server/vnode/tq/tq.h
+3
-0
source/dnode/vnode/tq/inc/tqMetaStore.h
source/dnode/vnode/tq/inc/tqMetaStore.h
+34
-21
source/dnode/vnode/tq/src/tq.c
source/dnode/vnode/tq/src/tq.c
+18
-18
source/dnode/vnode/tq/src/tqMetaStore.c
source/dnode/vnode/tq/src/tqMetaStore.c
+300
-22
未找到文件。
include/server/vnode/tq/tq.h
浏览文件 @
98899a12
...
...
@@ -19,6 +19,9 @@
#include "os.h"
#include "tutil.h"
#define TQ_ACTION_INSERT 0x7f7f7f7fULL
#define TQ_ACTION_DELETE 0x80808080ULL
#ifdef __cplusplus
extern
"C"
{
#endif
...
...
source/dnode/vnode/tq/inc/tqMetaStore.h
浏览文件 @
98899a12
...
...
@@ -17,9 +17,20 @@
#define _TQ_META_STORE_H_
#include "os.h"
#include "tq.h"
#define TQ_
INUSE
_SIZE 0xFF
#define TQ_
BUCKET
_SIZE 0xFF
#define TQ_PAGE_SIZE 4096
//key + offset + size
#define TQ_IDX_ENTRY_SIZE 24
inline
static
int
TqMaxEntryOnePage
()
{
//170
return
TQ_PAGE_SIZE
/
TQ_IDX_ENTRY_SIZE
;
}
inline
static
int
TqEmptyTail
()
{
//16
return
TQ_PAGE_SIZE
-
TqMaxEntryOnePage
();
}
#ifdef __cplusplus
extern
"C"
{
...
...
@@ -28,47 +39,49 @@ extern "C" {
typedef
struct
TqMetaHandle
{
int64_t
key
;
int64_t
offset
;
void
*
valueInUse
;
void
*
valueInTxn
;
int64_t
serializedSize
;
void
*
valueInUse
;
void
*
valueInTxn
;
}
TqMetaHandle
;
typedef
struct
TqMetaList
{
TqMetaHandle
handle
;
struct
TqMetaList
*
next
;
struct
TqMetaList
*
inTxnPrev
;
struct
TqMetaList
*
inTxnNext
;
//
struct TqMetaList* inTxnPrev;
//
struct TqMetaList* inTxnNext;
struct
TqMetaList
*
unpersistPrev
;
struct
TqMetaList
*
unpersistNext
;
}
TqMetaList
;
typedef
struct
TqMetaStore
{
TqMetaList
*
inUse
[
TQ_INUSE
_SIZE
];
//a table head
, key is empty
TqMetaList
*
bucket
[
TQ_BUCKET
_SIZE
];
//a table head
TqMetaList
*
unpersistHead
;
int
fileFd
;
//TODO:temporaral use
int
idxFd
;
//TODO:temporaral use
void
*
(
*
serializer
)(
void
*
);
void
*
(
*
deserializer
)(
void
*
);
int
fileFd
;
//TODO:temporaral use
, to be replaced by unified tfile
int
idxFd
;
//TODO:temporaral use
, to be replaced by unified tfile
int
(
*
serializer
)(
TqGroupHandle
*
,
void
*
*
);
const
void
*
(
*
deserializer
)(
const
void
*
,
TqGroupHandle
*
);
void
(
*
deleter
)(
void
*
);
}
TqMetaStore
;
TqMetaStore
*
tqStoreOpen
(
const
char
*
path
,
void
*
serializer
(
void
*
),
void
*
deserializer
(
void
*
),
void
deleter
(
void
*
));
TqMetaStore
*
tqStoreOpen
(
const
char
*
path
,
int
serializer
(
TqGroupHandle
*
,
void
**
),
const
void
*
deserializer
(
const
void
*
,
TqGroupHandle
*
),
void
deleter
(
void
*
));
int32_t
tqStoreClose
(
TqMetaStore
*
);
int32_t
tqStoreDelete
(
TqMetaStore
*
);
//
int32_t tqStoreDelete(TqMetaStore*);
//int32_t TqStoreCommitAll(TqMetaStore*);
int32_t
tqStorePersist
(
TqMetaStore
*
);
TqMetaHandle
*
tqHandleGetInUse
(
TqMetaStore
*
,
int64_t
key
);
int32_t
tqHandlePutInUse
(
TqMetaStore
*
,
TqMetaHandle
*
handle
);
TqMetaHandle
*
tqHandleGetInTxn
(
TqMetaStore
*
,
int64_t
key
);
int32_t
tqHandlePutInTxn
(
TqMetaStore
*
,
TqMetaHandle
*
handle
);
//delete in-use-handle, make in-txn-handle in use
TqMetaHandle
*
tqHandleGet
(
TqMetaStore
*
,
int64_t
key
);
int32_t
tqHandlePut
(
TqMetaStore
*
,
int64_t
key
,
void
*
value
);
//do commit
int32_t
tqHandleCommit
(
TqMetaStore
*
,
int64_t
key
);
//delete
in-txn-handle
//delete
uncommitted
int32_t
tqHandleAbort
(
TqMetaStore
*
,
int64_t
key
);
//delete
in-use-handle
//delete
committed
int32_t
tqHandleDel
(
TqMetaStore
*
,
int64_t
key
);
//delete
in-use-handle and in-txn-handle
//delete
both committed and uncommitted
int32_t
tqHandleClear
(
TqMetaStore
*
,
int64_t
key
);
#ifdef __cplusplus
...
...
source/dnode/vnode/tq/src/tq.c
浏览文件 @
98899a12
...
...
@@ -26,12 +26,12 @@ static int tqProtoCheck(TmqMsgHead *pMsg) {
return
pMsg
->
protoVer
==
0
;
}
static
int
tqAckOneTopic
(
TqBufferHandle
*
b
h
andle
,
TmqOneAck
*
pAck
,
TqQueryMsg
**
ppQuery
)
{
static
int
tqAckOneTopic
(
TqBufferHandle
*
b
H
andle
,
TmqOneAck
*
pAck
,
TqQueryMsg
**
ppQuery
)
{
//clean old item and move forward
int32_t
consumeOffset
=
pAck
->
consumeOffset
;
int
idx
=
consumeOffset
%
TQ_BUFFER_SIZE
;
ASSERT
(
b
handle
->
buffer
[
idx
].
content
&&
bh
andle
->
buffer
[
idx
].
executor
);
tfree
(
b
h
andle
->
buffer
[
idx
].
content
);
ASSERT
(
b
Handle
->
buffer
[
idx
].
content
&&
bH
andle
->
buffer
[
idx
].
executor
);
tfree
(
b
H
andle
->
buffer
[
idx
].
content
);
if
(
1
/* TODO: need to launch new query */
)
{
TqQueryMsg
*
pNewQuery
=
malloc
(
sizeof
(
TqQueryMsg
));
if
(
pNewQuery
==
NULL
)
{
...
...
@@ -39,22 +39,22 @@ static int tqAckOneTopic(TqBufferHandle *bhandle, TmqOneAck *pAck, TqQueryMsg**
return
-
1
;
}
//TODO: lock executor
pNewQuery
->
exec
->
executor
=
b
h
andle
->
buffer
[
idx
].
executor
;
pNewQuery
->
exec
->
executor
=
b
H
andle
->
buffer
[
idx
].
executor
;
//TODO: read from wal and assign to src
pNewQuery
->
exec
->
src
=
0
;
pNewQuery
->
exec
->
dest
=
&
b
h
andle
->
buffer
[
idx
];
pNewQuery
->
exec
->
dest
=
&
b
H
andle
->
buffer
[
idx
];
pNewQuery
->
next
=
*
ppQuery
;
*
ppQuery
=
pNewQuery
;
}
return
0
;
}
static
int
tqAck
(
TqGroupHandle
*
g
h
andle
,
TmqAcks
*
pAcks
)
{
static
int
tqAck
(
TqGroupHandle
*
g
H
andle
,
TmqAcks
*
pAcks
)
{
int32_t
ackNum
=
pAcks
->
ackNum
;
TmqOneAck
*
acks
=
pAcks
->
acks
;
//double ptr for acks and list
int
i
=
0
;
TqListHandle
*
node
=
g
h
andle
->
head
;
TqListHandle
*
node
=
g
H
andle
->
head
;
int
ackCnt
=
0
;
TqQueryMsg
*
pQuery
=
NULL
;
while
(
i
<
ackNum
&&
node
->
next
)
{
...
...
@@ -99,8 +99,8 @@ int tqDropTCGroup(STQ* pTq, int64_t topicId, int64_t cgId, int64_t cId) {
return
0
;
}
static
int
tqFetch
(
TqGroupHandle
*
g
h
andle
,
void
**
msg
)
{
TqListHandle
*
head
=
g
h
andle
->
head
;
static
int
tqFetch
(
TqGroupHandle
*
g
H
andle
,
void
**
msg
)
{
TqListHandle
*
head
=
g
H
andle
->
head
;
TqListHandle
*
node
=
head
;
int
totSize
=
0
;
//TODO: make it a macro
...
...
@@ -148,7 +148,7 @@ TqGroupHandle* tqGetGroupHandle(STQ* pTq, int64_t cId) {
return
NULL
;
}
int
tqLaunchQuery
(
TqGroupHandle
*
g
h
andle
)
{
int
tqLaunchQuery
(
TqGroupHandle
*
g
H
andle
)
{
return
0
;
}
...
...
@@ -156,7 +156,7 @@ int tqSendLaunchQuery(TqGroupHandle* gHandle) {
return
0
;
}
/*int tqMoveOffsetToNext(TqGroupHandle* g
h
andle) {*/
/*int tqMoveOffsetToNext(TqGroupHandle* g
H
andle) {*/
/*return 0;*/
/*}*/
...
...
@@ -177,13 +177,13 @@ int tqConsume(STQ* pTq, TmqConsumeReq* pMsg) {
return
-
1
;
}
int64_t
clientId
=
pMsg
->
head
.
clientId
;
TqGroupHandle
*
g
h
andle
=
tqGetGroupHandle
(
pTq
,
clientId
);
if
(
g
h
andle
==
NULL
)
{
TqGroupHandle
*
g
H
andle
=
tqGetGroupHandle
(
pTq
,
clientId
);
if
(
g
H
andle
==
NULL
)
{
//client not connect
return
-
1
;
}
if
(
pMsg
->
acks
.
ackNum
!=
0
)
{
if
(
tqAck
(
g
h
andle
,
&
pMsg
->
acks
)
!=
0
)
{
if
(
tqAck
(
g
H
andle
,
&
pMsg
->
acks
)
!=
0
)
{
//ack not success
return
-
1
;
}
...
...
@@ -191,13 +191,13 @@ int tqConsume(STQ* pTq, TmqConsumeReq* pMsg) {
TmqConsumeRsp
*
pRsp
=
(
TmqConsumeRsp
*
)
pMsg
;
if
(
tqFetch
(
g
h
andle
,
(
void
**
)
&
pRsp
->
msgs
)
<=
0
)
{
if
(
tqFetch
(
g
H
andle
,
(
void
**
)
&
pRsp
->
msgs
)
<=
0
)
{
//fetch error
return
-
1
;
}
//judge and launch new query
if
(
tqLaunchQuery
(
g
h
andle
))
{
if
(
tqLaunchQuery
(
g
H
andle
))
{
//launch query error
return
-
1
;
}
...
...
@@ -206,7 +206,7 @@ int tqConsume(STQ* pTq, TmqConsumeReq* pMsg) {
int
tqSerializeGroupHandle
(
TqGroupHandle
*
gHandle
,
void
**
ppBytes
)
{
//calculate size
int
sz
=
tqGet
G
HandleSSize
(
gHandle
);
int
sz
=
tqGet
g
HandleSSize
(
gHandle
);
void
*
ptr
=
realloc
(
*
ppBytes
,
sz
);
if
(
ptr
==
NULL
)
{
free
(
ppBytes
);
...
...
@@ -313,7 +313,7 @@ const void* tqDeserializeBufItem(const void* pBytes, TqBufferItem *bufItem) {
}
//TODO: make this a macro
int
tqGet
G
HandleSSize
(
const
TqGroupHandle
*
gHandle
)
{
int
tqGet
g
HandleSSize
(
const
TqGroupHandle
*
gHandle
)
{
return
sizeof
(
int64_t
)
*
2
+
sizeof
(
int32_t
)
+
gHandle
->
topicNum
*
tqBufHandleSSize
();
...
...
source/dnode/vnode/tq/src/tqMetaStore.c
浏览文件 @
98899a12
...
...
@@ -13,34 +13,46 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "tqMetaStore.h"
//TODO:replace by a abstract file layer
//TODO:replace by a
n
abstract file layer
#include <fcntl.h>
#include <string.h>
#include <unistd.h>
#define TQ_META_NAME "tq.meta"
#define TQ_IDX_NAME "tq.idx"
static
int32_t
tqHandlePutCommitted
(
TqMetaStore
*
,
int64_t
key
,
void
*
value
);
static
TqMetaHandle
*
tqHandleGetUncommitted
(
TqMetaStore
*
,
int64_t
key
);
typedef
struct
TqMetaPageBuf
{
int16_t
offset
;
char
buffer
[
TQ_PAGE_SIZE
];
}
TqMetaPageBuf
;
TqMetaStore
*
tqStoreOpen
(
const
char
*
path
,
void
*
serializer
(
void
*
),
void
*
deserializer
(
void
*
),
void
deleter
(
void
*
))
{
//concat data file name and index file name
int
fileFd
=
open
(
path
,
O_WRONLY
|
O_CREAT
|
O_EXCL
,
0755
);
if
(
fileFd
<
0
)
return
NULL
;
TqMetaStore
*
tqStoreOpen
(
const
char
*
path
,
int
serializer
(
TqGroupHandle
*
,
void
**
),
const
void
*
deserializer
(
const
void
*
,
TqGroupHandle
*
),
void
deleter
(
void
*
))
{
TqMetaStore
*
pMeta
=
malloc
(
sizeof
(
TqMetaStore
));
if
(
pMeta
==
NULL
)
{
//close
return
NULL
;
}
memset
(
pMeta
,
0
,
sizeof
(
TqMetaStore
));
pMeta
->
fileFd
=
fileFd
;
int
idxFd
=
open
(
path
,
O_WRONLY
|
O_CREAT
|
O_EXCL
,
0755
);
//concat data file name and index file name
size_t
pathLen
=
strlen
(
path
);
char
name
[
pathLen
+
10
];
strcpy
(
name
,
path
);
strcat
(
name
,
"/"
TQ_IDX_NAME
);
int
idxFd
=
open
(
name
,
O_WRONLY
|
O_CREAT
|
O_EXCL
,
0755
);
if
(
idxFd
<
0
)
{
//close file
//free memory
return
NULL
;
}
pMeta
->
idxFd
=
idxFd
;
pMeta
->
unpersistHead
=
malloc
(
sizeof
(
TqMetaList
));
if
(
pMeta
->
unpersistHead
==
NULL
)
{
...
...
@@ -48,16 +60,65 @@ TqMetaStore* tqStoreOpen(const char* path, void* serializer(void*),
//free memory
return
NULL
;
}
strcpy
(
name
,
path
);
strcat
(
name
,
"/"
TQ_META_NAME
);
int
fileFd
=
open
(
name
,
O_WRONLY
|
O_CREAT
|
O_EXCL
,
0755
);
if
(
fileFd
<
0
)
return
NULL
;
memset
(
pMeta
,
0
,
sizeof
(
TqMetaStore
));
pMeta
->
fileFd
=
fileFd
;
pMeta
->
serializer
=
serializer
;
pMeta
->
deserializer
=
deserializer
;
pMeta
->
deleter
=
deleter
;
//read idx file and load into memory
char
readBuf
[
TQ_PAGE_SIZE
];
int
readSize
;
while
((
readSize
=
read
(
idxFd
,
readBuf
,
TQ_PAGE_SIZE
))
!=
-
1
)
{
//loop read every entry
for
(
int
i
=
0
;
i
<
readSize
;
i
+=
TQ_IDX_ENTRY_SIZE
)
{
TqMetaList
*
pNode
=
malloc
(
sizeof
(
TqMetaHandle
));
memset
(
pNode
,
0
,
sizeof
(
TqMetaList
));
if
(
pNode
==
NULL
)
{
//TODO: free memory and return error
}
memcpy
(
&
pNode
->
handle
,
&
readBuf
[
i
],
TQ_IDX_ENTRY_SIZE
);
int
bucketKey
=
pNode
->
handle
.
key
&
TQ_BUCKET_SIZE
;
pNode
->
next
=
pMeta
->
bucket
[
bucketKey
];
pMeta
->
bucket
[
bucketKey
]
=
pNode
;
}
}
return
pMeta
;
}
int32_t
tqStoreClose
(
TqMetaStore
*
pMeta
)
{
//commit data and idx
//close file
tqStorePersist
(
pMeta
);
ASSERT
(
pMeta
->
unpersistHead
&&
pMeta
->
unpersistHead
->
next
==
NULL
);
close
(
pMeta
->
fileFd
);
close
(
pMeta
->
idxFd
);
//free memory
for
(
int
i
=
0
;
i
<
TQ_BUCKET_SIZE
;
i
++
)
{
TqMetaList
*
node
=
pMeta
->
bucket
[
i
];
pMeta
->
bucket
[
i
]
=
NULL
;
while
(
node
)
{
ASSERT
(
node
->
unpersistNext
==
NULL
);
ASSERT
(
node
->
unpersistPrev
==
NULL
);
if
(
node
->
handle
.
valueInTxn
)
{
pMeta
->
deleter
(
node
->
handle
.
valueInTxn
);
}
if
(
node
->
handle
.
valueInUse
)
{
pMeta
->
deleter
(
node
->
handle
.
valueInUse
);
}
TqMetaList
*
next
=
node
->
next
;
free
(
node
);
node
=
next
;
}
}
free
(
pMeta
);
return
0
;
}
...
...
@@ -68,45 +129,262 @@ int32_t tqStoreDelete(TqMetaStore* pMeta) {
return
0
;
}
//TODO: wrap in tfile
int32_t
tqStorePersist
(
TqMetaStore
*
pMeta
)
{
TqMetaList
*
node
=
pMeta
->
unpersistHead
;
while
(
node
->
unpersistNext
)
{
char
writeBuf
[
TQ_PAGE_SIZE
];
int64_t
*
bufPtr
=
(
int64_t
*
)
writeBuf
;
TqMetaList
*
pHead
=
pMeta
->
unpersistHead
;
TqMetaList
*
pNode
=
pHead
->
unpersistNext
;
while
(
pHead
!=
pNode
)
{
if
(
pNode
->
handle
.
valueInUse
==
NULL
)
{
//put delete token in data file
uint32_t
delete
=
TQ_ACTION_DELETE
;
int
nBytes
=
write
(
pMeta
->
fileFd
,
&
delete
,
sizeof
(
uint32_t
));
ASSERT
(
nBytes
==
sizeof
(
uint32_t
));
//remove from list
int
bucketKey
=
pNode
->
handle
.
key
&
TQ_BUCKET_SIZE
;
TqMetaList
*
pBucketHead
=
pMeta
->
bucket
[
bucketKey
];
if
(
pBucketHead
==
pNode
)
{
pMeta
->
bucket
[
bucketKey
]
=
pBucketHead
->
next
;
}
else
{
TqMetaList
*
pBucketNode
=
pBucketHead
;
while
(
pBucketNode
->
next
!=
NULL
&&
pBucketNode
->
next
!=
pNode
)
{
pBucketNode
=
pBucketNode
->
next
;
}
if
(
pBucketNode
->
next
!=
NULL
)
{
ASSERT
(
pBucketNode
->
next
==
pNode
);
pBucketNode
->
next
=
pNode
->
next
;
if
(
pNode
->
handle
.
valueInUse
)
{
pMeta
->
deleter
(
pNode
->
handle
.
valueInUse
);
}
free
(
pNode
);
}
}
}
//serialize
void
*
pBytes
=
NULL
;
int
sz
=
pMeta
->
serializer
(
pNode
->
handle
.
valueInUse
,
&
pBytes
);
ASSERT
(
pBytes
!=
NULL
);
//get current offset
//append data
//write offset and idx
int64_t
offset
=
lseek
(
pMeta
->
fileFd
,
0
,
SEEK_CUR
);
int
nBytes
=
write
(
pMeta
->
fileFd
,
pBytes
,
sz
);
//TODO: handle error in tfile
ASSERT
(
nBytes
==
sz
);
pNode
->
handle
.
offset
=
offset
;
pNode
->
handle
.
serializedSize
=
sz
;
//write idx
//TODO: endian check and convert
*
(
bufPtr
++
)
=
pNode
->
handle
.
key
;
*
(
bufPtr
++
)
=
pNode
->
handle
.
offset
;
*
(
bufPtr
++
)
=
(
int64_t
)
sz
;
if
((
char
*
)(
bufPtr
+
3
)
>
writeBuf
+
TQ_PAGE_SIZE
)
{
nBytes
=
write
(
pMeta
->
idxFd
,
writeBuf
,
sizeof
(
writeBuf
));
//TODO: handle error in tfile
ASSERT
(
nBytes
==
sizeof
(
writeBuf
));
memset
(
writeBuf
,
0
,
TQ_PAGE_SIZE
);
bufPtr
=
(
int64_t
*
)
writeBuf
;
}
//remove from unpersist list
pHead
->
unpersistNext
=
pNode
->
unpersistNext
;
pHead
->
unpersistNext
->
unpersistPrev
=
pHead
;
pNode
->
unpersistPrev
=
pNode
->
unpersistNext
=
NULL
;
pNode
=
pHead
->
unpersistNext
;
}
//write left bytes
if
((
char
*
)
bufPtr
!=
writeBuf
)
{
int
used
=
(
char
*
)
bufPtr
-
writeBuf
;
int
nBytes
=
write
(
pMeta
->
idxFd
,
writeBuf
,
used
);
//TODO: handle error in tfile
ASSERT
(
nBytes
==
used
);
}
//TODO: using fsync in tfile
fsync
(
pMeta
->
idxFd
);
fsync
(
pMeta
->
fileFd
);
return
0
;
}
int32_t
tqHandlePutInUse
(
TqMetaStore
*
pMeta
,
TqMetaHandle
*
handle
)
{
static
int32_t
tqHandlePutCommitted
(
TqMetaStore
*
pMeta
,
int64_t
key
,
void
*
value
)
{
int64_t
bucketKey
=
key
&
TQ_BUCKET_SIZE
;
TqMetaList
*
pNode
=
pMeta
->
bucket
[
bucketKey
];
while
(
pNode
)
{
if
(
pNode
->
handle
.
key
==
key
)
{
//TODO: think about thread safety
pMeta
->
deleter
(
pNode
->
handle
.
valueInUse
);
//change pointer ownership
pNode
->
handle
.
valueInUse
=
value
;
return
0
;
}
else
{
pNode
=
pNode
->
next
;
}
}
TqMetaList
*
pNewNode
=
malloc
(
sizeof
(
TqMetaList
));
if
(
pNewNode
==
NULL
)
{
//TODO: memory error
return
-
1
;
}
memset
(
pNewNode
,
0
,
sizeof
(
TqMetaList
));
pNewNode
->
handle
.
key
=
key
;
pNewNode
->
handle
.
valueInUse
=
value
;
//put into unpersist list
pNewNode
->
unpersistPrev
=
pMeta
->
unpersistHead
;
pNewNode
->
unpersistNext
=
pMeta
->
unpersistHead
->
unpersistNext
;
pMeta
->
unpersistHead
->
unpersistNext
->
unpersistPrev
=
pNewNode
;
pMeta
->
unpersistHead
->
unpersistNext
=
pNewNode
;
return
0
;
}
TqMetaHandle
*
tqHandleGetInUse
(
TqMetaStore
*
pMeta
,
int64_t
key
)
{
TqMetaHandle
*
tqHandleGet
(
TqMetaStore
*
pMeta
,
int64_t
key
)
{
int64_t
bucketKey
=
key
&
TQ_BUCKET_SIZE
;
TqMetaList
*
pNode
=
pMeta
->
bucket
[
bucketKey
];
while
(
pNode
)
{
if
(
pNode
->
handle
.
key
==
key
)
{
if
(
pNode
->
handle
.
valueInUse
!=
NULL
)
{
return
&
pNode
->
handle
;
}
else
{
return
NULL
;
}
}
else
{
pNode
=
pNode
->
next
;
}
}
return
NULL
;
}
int32_t
tqHandlePutInTxn
(
TqMetaStore
*
pMeta
,
TqMetaHandle
*
handle
)
{
int32_t
tqHandlePut
(
TqMetaStore
*
pMeta
,
int64_t
key
,
void
*
value
)
{
int64_t
bucketKey
=
key
&
TQ_BUCKET_SIZE
;
TqMetaList
*
pNode
=
pMeta
->
bucket
[
bucketKey
];
while
(
pNode
)
{
if
(
pNode
->
handle
.
key
==
key
)
{
//TODO: think about thread safety
pMeta
->
deleter
(
pNode
->
handle
.
valueInTxn
);
//change pointer ownership
pNode
->
handle
.
valueInTxn
=
value
;
return
0
;
}
else
{
pNode
=
pNode
->
next
;
}
}
TqMetaList
*
pNewNode
=
malloc
(
sizeof
(
TqMetaList
));
if
(
pNewNode
==
NULL
)
{
//TODO: memory error
return
-
1
;
}
memset
(
pNewNode
,
0
,
sizeof
(
TqMetaList
));
pNewNode
->
handle
.
key
=
key
;
pNewNode
->
handle
.
valueInTxn
=
value
;
return
0
;
}
TqMetaHandle
*
tqHandleGetInTxn
(
TqMetaStore
*
pMeta
,
int64_t
key
)
{
static
TqMetaHandle
*
tqHandleGetUncommitted
(
TqMetaStore
*
pMeta
,
int64_t
key
)
{
int64_t
bucketKey
=
key
&
TQ_BUCKET_SIZE
;
TqMetaList
*
pNode
=
pMeta
->
bucket
[
bucketKey
];
while
(
pNode
)
{
if
(
pNode
->
handle
.
key
==
key
)
{
if
(
pNode
->
handle
.
valueInTxn
!=
NULL
)
{
return
&
pNode
->
handle
;
}
else
{
return
NULL
;
}
}
else
{
pNode
=
pNode
->
next
;
}
}
return
NULL
;
}
int32_t
tqHandleCommit
(
TqMetaStore
*
pMeta
,
int64_t
key
)
{
int64_t
bucketKey
=
key
&
TQ_BUCKET_SIZE
;
TqMetaList
*
pNode
=
pMeta
->
bucket
[
bucketKey
];
while
(
pNode
)
{
if
(
pNode
->
handle
.
key
==
key
)
{
if
(
pNode
->
handle
.
valueInUse
!=
NULL
)
{
pMeta
->
deleter
(
pNode
->
handle
.
valueInUse
);
}
pNode
->
handle
.
valueInUse
=
pNode
->
handle
.
valueInTxn
;
if
(
pNode
->
unpersistNext
==
NULL
)
{
pNode
->
unpersistNext
=
pMeta
->
unpersistHead
->
unpersistNext
;
pNode
->
unpersistPrev
=
pMeta
->
unpersistHead
;
pMeta
->
unpersistHead
->
unpersistNext
->
unpersistPrev
=
pNode
;
pMeta
->
unpersistHead
->
unpersistNext
=
pNode
;
}
return
0
;
}
else
{
pNode
=
pNode
->
next
;
}
}
return
-
1
;
}
int32_t
tqHandleAbort
(
TqMetaStore
*
pMeta
,
int64_t
key
)
{
int64_t
bucketKey
=
key
&
TQ_BUCKET_SIZE
;
TqMetaList
*
pNode
=
pMeta
->
bucket
[
bucketKey
];
while
(
pNode
)
{
if
(
pNode
->
handle
.
key
==
key
)
{
if
(
pNode
->
handle
.
valueInTxn
!=
NULL
)
{
pMeta
->
deleter
(
pNode
->
handle
.
valueInTxn
);
pNode
->
handle
.
valueInTxn
=
NULL
;
return
0
;
}
return
-
1
;
}
else
{
pNode
=
pNode
->
next
;
}
}
return
-
2
;
}
int32_t
tqHandleDel
(
TqMetaStore
*
pMeta
,
int64_t
key
)
{
int64_t
bucketKey
=
key
&
TQ_BUCKET_SIZE
;
TqMetaList
*
pNode
=
pMeta
->
bucket
[
bucketKey
];
while
(
pNode
)
{
if
(
pNode
->
handle
.
key
==
key
)
{
pMeta
->
deleter
(
pNode
->
handle
.
valueInTxn
);
pNode
->
handle
.
valueInTxn
=
NULL
;
return
0
;
}
else
{
pNode
=
pNode
->
next
;
}
}
//no such key
return
-
1
;
}
int32_t
tqHandleClear
(
TqMetaStore
*
pMeta
,
int64_t
key
)
{
int64_t
bucketKey
=
key
&
TQ_BUCKET_SIZE
;
TqMetaList
*
pNode
=
pMeta
->
bucket
[
bucketKey
];
bool
exist
=
false
;
while
(
pNode
)
{
if
(
pNode
->
handle
.
key
==
key
)
{
if
(
pNode
->
handle
.
valueInUse
!=
NULL
)
{
exist
=
true
;
pMeta
->
deleter
(
pNode
->
handle
.
valueInUse
);
pNode
->
handle
.
valueInUse
=
NULL
;
}
if
(
pNode
->
handle
.
valueInTxn
!=
NULL
)
{
exist
=
true
;
pMeta
->
deleter
(
pNode
->
handle
.
valueInTxn
);
pNode
->
handle
.
valueInTxn
=
NULL
;
}
if
(
exist
)
{
if
(
pNode
->
unpersistNext
==
NULL
)
{
pNode
->
unpersistNext
=
pMeta
->
unpersistHead
->
unpersistNext
;
pNode
->
unpersistPrev
=
pMeta
->
unpersistHead
;
pMeta
->
unpersistHead
->
unpersistNext
->
unpersistPrev
=
pNode
;
pMeta
->
unpersistHead
->
unpersistNext
=
pNode
;
}
return
0
;
}
return
-
1
;
}
else
{
pNode
=
pNode
->
next
;
}
}
return
-
2
;
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录