Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
loser646
TDengine
提交
78462462
T
TDengine
项目概览
loser646
/
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,发现更多精彩内容 >>
未验证
提交
78462462
编写于
3月 23, 2020
作者:
H
haojun Liao
提交者:
GitHub
3月 23, 2020
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #1404 from taosdata/feature/2.0tsdb
Feature/2.0tsdb
上级
19025969
ed7d5fe8
变更
7
隐藏空白更改
内联
并排
Showing
7 changed file
with
120 addition
and
36 deletion
+120
-36
src/common/src/dataformat.c
src/common/src/dataformat.c
+1
-1
src/vnode/tsdb/inc/tsdb.h
src/vnode/tsdb/inc/tsdb.h
+2
-0
src/vnode/tsdb/inc/tsdbCache.h
src/vnode/tsdb/inc/tsdbCache.h
+4
-1
src/vnode/tsdb/src/tsdbCache.c
src/vnode/tsdb/src/tsdbCache.c
+24
-17
src/vnode/tsdb/src/tsdbMain.c
src/vnode/tsdb/src/tsdbMain.c
+62
-10
src/vnode/tsdb/src/tsdbMeta.c
src/vnode/tsdb/src/tsdbMeta.c
+4
-0
src/vnode/tsdb/tests/tsdbTests.cpp
src/vnode/tsdb/tests/tsdbTests.cpp
+23
-7
未找到文件。
src/common/src/dataformat.c
浏览文件 @
78462462
...
...
@@ -150,7 +150,7 @@ void tdFreeSchema(STSchema *pSchema) {
*/
void
tdUpdateSchema
(
STSchema
*
pSchema
)
{
STColumn
*
pCol
=
NULL
;
int32_t
offset
=
0
;
int32_t
offset
=
TD_DATA_ROW_HEAD_SIZE
;
for
(
int
i
=
0
;
i
<
schemaNCols
(
pSchema
);
i
++
)
{
pCol
=
schemaColAt
(
pSchema
,
i
);
colSetOffset
(
pCol
,
offset
);
...
...
src/vnode/tsdb/inc/tsdb.h
浏览文件 @
78462462
...
...
@@ -59,6 +59,8 @@ tsdb_repo_t * tsdbOpenRepo(char *tsdbDir);
int32_t
tsdbCloseRepo
(
tsdb_repo_t
*
repo
);
int32_t
tsdbConfigRepo
(
tsdb_repo_t
*
repo
,
STsdbCfg
*
pCfg
);
int32_t
tsdbTriggerCommit
(
tsdb_repo_t
*
repo
);
int32_t
tsdbLockRepo
(
tsdb_repo_t
*
repo
);
int32_t
tsdbUnLockRepo
(
tsdb_repo_t
*
repo
);
// --------- TSDB TABLE DEFINITION
typedef
struct
{
...
...
src/vnode/tsdb/inc/tsdbCache.h
浏览文件 @
78462462
...
...
@@ -19,6 +19,7 @@
#include "taosdef.h"
#include "tlist.h"
#include "tsdb.h"
#ifdef __cplusplus
extern
"C"
{
...
...
@@ -49,13 +50,15 @@ typedef struct {
typedef
struct
{
int
maxBytes
;
int
cacheBlockSize
;
int
totalCacheBlocks
;
STsdbCachePool
pool
;
STsdbCacheBlock
*
curBlock
;
SCacheMem
*
mem
;
SCacheMem
*
imem
;
tsdb_repo_t
*
pRepo
;
}
STsdbCache
;
STsdbCache
*
tsdbInitCache
(
int
maxBytes
,
int
cacheBlockSize
);
STsdbCache
*
tsdbInitCache
(
int
maxBytes
,
int
cacheBlockSize
,
tsdb_repo_t
*
pRepo
);
void
tsdbFreeCache
(
STsdbCache
*
pCache
);
void
*
tsdbAllocFromCache
(
STsdbCache
*
pCache
,
int
bytes
,
TSKEY
key
);
...
...
src/vnode/tsdb/src/tsdbCache.c
浏览文件 @
78462462
...
...
@@ -14,12 +14,13 @@
*/
#include <stdlib.h>
#include "tsdb.h"
#include "tsdbCache.h"
static
int
tsdbAllocBlockFromPool
(
STsdbCache
*
pCache
);
static
void
tsdbFreeBlockList
(
SCacheMem
*
mem
);
STsdbCache
*
tsdbInitCache
(
int
maxBytes
,
int
cacheBlockSize
)
{
STsdbCache
*
tsdbInitCache
(
int
maxBytes
,
int
cacheBlockSize
,
tsdb_repo_t
*
pRepo
)
{
STsdbCache
*
pCache
=
(
STsdbCache
*
)
calloc
(
1
,
sizeof
(
STsdbCache
));
if
(
pCache
==
NULL
)
return
NULL
;
...
...
@@ -27,9 +28,11 @@ STsdbCache *tsdbInitCache(int maxBytes, int cacheBlockSize) {
pCache
->
maxBytes
=
maxBytes
;
pCache
->
cacheBlockSize
=
cacheBlockSize
;
pCache
->
pRepo
=
pRepo
;
int
nBlocks
=
maxBytes
/
cacheBlockSize
+
1
;
if
(
nBlocks
<=
1
)
nBlocks
=
2
;
pCache
->
totalCacheBlocks
=
nBlocks
;
STsdbCachePool
*
pPool
=
&
(
pCache
->
pool
);
pPool
->
index
=
0
;
...
...
@@ -67,22 +70,10 @@ void *tsdbAllocFromCache(STsdbCache *pCache, int bytes, TSKEY key) {
if
(
pCache
==
NULL
)
return
NULL
;
if
(
bytes
>
pCache
->
cacheBlockSize
)
return
NULL
;
if
(
pCache
->
mem
==
NULL
)
{
// Create a new one
pCache
->
mem
=
(
SCacheMem
*
)
malloc
(
sizeof
(
SCacheMem
));
if
(
pCache
->
mem
==
NULL
)
return
NULL
;
pCache
->
mem
->
keyFirst
=
INT64_MAX
;
pCache
->
mem
->
keyLast
=
0
;
pCache
->
mem
->
numOfPoints
=
0
;
pCache
->
mem
->
list
=
tdListNew
(
sizeof
(
STsdbCacheBlock
*
));
}
if
(
isListEmpty
(
pCache
->
mem
->
list
))
{
if
(
tsdbAllocBlockFromPool
(
pCache
)
<
0
)
{
// TODO: deal with the error
if
(
pCache
->
curBlock
==
NULL
||
pCache
->
curBlock
->
remain
<
bytes
)
{
if
(
pCache
->
curBlock
!=
NULL
&&
(
pCache
->
mem
->
list
)
>=
pCache
->
totalCacheBlocks
/
2
)
{
tsdbTriggerCommit
(
pCache
->
pRepo
);
}
}
if
(
pCache
->
curBlock
->
remain
<
bytes
)
{
if
(
tsdbAllocBlockFromPool
(
pCache
)
<
0
)
{
// TODO: deal with the error
}
...
...
@@ -115,7 +106,12 @@ static void tsdbFreeBlockList(SCacheMem *mem) {
static
int
tsdbAllocBlockFromPool
(
STsdbCache
*
pCache
)
{
STsdbCachePool
*
pPool
=
&
(
pCache
->
pool
);
if
(
listNEles
(
pPool
->
memPool
)
==
0
)
return
-
1
;
tsdbLockRepo
(
pCache
->
pRepo
);
if
(
listNEles
(
pPool
->
memPool
)
==
0
)
{
tsdbUnLockRepo
(
pCache
->
pRepo
);
return
-
1
;
}
SListNode
*
node
=
tdListPopHead
(
pPool
->
memPool
);
...
...
@@ -125,8 +121,19 @@ static int tsdbAllocBlockFromPool(STsdbCache *pCache) {
pBlock
->
offset
=
0
;
pBlock
->
remain
=
pCache
->
cacheBlockSize
;
if
(
pCache
->
mem
==
NULL
)
{
// Create a new one
pCache
->
mem
=
(
SCacheMem
*
)
malloc
(
sizeof
(
SCacheMem
));
if
(
pCache
->
mem
==
NULL
)
return
NULL
;
pCache
->
mem
->
keyFirst
=
INT64_MAX
;
pCache
->
mem
->
keyLast
=
0
;
pCache
->
mem
->
numOfPoints
=
0
;
pCache
->
mem
->
list
=
tdListNew
(
sizeof
(
STsdbCacheBlock
*
));
}
tdListAppendNode
(
pCache
->
mem
->
list
,
node
);
pCache
->
curBlock
=
pBlock
;
tsdbUnLockRepo
(
pCache
->
pRepo
);
return
0
;
}
\ No newline at end of file
src/vnode/tsdb/src/tsdbMain.c
浏览文件 @
78462462
...
...
@@ -150,6 +150,7 @@ tsdb_repo_t *tsdbCreateRepo(char *rootDir, STsdbCfg *pCfg, void *limiter /* TODO
pRepo
->
rootDir
=
strdup
(
rootDir
);
pRepo
->
config
=
*
pCfg
;
pRepo
->
limiter
=
limiter
;
pthread_mutex_init
(
&
pRepo
->
mutex
,
NULL
);
// Create the environment files and directories
if
(
tsdbSetRepoEnv
(
pRepo
)
<
0
)
{
...
...
@@ -168,7 +169,7 @@ tsdb_repo_t *tsdbCreateRepo(char *rootDir, STsdbCfg *pCfg, void *limiter /* TODO
pRepo
->
tsdbMeta
=
pMeta
;
// Initialize cache
STsdbCache
*
pCache
=
tsdbInitCache
(
pCfg
->
maxCacheSize
,
-
1
);
STsdbCache
*
pCache
=
tsdbInitCache
(
pCfg
->
maxCacheSize
,
-
1
,
(
tsdb_repo_t
*
)
pRepo
);
if
(
pCache
==
NULL
)
{
free
(
pRepo
->
rootDir
);
tsdbFreeMeta
(
pRepo
->
tsdbMeta
);
...
...
@@ -249,7 +250,7 @@ tsdb_repo_t *tsdbOpenRepo(char *tsdbDir) {
return
NULL
;
}
pRepo
->
tsdbCache
=
tsdbInitCache
(
pRepo
->
config
.
maxCacheSize
,
-
1
);
pRepo
->
tsdbCache
=
tsdbInitCache
(
pRepo
->
config
.
maxCacheSize
,
-
1
,
(
tsdb_repo_t
*
)
pRepo
);
if
(
pRepo
->
tsdbCache
==
NULL
)
{
tsdbFreeMeta
(
pRepo
->
tsdbMeta
);
free
(
pRepo
->
rootDir
);
...
...
@@ -305,9 +306,12 @@ int32_t tsdbConfigRepo(tsdb_repo_t *repo, STsdbCfg *pCfg) {
int32_t
tsdbTriggerCommit
(
tsdb_repo_t
*
repo
)
{
STsdbRepo
*
pRepo
=
(
STsdbRepo
*
)
repo
;
if
(
pthread_mutex_lock
(
&
(
pRepo
->
mutex
))
<
0
)
return
-
1
;
if
(
pRepo
->
commit
)
return
0
;
tsdbLockRepo
(
repo
);
if
(
pRepo
->
commit
)
{
tsdbUnLockRepo
(
repo
);
return
-
1
;
}
pRepo
->
commit
=
1
;
// Loop to move pData to iData
for
(
int
i
=
0
;
i
<
pRepo
->
config
.
maxTables
;
i
++
)
{
...
...
@@ -320,15 +324,25 @@ int32_t tsdbTriggerCommit(tsdb_repo_t *repo) {
// TODO: Loop to move mem to imem
pRepo
->
tsdbCache
->
imem
=
pRepo
->
tsdbCache
->
mem
;
pRepo
->
tsdbCache
->
mem
=
NULL
;
pRepo
->
tsdbCache
->
curBlock
=
NULL
;
// TODO: here should set as detached or use join for memory leak
pthread_create
(
&
(
pRepo
->
commitThread
),
NULL
,
tsdbCommitToFile
,
(
void
*
)
repo
);
pthread_mutex_unlock
(
&
(
pRepo
->
mutex
));
pthread_join
(
pRepo
->
commitThread
,
NULL
);
tsdbUnLockRepo
(
repo
);
return
0
;
}
int32_t
tsdbLockRepo
(
tsdb_repo_t
*
repo
)
{
STsdbRepo
*
pRepo
=
(
STsdbRepo
*
)
repo
;
return
pthread_mutex_lock
(
repo
);
}
int32_t
tsdbUnLockRepo
(
tsdb_repo_t
*
repo
)
{
STsdbRepo
*
pRepo
=
(
STsdbRepo
*
)
repo
;
return
pthread_mutex_unlock
(
repo
);
}
/**
* Get the TSDB repository information, including some statistics
* @param pRepo the TSDB repository handle
...
...
@@ -691,6 +705,13 @@ static int32_t tdInsertRowToTable(STsdbRepo *pRepo, SDataRow row, STable *pTable
dataRowCpy
(
SL_GET_NODE_DATA
(
pNode
),
row
);
// Insert the skiplist node into the data
if
(
pTable
->
mem
==
NULL
)
{
pTable
->
mem
=
(
SMemTable
*
)
calloc
(
1
,
sizeof
(
SMemTable
));
if
(
pTable
->
mem
==
NULL
)
return
-
1
;
pTable
->
mem
->
pData
=
tSkipListCreate
(
5
,
TSDB_DATA_TYPE_TIMESTAMP
,
TYPE_BYTES
[
TSDB_DATA_TYPE_TIMESTAMP
],
0
,
0
,
getTupleKey
);
pTable
->
mem
->
keyFirst
=
INT64_MAX
;
pTable
->
mem
->
keyLast
=
0
;
}
tSkipListPut
(
pTable
->
mem
->
pData
,
pNode
);
if
(
key
>
pTable
->
mem
->
keyLast
)
pTable
->
mem
->
keyLast
=
key
;
if
(
key
<
pTable
->
mem
->
keyFirst
)
pTable
->
mem
->
keyFirst
=
key
;
...
...
@@ -718,7 +739,7 @@ static int32_t tsdbInsertDataToTable(tsdb_repo_t *repo, SSubmitBlk *pBlock) {
return
0
;
}
static
int
tsdbReadRowsFromCache
(
SSkipListIterator
*
pIter
,
TSKEY
maxKey
,
int
maxRowsToRead
,
void
*
dst
)
{
static
int
tsdbReadRowsFromCache
(
SSkipListIterator
*
pIter
,
TSKEY
maxKey
,
int
maxRowsToRead
,
SDataCol
**
cols
,
STSchema
*
pSchema
)
{
int
numOfRows
=
0
;
do
{
SSkipListNode
*
node
=
tSkipListIterGet
(
pIter
);
...
...
@@ -727,6 +748,11 @@ static int tsdbReadRowsFromCache(SSkipListIterator *pIter, TSKEY maxKey, int max
SDataRow
row
=
SL_GET_NODE_DATA
(
node
);
if
(
dataRowKey
(
row
)
>
maxKey
)
break
;
// Convert row data to column data
// for (int i = 0; i < schemaNCols(pSchema); i++) {
// STColumn *pCol = schemaColAt(pSchema, i);
// memcpy(cols[i]->data + TYPE_BYTES[colType(pCol)] * numOfRows, dataRowAt(row, pCol->offset),
// TYPE_BYTES[colType(pCol)]);
// }
numOfRows
++
;
if
(
numOfRows
>
maxRowsToRead
)
break
;
...
...
@@ -754,6 +780,8 @@ static void *tsdbCommitToFile(void *arg) {
int
maxCols
=
pMeta
->
maxCols
;
int
maxBytes
=
pMeta
->
maxRowBytes
;
SDataCol
**
cols
=
(
SDataCol
**
)
malloc
(
sizeof
(
SDataCol
*
)
*
maxCols
);
void
*
buf
=
malloc
((
maxBytes
+
sizeof
(
SDataCol
))
*
pCfg
->
maxRowsPerFileBlock
);
for
(
int
fid
=
sfid
;
fid
<=
efid
;
fid
++
)
{
TSKEY
minKey
=
0
,
maxKey
=
0
;
...
...
@@ -771,9 +799,17 @@ static void *tsdbCommitToFile(void *arg) {
}
}
// Init row data part
cols
[
0
]
=
(
SDataCol
*
)
buf
;
for
(
int
col
=
1
;
col
<
schemaNCols
(
pTable
->
schema
);
col
++
)
{
cols
[
col
]
=
(
SDataCol
*
)((
char
*
)(
cols
[
col
-
1
])
+
sizeof
(
SDataCol
)
+
colBytes
(
schemaColAt
(
pTable
->
schema
,
col
-
1
))
*
pCfg
->
maxRowsPerFileBlock
);
}
// Loop the iterator
int
rowsRead
=
0
;
while
((
rowsRead
=
tsdbReadRowsFromCache
(
iters
[
tid
],
maxKey
,
pCfg
->
maxRowsPerFileBlock
,
NULL
))
>
0
)
{
while
((
rowsRead
=
tsdbReadRowsFromCache
(
iters
[
tid
],
maxKey
,
pCfg
->
maxRowsPerFileBlock
,
cols
,
pTable
->
schema
))
>
0
)
{
// printf("rowsRead:%d-----------\n", rowsRead);
int
k
=
0
;
}
}
...
...
@@ -784,7 +820,23 @@ static void *tsdbCommitToFile(void *arg) {
if
(
iters
[
tid
]
!=
NULL
)
tSkipListDestroyIter
(
iters
[
tid
]);
}
free
(
buf
);
free
(
cols
);
free
(
iters
);
tsdbLockRepo
(
arg
);
tdListMove
(
pCache
->
imem
->
list
,
pCache
->
pool
.
memPool
);
free
(
pCache
->
imem
);
pCache
->
imem
=
NULL
;
pRepo
->
commit
=
0
;
// TODO: free the skiplist
for
(
int
i
=
0
;
i
<
pCfg
->
maxTables
;
i
++
)
{
STable
*
pTable
=
pMeta
->
tables
[
i
];
if
(
pTable
&&
pTable
->
imem
)
{
// Here has memory leak
pTable
->
imem
=
NULL
;
}
}
tsdbUnLockRepo
(
arg
);
return
NULL
;
}
\ No newline at end of file
src/vnode/tsdb/src/tsdbMeta.c
浏览文件 @
78462462
...
...
@@ -236,6 +236,10 @@ int32_t tsdbCreateTableImpl(STsdbMeta *pMeta, STableCfg *pCfg) {
table
->
type
=
TSDB_NORMAL_TABLE
;
table
->
superUid
=
-
1
;
table
->
schema
=
tdDupSchema
(
pCfg
->
schema
);
if
(
schemaNCols
(
table
->
schema
)
>
pMeta
->
maxCols
)
pMeta
->
maxCols
=
schemaNCols
(
table
->
schema
);
tdUpdateSchema
(
table
->
schema
);
int
bytes
=
tdMaxRowBytesFromSchema
(
table
->
schema
);
if
(
bytes
>
pMeta
->
maxRowBytes
)
pMeta
->
maxRowBytes
=
bytes
;
}
// Register to meta
...
...
src/vnode/tsdb/tests/tsdbTests.cpp
浏览文件 @
78462462
#include <gtest/gtest.h>
#include <stdlib.h>
#include <sys/time.h>
#include "tsdb.h"
#include "dataformat.h"
#include "tsdbFile.h"
#include "tsdbMeta.h"
TEST
(
TsdbTest
,
tableEncodeDecode
)
{
double
getCurTime
()
{
struct
timeval
tv
;
gettimeofday
(
&
tv
,
NULL
);
return
tv
.
tv_sec
+
tv
.
tv_usec
*
1E-6
;
}
TEST
(
TsdbTest
,
DISABLED_tableEncodeDecode
)
{
STable
*
pTable
=
(
STable
*
)
malloc
(
sizeof
(
STable
));
pTable
->
type
=
TSDB_NORMAL_TABLE
;
...
...
@@ -71,19 +78,22 @@ TEST(TsdbTest, createRepo) {
tsdbCreateTable
(
pRepo
,
&
tCfg
);
// // 3. Loop to write some simple data
int
nRows
=
1000
;
int
rowsPerSubmit
=
10
;
int
nRows
=
1000
0000
;
int
rowsPerSubmit
=
10
0
;
int64_t
start_time
=
1584081000000
;
SSubmitMsg
*
pMsg
=
(
SSubmitMsg
*
)
malloc
(
sizeof
(
SSubmitMsg
)
+
sizeof
(
SSubmitBlk
)
+
tdMaxRowBytesFromSchema
(
schema
)
*
rowsPerSubmit
);
double
stime
=
getCurTime
();
for
(
int
k
=
0
;
k
<
nRows
/
rowsPerSubmit
;
k
++
)
{
SSubmitBlk
*
pBlock
=
pMsg
->
blocks
;
pBlock
->
tableId
=
{.
uid
=
987607499877672L
,
.
tid
=
0
};
pBlock
->
sversion
=
0
;
pBlock
->
len
=
0
;
for
(
int
i
=
0
;
i
<
rowsPerSubmit
;
i
++
)
{
start_time
+=
1000
;
// start_time += 1000;
start_time
-=
1000
;
SDataRow
row
=
(
SDataRow
)(
pBlock
->
data
+
pBlock
->
len
);
tdInitDataRow
(
row
,
schema
);
...
...
@@ -102,16 +112,22 @@ TEST(TsdbTest, createRepo) {
tsdbInsertData
(
pRepo
,
pMsg
);
}
tsdbTriggerCommit
(
pRepo
);
double
etime
=
getCurTime
();
printf
(
"Spent %f seconds to write %d records
\n
"
,
etime
-
stime
,
nRows
);
// tsdbTriggerCommit(pRepo);
}
TEST
(
TsdbTest
,
openRepo
)
{
TEST
(
TsdbTest
,
DISABLED_
openRepo
)
{
tsdb_repo_t
*
pRepo
=
tsdbOpenRepo
(
"/home/ubuntu/work/ttest/vnode0"
);
ASSERT_NE
(
pRepo
,
nullptr
);
}
TEST
(
TsdbTest
,
createFileGroup
)
{
TEST
(
TsdbTest
,
DISABLED_
createFileGroup
)
{
SFileGroup
fGroup
;
ASSERT_EQ
(
tsdbCreateFileGroup
(
"/home/ubuntu/work/ttest/vnode0/data"
,
1820
,
&
fGroup
,
1000
),
0
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录