Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
loser646
TDengine
提交
f98efaf2
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,发现更多精彩内容 >>
提交
f98efaf2
编写于
3月 14, 2020
作者:
S
slguan
浏览文件
操作
浏览文件
下载
差异文件
Merge branch '2.0' into refact/slguan
上级
419a0610
e54e1bae
变更
19
展开全部
隐藏空白更改
内联
并排
Showing
19 changed file
with
797 addition
and
837 deletion
+797
-837
src/client/src/tcache.c
src/client/src/tcache.c
+302
-687
src/client/src/tscLocal.c
src/client/src/tscLocal.c
+1
-1
src/client/src/tscServer.c
src/client/src/tscServer.c
+25
-26
src/client/src/tscSub.c
src/client/src/tscSub.c
+1
-1
src/client/src/tscSystem.c
src/client/src/tscSystem.c
+1
-1
src/client/src/tscUtil.c
src/client/src/tscUtil.c
+10
-10
src/query/inc/tcache.h
src/query/inc/tcache.h
+92
-41
src/util/inc/hash.h
src/util/inc/hash.h
+57
-12
src/util/inc/tref.h
src/util/inc/tref.h
+58
-0
src/util/src/hash.c
src/util/src/hash.c
+141
-39
src/util/src/tskiplist.c
src/util/src/tskiplist.c
+1
-0
src/vnode/tsdb/inc/tsdb.h
src/vnode/tsdb/inc/tsdb.h
+1
-0
src/vnode/tsdb/inc/tsdbFile.h
src/vnode/tsdb/inc/tsdbFile.h
+23
-11
src/vnode/tsdb/inc/tsdbMeta.h
src/vnode/tsdb/inc/tsdbMeta.h
+1
-0
src/vnode/tsdb/src/tsdbFile.c
src/vnode/tsdb/src/tsdbFile.c
+67
-0
src/vnode/tsdb/src/tsdbMain.c
src/vnode/tsdb/src/tsdbMain.c
+0
-1
src/vnode/tsdb/src/tsdbMeta.c
src/vnode/tsdb/src/tsdbMeta.c
+9
-2
src/vnode/tsdb/src/tsdbMetaFile.c
src/vnode/tsdb/src/tsdbMetaFile.c
+4
-4
src/vnode/tsdb/tests/tsdbTests.cpp
src/vnode/tsdb/tests/tsdbTests.cpp
+3
-1
未找到文件。
src/client/src/tcache.c
浏览文件 @
f98efaf2
此差异已折叠。
点击以展开。
src/client/src/tscLocal.c
浏览文件 @
f98efaf2
...
...
@@ -484,7 +484,7 @@ int tscProcessLocalCmd(SSqlObj *pSql) {
pSql
->
res
.
qhandle
=
0x1
;
pSql
->
res
.
numOfRows
=
0
;
}
else
if
(
pCmd
->
command
==
TSDB_SQL_RESET_CACHE
)
{
taosC
learDataCache
(
tscCacheHandle
);
taosC
acheEmpty
(
tscCacheHandle
);
}
else
if
(
pCmd
->
command
==
TSDB_SQL_SERV_VERSION
)
{
tscProcessServerVer
(
pSql
);
}
else
if
(
pCmd
->
command
==
TSDB_SQL_CLI_VERSION
)
{
...
...
src/client/src/tscServer.c
浏览文件 @
f98efaf2
...
...
@@ -182,7 +182,6 @@ int tscSendMsgToServer(SSqlObj *pSql) {
}
pSql
->
ipList
->
ip
[
0
]
=
inet_addr
(
"192.168.0.1"
);
SSqlCmd
*
pCmd
=
&
pSql
->
cmd
;
if
(
pSql
->
cmd
.
command
<
TSDB_SQL_MGMT
)
{
pSql
->
ipList
->
port
=
tsDnodeShellPort
;
...
...
@@ -2641,7 +2640,7 @@ int tscProcessMeterMetaRsp(SSqlObj *pSql) {
SMeterMetaInfo
*
pMeterMetaInfo
=
tscGetMeterMetaInfo
(
&
pSql
->
cmd
,
0
,
0
);
assert
(
pMeterMetaInfo
->
pMeterMeta
==
NULL
);
pMeterMetaInfo
->
pMeterMeta
=
(
STableMeta
*
)
taos
AddDataIntoCache
(
tscCacheHandle
,
pMeterMetaInfo
->
name
,
(
char
*
)
pMeta
,
pMeterMetaInfo
->
pMeterMeta
=
(
STableMeta
*
)
taos
CachePut
(
tscCacheHandle
,
pMeterMetaInfo
->
name
,
(
char
*
)
pMeta
,
pMeta
->
contLen
,
tsMeterMetaKeepTimer
);
// todo handle out of memory case
if
(
pMeterMetaInfo
->
pMeterMeta
==
NULL
)
return
0
;
...
...
@@ -2750,7 +2749,7 @@ int tscProcessMultiMeterMetaRsp(SSqlObj *pSql) {
int32_t
size
=
(
int32_t
)(
rsp
-
((
char
*
)
pMeta
));
// Consistent with STableMeta in cache
pMeta
->
index
=
0
;
(
void
)
taos
AddDataIntoCache
(
tscCacheHandle
,
pMeta
->
tableId
,
(
char
*
)
pMeta
,
size
,
tsMeterMetaKeepTimer
);
(
void
)
taos
CachePut
(
tscCacheHandle
,
pMeta
->
tableId
,
(
char
*
)
pMeta
,
size
,
tsMeterMetaKeepTimer
);
}
pSql
->
res
.
code
=
TSDB_CODE_SUCCESS
;
...
...
@@ -2857,9 +2856,9 @@ int tscProcessMetricMetaRsp(SSqlObj *pSql) {
#endif
// release the used metricmeta
taos
RemoveDataFromCach
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMetricMeta
),
false
);
taos
CacheReleas
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMetricMeta
),
false
);
pMeterMetaInfo
->
pMetricMeta
=
(
SSuperTableMeta
*
)
taos
AddDataIntoCache
(
tscCacheHandle
,
name
,
(
char
*
)
metricMetaList
[
i
],
pMeterMetaInfo
->
pMetricMeta
=
(
SSuperTableMeta
*
)
taos
CachePut
(
tscCacheHandle
,
name
,
(
char
*
)
metricMetaList
[
i
],
sizes
[
i
],
tsMetricMetaKeepTimer
);
tfree
(
metricMetaList
[
i
]);
...
...
@@ -2917,11 +2916,11 @@ int tscProcessShowRsp(SSqlObj *pSql) {
key
[
0
]
=
pCmd
->
msgType
+
'a'
;
strcpy
(
key
+
1
,
"showlist"
);
taos
RemoveDataFromCach
e
(
tscCacheHandle
,
(
void
*
)
&
(
pMeterMetaInfo
->
pMeterMeta
),
false
);
taos
CacheReleas
e
(
tscCacheHandle
,
(
void
*
)
&
(
pMeterMetaInfo
->
pMeterMeta
),
false
);
int32_t
size
=
pMeta
->
numOfColumns
*
sizeof
(
SSchema
)
+
sizeof
(
STableMeta
);
pMeterMetaInfo
->
pMeterMeta
=
(
STableMeta
*
)
taos
AddDataIntoCache
(
tscCacheHandle
,
key
,
(
char
*
)
pMeta
,
size
,
tsMeterMetaKeepTimer
);
(
STableMeta
*
)
taos
CachePut
(
tscCacheHandle
,
key
,
(
char
*
)
pMeta
,
size
,
tsMeterMetaKeepTimer
);
pCmd
->
numOfCols
=
pQueryInfo
->
fieldsInfo
.
numOfOutputCols
;
SSchema
*
pMeterSchema
=
tsGetSchema
(
pMeterMetaInfo
->
pMeterMeta
);
...
...
@@ -2975,14 +2974,14 @@ int tscProcessUseDbRsp(SSqlObj *pSql) {
}
int
tscProcessDropDbRsp
(
SSqlObj
*
UNUSED_PARAM
(
pSql
))
{
taosC
learDataCache
(
tscCacheHandle
);
taosC
acheEmpty
(
tscCacheHandle
);
return
0
;
}
int
tscProcessDropTableRsp
(
SSqlObj
*
pSql
)
{
SMeterMetaInfo
*
pMeterMetaInfo
=
tscGetMeterMetaInfo
(
&
pSql
->
cmd
,
0
,
0
);
STableMeta
*
pMeterMeta
=
taos
GetDataFromCach
e
(
tscCacheHandle
,
pMeterMetaInfo
->
name
);
STableMeta
*
pMeterMeta
=
taos
CacheAcquireByNam
e
(
tscCacheHandle
,
pMeterMetaInfo
->
name
);
if
(
pMeterMeta
==
NULL
)
{
/* not in cache, abort */
return
0
;
...
...
@@ -2996,11 +2995,11 @@ int tscProcessDropTableRsp(SSqlObj *pSql) {
* instead.
*/
tscTrace
(
"%p force release metermeta after drop table:%s"
,
pSql
,
pMeterMetaInfo
->
name
);
taos
RemoveDataFromCach
e
(
tscCacheHandle
,
(
void
**
)
&
pMeterMeta
,
true
);
taos
CacheReleas
e
(
tscCacheHandle
,
(
void
**
)
&
pMeterMeta
,
true
);
if
(
pMeterMetaInfo
->
pMeterMeta
)
{
taos
RemoveDataFromCach
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMeterMeta
),
true
);
taos
RemoveDataFromCach
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMetricMeta
),
true
);
taos
CacheReleas
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMeterMeta
),
true
);
taos
CacheReleas
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMetricMeta
),
true
);
}
return
0
;
...
...
@@ -3009,23 +3008,23 @@ int tscProcessDropTableRsp(SSqlObj *pSql) {
int
tscProcessAlterTableMsgRsp
(
SSqlObj
*
pSql
)
{
SMeterMetaInfo
*
pMeterMetaInfo
=
tscGetMeterMetaInfo
(
&
pSql
->
cmd
,
0
,
0
);
STableMeta
*
pMeterMeta
=
taos
GetDataFromCach
e
(
tscCacheHandle
,
pMeterMetaInfo
->
name
);
STableMeta
*
pMeterMeta
=
taos
CacheAcquireByNam
e
(
tscCacheHandle
,
pMeterMetaInfo
->
name
);
if
(
pMeterMeta
==
NULL
)
{
/* not in cache, abort */
return
0
;
}
tscTrace
(
"%p force release metermeta in cache after alter-table: %s"
,
pSql
,
pMeterMetaInfo
->
name
);
taos
RemoveDataFromCach
e
(
tscCacheHandle
,
(
void
**
)
&
pMeterMeta
,
true
);
taos
CacheReleas
e
(
tscCacheHandle
,
(
void
**
)
&
pMeterMeta
,
true
);
if
(
pMeterMetaInfo
->
pMeterMeta
)
{
bool
isSuperTable
=
UTIL_METER_IS_SUPERTABLE
(
pMeterMetaInfo
);
taos
RemoveDataFromCach
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMeterMeta
),
true
);
taos
RemoveDataFromCach
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMetricMeta
),
true
);
taos
CacheReleas
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMeterMeta
),
true
);
taos
CacheReleas
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMetricMeta
),
true
);
if
(
isSuperTable
)
{
// if it is a super table, reset whole query cache
tscTrace
(
"%p reset query cache since table:%s is stable"
,
pSql
,
pMeterMetaInfo
->
name
);
taosC
learDataCache
(
tscCacheHandle
);
taosC
acheEmpty
(
tscCacheHandle
);
}
}
...
...
@@ -3151,7 +3150,7 @@ static int32_t doGetMeterMetaFromServer(SSqlObj *pSql, SMeterMetaInfo *pMeterMet
* Transfer the ownership of metermeta to the new object, instead of invoking the release/acquire routine
*/
if
(
code
==
TSDB_CODE_SUCCESS
)
{
pMeterMetaInfo
->
pMeterMeta
=
taos
TransferDataInCache
(
tscCacheHandle
,
(
void
**
)
&
pNewMeterMetaInfo
->
pMeterMeta
);
pMeterMetaInfo
->
pMeterMeta
=
taos
CacheTransfer
(
tscCacheHandle
,
(
void
**
)
&
pNewMeterMetaInfo
->
pMeterMeta
);
assert
(
pMeterMetaInfo
->
pMeterMeta
!=
NULL
);
}
...
...
@@ -3177,10 +3176,10 @@ int tscGetMeterMeta(SSqlObj *pSql, SMeterMetaInfo *pMeterMetaInfo) {
// If this SMeterMetaInfo owns a metermeta, release it first
if
(
pMeterMetaInfo
->
pMeterMeta
!=
NULL
)
{
taos
RemoveDataFromCach
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMeterMeta
),
false
);
taos
CacheReleas
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMeterMeta
),
false
);
}
pMeterMetaInfo
->
pMeterMeta
=
(
STableMeta
*
)
taos
GetDataFromCach
e
(
tscCacheHandle
,
pMeterMetaInfo
->
name
);
pMeterMetaInfo
->
pMeterMeta
=
(
STableMeta
*
)
taos
CacheAcquireByNam
e
(
tscCacheHandle
,
pMeterMetaInfo
->
name
);
if
(
pMeterMetaInfo
->
pMeterMeta
!=
NULL
)
{
STableMeta
*
pMeterMeta
=
pMeterMetaInfo
->
pMeterMeta
;
...
...
@@ -3244,7 +3243,7 @@ int tscRenewMeterMeta(SSqlObj *pSql, char *tableId) {
}
tscWaitingForCreateTable
(
pCmd
);
taos
RemoveDataFromCach
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMeterMeta
),
true
);
taos
CacheReleas
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMeterMeta
),
true
);
code
=
doGetMeterMetaFromServer
(
pSql
,
pMeterMetaInfo
);
// todo ??
}
else
{
...
...
@@ -3278,9 +3277,9 @@ int tscGetMetricMeta(SSqlObj *pSql, int32_t clauseIndex) {
SMeterMetaInfo
*
pMeterMetaInfo
=
tscGetMeterMetaInfoFromQueryInfo
(
pQueryInfo
,
i
);
tscGetMetricMetaCacheKey
(
pQueryInfo
,
tagstr
,
pMeterMetaInfo
->
pMeterMeta
->
uid
);
taos
RemoveDataFromCach
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMetricMeta
),
false
);
taos
CacheReleas
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMetricMeta
),
false
);
SSuperTableMeta
*
ppMeta
=
(
SSuperTableMeta
*
)
taos
GetDataFromCach
e
(
tscCacheHandle
,
tagstr
);
SSuperTableMeta
*
ppMeta
=
(
SSuperTableMeta
*
)
taos
CacheAcquireByNam
e
(
tscCacheHandle
,
tagstr
);
if
(
ppMeta
==
NULL
)
{
required
=
true
;
break
;
...
...
@@ -3308,7 +3307,7 @@ int tscGetMetricMeta(SSqlObj *pSql, int32_t clauseIndex) {
for
(
int32_t
i
=
0
;
i
<
pQueryInfo
->
numOfTables
;
++
i
)
{
SMeterMetaInfo
*
pMMInfo
=
tscGetMeterMetaInfoFromQueryInfo
(
pQueryInfo
,
i
);
STableMeta
*
pMeterMeta
=
taos
GetDataFromCach
e
(
tscCacheHandle
,
pMMInfo
->
name
);
STableMeta
*
pMeterMeta
=
taos
CacheAcquireByNam
e
(
tscCacheHandle
,
pMMInfo
->
name
);
tscAddMeterMetaInfo
(
pNewQueryInfo
,
pMMInfo
->
name
,
pMeterMeta
,
NULL
,
pMMInfo
->
numOfTags
,
pMMInfo
->
tagColumnIndex
);
}
...
...
@@ -3353,8 +3352,8 @@ int tscGetMetricMeta(SSqlObj *pSql, int32_t clauseIndex) {
printf
(
"create metric key:%s, index:%d
\n
"
,
tagstr
,
i
);
#endif
taos
RemoveDataFromCach
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMetricMeta
),
false
);
pMeterMetaInfo
->
pMetricMeta
=
(
SSuperTableMeta
*
)
taosGetDataFromCach
e
(
tscCacheHandle
,
tagstr
);
taos
CacheReleas
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMetricMeta
),
false
);
pMeterMetaInfo
->
pMetricMeta
=
(
SSuperTableMeta
*
)
taosCacheAcquireByNam
e
(
tscCacheHandle
,
tagstr
);
}
}
...
...
src/client/src/tscSub.c
浏览文件 @
f98efaf2
...
...
@@ -371,7 +371,7 @@ TAOS_RES *taos_consume(TAOS_SUB *tsub) {
pSql
->
sqlstr
=
NULL
;
taos_free_result_imp
(
pSql
,
0
);
pSql
->
sqlstr
=
sqlstr
;
taosC
learDataCache
(
tscCacheHandle
);
taosC
acheEmpty
(
tscCacheHandle
);
if
(
!
tscUpdateSubscription
(
pSub
->
taos
,
pSub
))
return
NULL
;
tscTrace
(
"meter synchronization completed"
);
}
else
{
...
...
src/client/src/tscSystem.c
浏览文件 @
f98efaf2
...
...
@@ -186,7 +186,7 @@ void taos_init_imp() {
refreshTime
=
refreshTime
>
2
?
2
:
refreshTime
;
refreshTime
=
refreshTime
<
1
?
1
:
refreshTime
;
if
(
tscCacheHandle
==
NULL
)
tscCacheHandle
=
taos
InitDataCache
(
tsMaxMeterConnections
/
2
,
tscTmr
,
refreshTime
);
if
(
tscCacheHandle
==
NULL
)
tscCacheHandle
=
taos
CacheInit
(
tscTmr
,
refreshTime
);
tscConnCache
=
taosOpenConnCache
(
tsMaxMeterConnections
*
2
,
NULL
/*taosCloseRpcConn*/
,
tscTmr
,
tsShellActivityTimer
*
1000
);
...
...
src/client/src/tscUtil.c
浏览文件 @
f98efaf2
...
...
@@ -505,7 +505,7 @@ void tscDestroyDataBlock(STableDataBlocks* pDataBlock) {
tfree
(
pDataBlock
->
params
);
// free the refcount for metermeta
taos
RemoveDataFromCach
e
(
tscCacheHandle
,
(
void
**
)
&
(
pDataBlock
->
pMeterMeta
),
false
);
taos
CacheReleas
e
(
tscCacheHandle
,
(
void
**
)
&
(
pDataBlock
->
pMeterMeta
),
false
);
tfree
(
pDataBlock
);
}
...
...
@@ -589,9 +589,9 @@ int32_t tscCopyDataBlockToPayload(SSqlObj* pSql, STableDataBlocks* pDataBlock) {
// set the correct metermeta object, the metermeta has been locked in pDataBlocks, so it must be in the cache
if
(
pMeterMetaInfo
->
pMeterMeta
!=
pDataBlock
->
pMeterMeta
)
{
strcpy
(
pMeterMetaInfo
->
name
,
pDataBlock
->
tableId
);
taos
RemoveDataFromCach
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMeterMeta
),
false
);
taos
CacheReleas
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMeterMeta
),
false
);
pMeterMetaInfo
->
pMeterMeta
=
taos
TransferDataInCache
(
tscCacheHandle
,
(
void
**
)
&
pDataBlock
->
pMeterMeta
);
pMeterMetaInfo
->
pMeterMeta
=
taos
CacheTransfer
(
tscCacheHandle
,
(
void
**
)
&
pDataBlock
->
pMeterMeta
);
}
else
{
assert
(
strncmp
(
pMeterMetaInfo
->
name
,
pDataBlock
->
tableId
,
tListLen
(
pDataBlock
->
tableId
))
==
0
);
}
...
...
@@ -665,7 +665,7 @@ int32_t tscCreateDataBlock(size_t initialSize, int32_t rowSize, int32_t startOff
* due to operation such as drop database. So here we add the reference count directly instead of invoke
* taosGetDataFromCache, which may return NULL value.
*/
dataBuf
->
pMeterMeta
=
taos
GetDataFromExists
(
tscCacheHandle
,
pMeterMeta
);
dataBuf
->
pMeterMeta
=
taos
CacheAcquireByData
(
tscCacheHandle
,
pMeterMeta
);
assert
(
initialSize
>
0
&&
pMeterMeta
!=
NULL
&&
dataBuf
->
pMeterMeta
!=
NULL
);
*
dataBlocks
=
dataBuf
;
...
...
@@ -1940,8 +1940,8 @@ void tscClearMeterMetaInfo(SMeterMetaInfo* pMeterMetaInfo, bool removeFromCache)
return
;
}
taos
RemoveDataFromCach
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMeterMeta
),
removeFromCache
);
taos
RemoveDataFromCach
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMetricMeta
),
removeFromCache
);
taos
CacheReleas
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMeterMeta
),
removeFromCache
);
taos
CacheReleas
e
(
tscCacheHandle
,
(
void
**
)
&
(
pMeterMetaInfo
->
pMetricMeta
),
removeFromCache
);
}
void
tscResetForNextRetrieve
(
SSqlRes
*
pRes
)
{
...
...
@@ -2071,16 +2071,16 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
SMeterMetaInfo
*
pFinalInfo
=
NULL
;
if
(
pPrevSql
==
NULL
)
{
STableMeta
*
pMeterMeta
=
taos
GetDataFromCach
e
(
tscCacheHandle
,
name
);
SSuperTableMeta
*
pMetricMeta
=
taos
GetDataFromCach
e
(
tscCacheHandle
,
key
);
STableMeta
*
pMeterMeta
=
taos
CacheAcquireByNam
e
(
tscCacheHandle
,
name
);
SSuperTableMeta
*
pMetricMeta
=
taos
CacheAcquireByNam
e
(
tscCacheHandle
,
key
);
pFinalInfo
=
tscAddMeterMetaInfo
(
pNewQueryInfo
,
name
,
pMeterMeta
,
pMetricMeta
,
pMeterMetaInfo
->
numOfTags
,
pMeterMetaInfo
->
tagColumnIndex
);
}
else
{
// transfer the ownership of pMeterMeta/pMetricMeta to the newly create sql object.
SMeterMetaInfo
*
pPrevInfo
=
tscGetMeterMetaInfo
(
&
pPrevSql
->
cmd
,
pPrevSql
->
cmd
.
clauseIndex
,
0
);
STableMeta
*
pPrevMeterMeta
=
taos
TransferDataInCache
(
tscCacheHandle
,
(
void
**
)
&
pPrevInfo
->
pMeterMeta
);
SSuperTableMeta
*
pPrevMetricMeta
=
taos
TransferDataInCache
(
tscCacheHandle
,
(
void
**
)
&
pPrevInfo
->
pMetricMeta
);
STableMeta
*
pPrevMeterMeta
=
taos
CacheTransfer
(
tscCacheHandle
,
(
void
**
)
&
pPrevInfo
->
pMeterMeta
);
SSuperTableMeta
*
pPrevMetricMeta
=
taos
CacheTransfer
(
tscCacheHandle
,
(
void
**
)
&
pPrevInfo
->
pMetricMeta
);
pFinalInfo
=
tscAddMeterMetaInfo
(
pNewQueryInfo
,
name
,
pPrevMeterMeta
,
pPrevMetricMeta
,
pMeterMetaInfo
->
numOfTags
,
pMeterMetaInfo
->
tagColumnIndex
);
...
...
src/query/inc/tcache.h
浏览文件 @
f98efaf2
...
...
@@ -20,7 +20,63 @@
extern
"C"
{
#endif
#include <stdbool.h>
#include "os.h"
#include "tref.h"
#include "hash.h"
typedef
struct
SCacheStatis
{
int64_t
missCount
;
int64_t
hitCount
;
int64_t
totalAccess
;
int64_t
refreshCount
;
int32_t
numOfCollision
;
}
SCacheStatis
;
typedef
struct
SCacheDataNode
{
uint64_t
addedTime
;
// the added time when this element is added or updated into cache
uint64_t
expiredTime
;
// expiredTime expiredTime when this element should be remove from cache
uint64_t
signature
;
uint32_t
size
;
// allocated size for current SCacheDataNode
uint16_t
keySize
:
15
;
bool
inTrash
:
1
;
// denote if it is in trash or not
T_REF_DECLARE
()
char
*
key
;
char
data
[];
}
SCacheDataNode
;
typedef
struct
STrashElem
{
struct
STrashElem
*
prev
;
struct
STrashElem
*
next
;
SCacheDataNode
*
pData
;
}
STrashElem
;
typedef
struct
{
int64_t
totalSize
;
// total allocated buffer in this hash table, SCacheObj is not included.
int64_t
refreshTime
;
/*
* to accommodate the old datanode which has the same key value of new one in hashList
* when an new node is put into cache, if an existed one with the same key:
* 1. if the old one does not be referenced, update it.
* 2. otherwise, move the old one to pTrash, addedTime the new one.
*
* when the node in pTrash does not be referenced, it will be release at the expired expiredTime
*/
STrashElem
*
pTrash
;
void
*
tmrCtrl
;
void
*
pTimer
;
SCacheStatis
statistics
;
SHashObj
*
pHashTable
;
int
numOfElemsInTrash
;
// number of element in trash
int16_t
deleting
;
// set the deleting flag to stop refreshing ASAP.
#if defined(LINUX)
pthread_rwlock_t
lock
;
#else
pthread_mutex_t
lock
;
#endif
}
SCacheObj
;
/**
*
...
...
@@ -30,7 +86,7 @@ extern "C" {
* not referenced by other objects
* @return
*/
void
*
taosInitDataCache
(
int
maxSessions
,
void
*
tmrCtrl
,
int64_t
refreshTimeInSeconds
);
SCacheObj
*
taosCacheInit
(
void
*
tmrCtrl
,
int64_t
refreshTimeInSeconds
);
/**
* add data into cache
...
...
@@ -42,28 +98,7 @@ void *taosInitDataCache(int maxSessions, void *tmrCtrl, int64_t refreshTimeInSec
* @param keepTime survival time in second
* @return cached element
*/
void
*
taosAddDataIntoCache
(
void
*
handle
,
char
*
key
,
char
*
pData
,
int
dataSize
,
int
keepTimeInSeconds
);
/**
* remove data in cache, the data will not be removed immediately.
* if it is referenced by other object, it will be remain in cache
* @param handle cache object
* @param data not the key, actually referenced data
* @param _remove force model, reduce the ref count and move the data into
* pTrash
*/
void
taosRemoveDataFromCache
(
void
*
handle
,
void
**
data
,
bool
_remove
);
/**
* update data in cache
* @param handle hash object handle(pointer)
* @param key key for hash
* @param pData actually data
* @param size length of data
* @param duration survival time of this object in cache
* @return new referenced data
*/
void
*
taosUpdateDataFromCache
(
void
*
handle
,
char
*
key
,
char
*
pData
,
int
size
,
int
duration
);
void
*
taosCachePut
(
void
*
handle
,
char
*
key
,
char
*
pData
,
int
dataSize
,
int
keepTimeInSeconds
);
/**
* get data from cache
...
...
@@ -71,40 +106,56 @@ void *taosUpdateDataFromCache(void *handle, char *key, char *pData, int size, in
* @param key key
* @return cached data or NULL
*/
void
*
taos
GetDataFromCach
e
(
void
*
handle
,
char
*
key
);
void
*
taos
CacheAcquireByNam
e
(
void
*
handle
,
char
*
key
);
/**
* release all allocated memory and destroy the cache object
* Add one reference count for the exist data, and assign this data for a new owner.
* The new owner needs to invoke the taosCacheRelease when it does not need this data anymore.
* This procedure is a faster version of taosCacheAcquireByName function, which avoids the sideeffect of the problem of
* the data is moved to trash, and taosCacheAcquireByName will fail to retrieve it again.
*
* @param handle
* @param data
* @return
*/
void
taosCleanUpDataCache
(
void
*
handle
);
void
*
taosCacheAcquireByData
(
void
*
handle
,
void
*
data
);
/**
*
move all data node into trash,clear node in trash can if it is not referenced by client
*
transfer the ownership of data in cache to another object without increasing reference count.
* @param handle
* @param data
* @return
*/
void
taosClearDataCache
(
void
*
handle
);
void
*
taosCacheTransfer
(
void
*
handle
,
void
**
data
);
/**
* Add one reference count for the exist data, and assign this data for a new owner.
* The new owner needs to invoke the taosRemoveDataFromCache when it does not need this data anymore.
* This procedure is a faster version of taosGetDataFromCache function, which avoids the sideeffect of the problem of the
* data is moved to trash, and taosGetDataFromCache will fail to retrieve it again.
*
* remove data in cache, the data will not be removed immediately.
* if it is referenced by other object, it will be remain in cache
* @param handle cache object
* @param data not the key, actually referenced data
* @param _remove force model, reduce the ref count and move the data into
* pTrash
*/
void
taosCacheRelease
(
void
*
handle
,
void
**
data
,
bool
_remove
);
/**
* move all data node into trash, clear node in trash can if it is not referenced by any clients
* @param handle
* @param data
* @return
*/
void
*
taosGetDataFromExists
(
void
*
handle
,
void
*
data
);
void
taosCacheEmpty
(
SCacheObj
*
pCacheObj
);
/**
* transfer the ownership of data in cache to another object without increasing reference count.
* release all allocated memory and destroy the cache object.
*
* This function only set the deleting flag, and the specific work of clean up cache is delegated to
* taosCacheRefresh function, which will executed every SCacheObj->refreshTime sec.
*
* If the value of SCacheObj->refreshTime is too large, the taosCacheRefresh function may not be invoked
* before the main thread terminated, in which case all allocated resources are simply recycled by OS.
*
* @param handle
* @param data
* @return
*/
void
*
taosTransferDataInCache
(
void
*
handle
,
void
**
data
);
void
taosCacheCleanup
(
SCacheObj
*
pCacheObj
);
#ifdef __cplusplus
}
...
...
src/util/inc/hash.h
浏览文件 @
f98efaf2
...
...
@@ -23,17 +23,18 @@ extern "C" {
#include "hashfunc.h"
#define HASH_MAX_CAPACITY (1024 * 1024 * 16)
#define HASH_VALUE_IN_TRASH (-1)
#define HASH_DEFAULT_LOAD_FACTOR (0.75)
#define HASH_INDEX(v, c) ((v) & ((c)-1))
typedef
void
(
*
_hash_free_fn_t
)(
void
*
param
);
typedef
struct
SHashNode
{
char
*
key
;
// null-terminated string
char
*
key
;
union
{
struct
SHashNode
*
prev
;
struct
SHashEntry
*
prev1
;
};
struct
SHashNode
*
next
;
uint32_t
hashVal
;
// the hash value of key, if hashVal == HASH_VALUE_IN_TRASH, this node is moved to trash
uint32_t
keyLen
;
// length of the key
...
...
@@ -46,18 +47,27 @@ typedef struct SHashEntry {
}
SHashEntry
;
typedef
struct
SHashObj
{
SHashEntry
**
hashList
;
size_t
capacity
;
// number of slots
size_t
size
;
// number of elements in hash table
_hash_fn_t
hashFp
;
// hash function
#if defined (LINUX)
pthread_rwlock_t
*
lock
;
SHashEntry
**
hashList
;
size_t
capacity
;
// number of slots
size_t
size
;
// number of elements in hash table
_hash_fn_t
hashFp
;
// hash function
_hash_free_fn_t
freeFp
;
// hash node free callback function
#if defined(LINUX)
pthread_rwlock_t
*
lock
;
#else
pthread_mutex_t
*
lock
;
pthread_mutex_t
*
lock
;
#endif
}
SHashObj
;
typedef
struct
SHashMutableIterator
{
SHashObj
*
pHashObj
;
int32_t
entryIndex
;
SHashNode
*
pCur
;
SHashNode
*
pNext
;
// current node can be deleted for mutable iterator, so keep the next one before return current
int32_t
num
;
// already check number of elements in hash table
}
SHashMutableIterator
;
/**
* init the hash table
*
...
...
@@ -102,7 +112,7 @@ void *taosHashGet(SHashObj *pHashObj, const char *key, size_t keyLen);
* @param key
* @param keyLen
*/
void
taosHashRemove
(
SHashObj
*
pHashObj
,
const
char
*
key
,
size_t
keyLen
);
void
taosHashRemove
(
SHashObj
*
pHashObj
,
const
char
*
key
,
size_t
keyLen
);
/**
* clean up hash table
...
...
@@ -110,6 +120,41 @@ void taosHashRemove(SHashObj *pHashObj, const char *key, size_t keyLen);
*/
void
taosHashCleanup
(
SHashObj
*
pHashObj
);
/**
* Set the free callback function
* This function if set will be invoked right before freeing each hash node
* @param pHashObj
*/
void
taosHashSetFreecb
(
SHashObj
*
pHashObj
,
_hash_free_fn_t
freeFp
);
/**
*
* @param pHashObj
* @return
*/
SHashMutableIterator
*
taosHashCreateIter
(
SHashObj
*
pHashObj
);
/**
*
* @param iter
* @return
*/
bool
taosHashIterNext
(
SHashMutableIterator
*
iter
);
/**
*
* @param iter
* @return
*/
void
*
taosHashIterGet
(
SHashMutableIterator
*
iter
);
/**
*
* @param iter
* @return
*/
void
*
taosHashDestroyIter
(
SHashMutableIterator
*
iter
);
/**
*
* @param pHashObj
...
...
src/util/inc/tref.h
0 → 100644
浏览文件 @
f98efaf2
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_TREF_H
#define TDENGINE_TREF_H
#include "os.h"
typedef
void
(
*
_ref_fn_t
)(
const
void
*
pObj
);
#define T_REF_DECLARE() \
struct { \
int16_t val; \
} _ref;
#define T_REF_REGISTER_FUNC(s, e) \
struct { \
_ref_fn_t start; \
_ref_fn_t end; \
} _ref_func = {.begin = (s), .end = (e)};
#define T_REF_INC(x) (atomic_add_fetch_16(&((x)->_ref.val), 1));
#define T_REF_INC_WITH_CB(x, p) \
do { \
int32_t v = atomic_add_fetch_32(&((x)->_ref.val), 1); \
if (v == 1 && (p)->_ref_func.begin != NULL) { \
(p)->_ref_func.begin((x)); \
} \
} while (0)
#define T_REF_DEC(x) (atomic_sub_fetch_16(&((x)->_ref.val), 1));
#define T_REF_DEC_WITH_CB(x, p) \
do { \
int32_t v = atomic_sub_fetch_16(&((x)->_ref.val), 1); \
if (v == 0 && (p)->_ref_func.end != NULL) { \
(p)->_ref_func.end((x)); \
} \
} while (0)
#define T_REF_VAL_CHECK(x) assert((x)->_ref.val >= 0);
#define T_REF_VAL_GET(x) (x)->_ref.val
#endif // TDENGINE_TREF_H
src/util/src/hash.c
浏览文件 @
f98efaf2
...
...
@@ -24,8 +24,8 @@ static FORCE_INLINE void __wr_lock(void *lock) {
if
(
lock
==
NULL
)
{
return
;
}
#if defined
(LINUX)
#if defined(LINUX)
pthread_rwlock_wrlock
(
lock
);
#else
pthread_mutex_lock
(
lock
);
...
...
@@ -36,8 +36,8 @@ static FORCE_INLINE void __rd_lock(void *lock) {
if
(
lock
==
NULL
)
{
return
;
}
#if defined
(LINUX)
#if defined(LINUX)
pthread_rwlock_rdlock
(
lock
);
#else
pthread_mutex_lock
(
lock
);
...
...
@@ -48,8 +48,8 @@ static FORCE_INLINE void __unlock(void *lock) {
if
(
lock
==
NULL
)
{
return
;
}
#if defined
(LINUX)
#if defined(LINUX)
pthread_rwlock_unlock
(
lock
);
#else
pthread_mutex_unlock
(
lock
);
...
...
@@ -60,8 +60,8 @@ static FORCE_INLINE int32_t __lock_init(void *lock) {
if
(
lock
==
NULL
)
{
return
0
;
}
#if defined
(LINUX)
#if defined(LINUX)
return
pthread_rwlock_init
(
lock
,
NULL
);
#else
return
pthread_mutex_init
(
lock
,
NULL
);
...
...
@@ -72,8 +72,8 @@ static FORCE_INLINE void __lock_destroy(void *lock) {
if
(
lock
==
NULL
)
{
return
;
}
#if defined
(LINUX)
#if defined(LINUX)
pthread_rwlock_destroy
(
lock
);
#else
pthread_mutex_destroy
(
lock
);
...
...
@@ -107,7 +107,7 @@ static void doUpdateHashTable(SHashObj *pHashObj, SHashNode *pNode) {
/**
* get SHashNode from hashlist, nodes from trash are not included.
* @param pHashObj
Cache objection
* @param pHashObj Cache objection
* @param key key for hash
* @param keyLen key length
* @return
...
...
@@ -155,24 +155,24 @@ static void taosHashTableResize(SHashObj *pHashObj) {
int32_t
newSize
=
pHashObj
->
capacity
<<
1U
;
if
(
newSize
>
HASH_MAX_CAPACITY
)
{
pTrace
(
"current capacity:%d, maximum capacity:%d, no resize applied due to limitation is reached"
,
pHashObj
->
capacity
,
HASH_MAX_CAPACITY
);
pTrace
(
"current capacity:%d, maximum capacity:%d, no resize applied due to limitation is reached"
,
pHashObj
->
capacity
,
HASH_MAX_CAPACITY
);
return
;
}
int64_t
st
=
taosGetTimestampUs
();
SHashEntry
**
pNewEntry
=
realloc
(
pHashObj
->
hashList
,
sizeof
(
SHashEntry
*
)
*
newSize
);
SHashEntry
**
pNewEntry
=
realloc
(
pHashObj
->
hashList
,
sizeof
(
SHashEntry
*
)
*
newSize
);
if
(
pNewEntry
==
NULL
)
{
pTrace
(
"cache resize failed due to out of memory, capacity remain:%d"
,
pHashObj
->
capacity
);
return
;
}
pHashObj
->
hashList
=
pNewEntry
;
for
(
int32_t
i
=
pHashObj
->
capacity
;
i
<
newSize
;
++
i
)
{
for
(
int32_t
i
=
pHashObj
->
capacity
;
i
<
newSize
;
++
i
)
{
pHashObj
->
hashList
[
i
]
=
calloc
(
1
,
sizeof
(
SHashEntry
));
}
pHashObj
->
capacity
=
newSize
;
for
(
int32_t
i
=
0
;
i
<
pHashObj
->
capacity
;
++
i
)
{
...
...
@@ -182,7 +182,7 @@ static void taosHashTableResize(SHashObj *pHashObj) {
if
(
pNode
!=
NULL
)
{
assert
(
pNode
->
prev1
==
pEntry
&&
pEntry
->
num
>
0
);
}
while
(
pNode
)
{
int32_t
j
=
HASH_INDEX
(
pNode
->
hashVal
,
pHashObj
->
capacity
);
if
(
j
==
i
)
{
// this key resides in the same slot, no need to relocate it
...
...
@@ -192,13 +192,13 @@ static void taosHashTableResize(SHashObj *pHashObj) {
// remove from current slot
assert
(
pNode
->
prev1
!=
NULL
);
if
(
pNode
->
prev1
==
pEntry
)
{
// first node of the overflow linked list
if
(
pNode
->
prev1
==
pEntry
)
{
// first node of the overflow linked list
pEntry
->
next
=
pNode
->
next
;
}
else
{
pNode
->
prev
->
next
=
pNode
->
next
;
}
pEntry
->
num
--
;
assert
(
pEntry
->
num
>=
0
);
...
...
@@ -214,13 +214,13 @@ static void taosHashTableResize(SHashObj *pHashObj) {
if
(
pNewIndexEntry
->
next
!=
NULL
)
{
assert
(
pNewIndexEntry
->
next
->
prev1
==
pNewIndexEntry
);
pNewIndexEntry
->
next
->
prev
=
pNode
;
}
pNode
->
next
=
pNewIndexEntry
->
next
;
pNode
->
prev1
=
pNewIndexEntry
;
pNewIndexEntry
->
next
=
pNode
;
pNewIndexEntry
->
num
++
;
...
...
@@ -258,14 +258,14 @@ SHashObj *taosHashInit(size_t capacity, _hash_fn_t fn, bool threadsafe) {
pHashObj
->
hashFp
=
fn
;
pHashObj
->
hashList
=
(
SHashEntry
**
)
calloc
(
pHashObj
->
capacity
,
sizeof
(
SHashEntry
*
));
pHashObj
->
hashList
=
(
SHashEntry
**
)
calloc
(
pHashObj
->
capacity
,
sizeof
(
SHashEntry
*
));
if
(
pHashObj
->
hashList
==
NULL
)
{
free
(
pHashObj
);
pError
(
"failed to allocate memory, reason:%s"
,
strerror
(
errno
));
return
NULL
;
}
for
(
int32_t
i
=
0
;
i
<
pHashObj
->
capacity
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pHashObj
->
capacity
;
++
i
)
{
pHashObj
->
hashList
[
i
]
=
calloc
(
1
,
sizeof
(
SHashEntry
));
}
...
...
@@ -276,7 +276,7 @@ SHashObj *taosHashInit(size_t capacity, _hash_fn_t fn, bool threadsafe) {
pHashObj
->
lock
=
calloc
(
1
,
sizeof
(
pthread_mutex_t
));
#endif
}
if
(
__lock_init
(
pHashObj
->
lock
)
!=
0
)
{
free
(
pHashObj
->
hashList
);
free
(
pHashObj
);
...
...
@@ -347,7 +347,7 @@ static void doAddToHashTable(SHashObj *pHashObj, SHashNode *pNode) {
int32_t
index
=
HASH_INDEX
(
pNode
->
hashVal
,
pHashObj
->
capacity
);
SHashEntry
*
pEntry
=
pHashObj
->
hashList
[
index
];
pNode
->
next
=
pEntry
->
next
;
if
(
pEntry
->
next
)
{
...
...
@@ -356,7 +356,7 @@ static void doAddToHashTable(SHashObj *pHashObj, SHashNode *pNode) {
pEntry
->
next
=
pNode
;
pNode
->
prev1
=
pEntry
;
pEntry
->
num
++
;
pHashObj
->
size
++
;
}
...
...
@@ -365,7 +365,7 @@ size_t taosHashGetSize(const SHashObj *pHashObj) {
if
(
pHashObj
==
NULL
)
{
return
0
;
}
return
pHashObj
->
size
;
}
...
...
@@ -430,7 +430,7 @@ void *taosHashGet(SHashObj *pHashObj, const char *key, size_t keyLen) {
void
taosHashRemove
(
SHashObj
*
pHashObj
,
const
char
*
key
,
size_t
keyLen
)
{
__wr_lock
(
pHashObj
->
lock
);
uint32_t
val
=
0
;
uint32_t
val
=
0
;
SHashNode
*
pNode
=
doGetNodeFromHashTable
(
pHashObj
,
key
,
keyLen
,
&
val
);
if
(
pNode
==
NULL
)
{
__unlock
(
pHashObj
->
lock
);
...
...
@@ -446,13 +446,13 @@ void taosHashRemove(SHashObj *pHashObj, const char *key, size_t keyLen) {
pNode
->
prev
->
next
=
pNext
;
}
}
if
(
pNext
!=
NULL
)
{
pNext
->
prev
=
pNode
->
prev
;
}
uint32_t
index
=
HASH_INDEX
(
pNode
->
hashVal
,
pHashObj
->
capacity
);
SHashEntry
*
pEntry
=
pHashObj
->
hashList
[
index
];
pEntry
->
num
--
;
...
...
@@ -483,10 +483,14 @@ void taosHashCleanup(SHashObj *pHashObj) {
while
(
pNode
)
{
pNext
=
pNode
->
next
;
if
(
pHashObj
->
freeFp
)
{
pHashObj
->
freeFp
(
pNode
->
data
);
}
free
(
pNode
);
pNode
=
pNext
;
}
tfree
(
pEntry
);
}
...
...
@@ -496,24 +500,122 @@ void taosHashCleanup(SHashObj *pHashObj) {
__unlock
(
pHashObj
->
lock
);
__lock_destroy
(
pHashObj
->
lock
);
tfree
(
pHashObj
->
lock
);
memset
(
pHashObj
,
0
,
sizeof
(
SHashObj
));
free
(
pHashObj
);
}
void
taosHashSetFreecb
(
SHashObj
*
pHashObj
,
_hash_free_fn_t
freeFp
)
{
if
(
pHashObj
==
NULL
||
freeFp
==
NULL
)
{
return
;
}
pHashObj
->
freeFp
=
freeFp
;
}
SHashMutableIterator
*
taosHashCreateIter
(
SHashObj
*
pHashObj
)
{
SHashMutableIterator
*
pIter
=
calloc
(
1
,
sizeof
(
SHashMutableIterator
));
if
(
pIter
==
NULL
)
{
return
NULL
;
}
pIter
->
pHashObj
=
pHashObj
;
}
static
SHashNode
*
getNextHashNode
(
SHashMutableIterator
*
pIter
)
{
assert
(
pIter
!=
NULL
);
while
(
pIter
->
entryIndex
<
pIter
->
pHashObj
->
capacity
)
{
SHashEntry
*
pEntry
=
pIter
->
pHashObj
->
hashList
[
pIter
->
entryIndex
];
if
(
pEntry
->
next
==
NULL
)
{
pIter
->
entryIndex
++
;
continue
;
}
return
pEntry
->
next
;
}
return
NULL
;
}
bool
taosHashIterNext
(
SHashMutableIterator
*
pIter
)
{
if
(
pIter
==
NULL
)
{
return
false
;
}
size_t
size
=
taosHashGetSize
(
pIter
->
pHashObj
);
if
(
size
==
0
||
pIter
->
num
>=
size
)
{
return
false
;
}
// check the first one
if
(
pIter
->
num
==
0
)
{
assert
(
pIter
->
pCur
==
NULL
&&
pIter
->
pNext
==
NULL
);
while
(
1
)
{
SHashEntry
*
pEntry
=
pIter
->
pHashObj
->
hashList
[
pIter
->
entryIndex
];
if
(
pEntry
->
next
==
NULL
)
{
pIter
->
entryIndex
++
;
continue
;
}
pIter
->
pCur
=
pEntry
->
next
;
if
(
pIter
->
pCur
->
next
)
{
pIter
->
pNext
=
pIter
->
pCur
->
next
;
}
else
{
pIter
->
pNext
=
getNextHashNode
(
pIter
);
}
break
;
}
pIter
->
num
++
;
return
true
;
}
else
{
assert
(
pIter
->
pCur
!=
NULL
);
if
(
pIter
->
pNext
)
{
pIter
->
pCur
=
pIter
->
pNext
;
}
else
{
// no more data in the hash list
return
false
;
}
pIter
->
num
++
;
if
(
pIter
->
pCur
->
next
)
{
pIter
->
pNext
=
pIter
->
pCur
->
next
;
}
else
{
pIter
->
pNext
=
getNextHashNode
(
pIter
);
}
return
true
;
}
}
void
*
taosHashIterGet
(
SHashMutableIterator
*
iter
)
{
return
(
iter
==
NULL
)
?
NULL
:
iter
->
pCur
->
data
;
}
void
*
taosHashDestroyIter
(
SHashMutableIterator
*
iter
)
{
if
(
iter
==
NULL
)
{
return
NULL
;
}
free
(
iter
);
}
// for profile only
int32_t
taosHashGetMaxOverflowLinkLength
(
const
SHashObj
*
pHashObj
)
{
int32_t
taosHashGetMaxOverflowLinkLength
(
const
SHashObj
*
pHashObj
)
{
if
(
pHashObj
==
NULL
||
pHashObj
->
size
==
0
)
{
return
0
;
}
int32_t
num
=
0
;
for
(
int32_t
i
=
0
;
i
<
pHashObj
->
size
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pHashObj
->
size
;
++
i
)
{
SHashEntry
*
pEntry
=
pHashObj
->
hashList
[
i
];
if
(
num
<
pEntry
->
num
)
{
num
=
pEntry
->
num
;
}
}
return
num
;
}
src/util/src/tskiplist.c
浏览文件 @
f98efaf2
...
...
@@ -253,6 +253,7 @@ static __compar_fn_t getKeyComparator(int32_t keyType) {
comparFn
=
compareInt32Val
;
break
;
case
TSDB_DATA_TYPE_BIGINT
:
case
TSDB_DATA_TYPE_TIMESTAMP
:
comparFn
=
compareInt64Val
;
break
;
case
TSDB_DATA_TYPE_BOOL
:
...
...
src/vnode/tsdb/inc/tsdb.h
浏览文件 @
f98efaf2
...
...
@@ -68,6 +68,7 @@ typedef struct {
typedef
struct
{
TSDB_TABLE_TYPE
type
;
STableId
tableId
;
int32_t
sversion
;
int64_t
superUid
;
STSchema
*
schema
;
STSchema
*
tagSchema
;
...
...
src/vnode/tsdb/inc/tsdbFile.h
浏览文件 @
f98efaf2
...
...
@@ -23,8 +23,6 @@
extern
"C"
{
#endif
typedef
int32_t
file_id_t
;
typedef
enum
{
TSDB_FILE_TYPE_HEAD
,
// .head file type
TSDB_FILE_TYPE_DATA
,
// .data file type
...
...
@@ -40,19 +38,33 @@ typedef struct {
}
SFileInfo
;
typedef
struct
{
char
*
fname
;
SFileInfo
fInfo
;
}
SFILE
;
int
fd
;
int64_t
size
;
// total size of the file
int64_t
tombSize
;
// unused file size
}
SFile
;
// typedef struct {
// int64_t offset;
// int64_t skey;
// int64_t ekey;
// int16_t numOfBlocks;
// } SDataBlock;
typedef
struct
{
int32_t
fileId
;
SFile
fhead
;
SFile
fdata
;
SFile
flast
;
}
SFileGroup
;
// TSDB file handle
typedef
struct
{
int32_t
daysPerFile
;
int32_t
keep
;
int32_t
minRowPerFBlock
;
int32_t
maxRowsPerFBlock
;
SFileGroup
fGroup
[];
}
STsdbFileH
;
#define IS_VALID_TSDB_FILE_TYPE(type) ((type) >= TSDB_FILE_TYPE_HEAD && (type) <= TSDB_FILE_TYPE_META)
STsdbFileH
*
tsdbInitFile
(
char
*
dataDir
,
int32_t
daysPerFile
,
int32_t
keep
,
int32_t
minRowsPerFBlock
,
int32_t
maxRowsPerFBlock
);
void
tsdbCloseFile
(
STsdbFileH
*
pFileH
);
char
*
tsdbGetFileName
(
char
*
dirName
,
char
*
fname
,
TSDB_FILE_TYPE
type
);
#ifdef __cplusplus
...
...
src/vnode/tsdb/inc/tsdbMeta.h
浏览文件 @
f98efaf2
...
...
@@ -38,6 +38,7 @@ typedef struct STable {
TSDB_TABLE_TYPE
type
;
STableId
tableId
;
int32_t
superUid
;
// Super table UID
int32_t
sversion
;
STSchema
*
schema
;
STSchema
*
tagSchema
;
SDataRow
tagVal
;
...
...
src/vnode/tsdb/src/tsdbFile.c
浏览文件 @
f98efaf2
...
...
@@ -14,9 +14,21 @@
*/
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <string.h>
#include <dirent.h>
#include "tsdbFile.h"
#include "tglobalcfg.h"
// int64_t tsMsPerDay[] = {
// 86400000L, // TSDB_PRECISION_MILLI
// 86400000000L, // TSDB_PRECISION_MICRO
// 86400000000000L // TSDB_PRECISION_NANO
// };
#define tsdbGetKeyFileId(key, daysPerFile, precision) ((key) / tsMsPerDay[(precision)] / (daysPerFile))
#define tsdbGetMaxNumOfFiles(keep, daysPerFile) ((keep) / (daysPerFile) + 3)
typedef
struct
{
int64_t
offset
;
...
...
@@ -71,6 +83,55 @@ const char *tsdbFileSuffix[] = {
".meta"
// TSDB_FILE_TYPE_META
};
/**
* Initialize the TSDB file handle
*/
STsdbFileH
*
tsdbInitFile
(
char
*
dataDir
,
int32_t
daysPerFile
,
int32_t
keep
,
int32_t
minRowsPerFBlock
,
int32_t
maxRowsPerFBlock
)
{
STsdbFileH
*
pTsdbFileH
=
(
STsdbFileH
*
)
calloc
(
1
,
sizeof
(
STsdbFileH
)
+
sizeof
(
SFileGroup
)
*
tsdbGetMaxNumOfFiles
(
keep
,
daysPerFile
));
if
(
pTsdbFileH
==
NULL
)
return
NULL
;
pTsdbFileH
->
daysPerFile
=
daysPerFile
;
pTsdbFileH
->
keep
=
keep
;
pTsdbFileH
->
minRowPerFBlock
=
minRowsPerFBlock
;
pTsdbFileH
->
maxRowsPerFBlock
=
maxRowsPerFBlock
;
// Open the directory to read information of each file
DIR
*
dir
=
opendir
(
dataDir
);
if
(
dir
==
NULL
)
{
free
(
pTsdbFileH
);
return
NULL
;
}
struct
dirent
*
dp
;
char
fname
[
256
];
while
((
dp
=
readdir
(
dir
))
!=
NULL
)
{
if
(
strncmp
(
dp
->
d_name
,
"."
,
1
)
==
0
||
strncmp
(
dp
->
d_name
,
".."
,
2
)
==
0
)
continue
;
if
(
true
/* check if the file is the .head file */
)
{
int
fileId
=
0
;
int
vgId
=
0
;
sscanf
(
dp
->
d_name
,
"v%df%d.head"
,
&
vgId
,
&
fileId
);
// TODO
// Open head file
// Open data file
// Open last file
}
}
return
pTsdbFileH
;
}
/**
* Closet the file handle
*/
void
tsdbCloseFile
(
STsdbFileH
*
pFileH
)
{
// TODO
}
char
*
tsdbGetFileName
(
char
*
dirName
,
char
*
fname
,
TSDB_FILE_TYPE
type
)
{
if
(
!
IS_VALID_TSDB_FILE_TYPE
(
type
))
return
NULL
;
...
...
@@ -79,4 +140,10 @@ char *tsdbGetFileName(char *dirName, char *fname, TSDB_FILE_TYPE type) {
sprintf
(
fileName
,
"%s/%s%s"
,
dirName
,
fname
,
tsdbFileSuffix
[
type
]);
return
fileName
;
}
static
void
tsdbGetKeyRangeOfFileId
(
int32_t
daysPerFile
,
int8_t
precision
,
int32_t
fileId
,
TSKEY
*
minKey
,
TSKEY
*
maxKey
)
{
*
minKey
=
fileId
*
daysPerFile
*
tsMsPerDay
[
precision
];
*
maxKey
=
*
minKey
+
daysPerFile
*
tsMsPerDay
[
precision
]
-
1
;
}
\ No newline at end of file
src/vnode/tsdb/src/tsdbMain.c
浏览文件 @
f98efaf2
...
...
@@ -308,7 +308,6 @@ int tsdbAlterTable(tsdb_repo_t *pRepo, STableCfg *pCfg) {
}
int
tsdbDropTable
(
tsdb_repo_t
*
repo
,
STableId
tableId
)
{
// TODO
if
(
repo
==
NULL
)
return
-
1
;
STsdbRepo
*
pRepo
=
(
STsdbRepo
*
)
repo
;
...
...
src/vnode/tsdb/src/tsdbMeta.c
浏览文件 @
f98efaf2
...
...
@@ -18,6 +18,7 @@ static int tsdbAddTableIntoMap(STsdbMeta *pMeta, STable *pTable);
static
int
tsdbAddTableIntoIndex
(
STsdbMeta
*
pMeta
,
STable
*
pTable
);
static
int
tsdbRemoveTableFromIndex
(
STsdbMeta
*
pMeta
,
STable
*
pTable
);
static
int
tsdbEstimateTableEncodeSize
(
STable
*
pTable
);
static
char
*
getTupleKey
(
const
void
*
data
);
/**
* Encode a TSDB table object as a binary content
...
...
@@ -153,7 +154,7 @@ int32_t tsdbCreateTableImpl(STsdbMeta *pMeta, STableCfg *pCfg) {
super
->
tagSchema
=
tdDupSchema
(
pCfg
->
tagSchema
);
super
->
tagVal
=
tdDataRowDup
(
pCfg
->
tagValues
);
super
->
content
.
pIndex
=
tSkipListCreate
(
TSDB_SUPER_TABLE_SL_LEVEL
,
TSDB_DATA_TYPE_TIMESTAMP
,
sizeof
(
int64_t
),
1
,
0
,
NULL
);
// Allow duplicate key, no lock
0
,
getTupleKey
);
// Allow duplicate key, no lock
if
(
super
->
content
.
pIndex
==
NULL
)
{
tdFreeSchema
(
super
->
schema
);
...
...
@@ -183,7 +184,7 @@ int32_t tsdbCreateTableImpl(STsdbMeta *pMeta, STableCfg *pCfg) {
table
->
superUid
=
-
1
;
table
->
schema
=
tdDupSchema
(
pCfg
->
schema
);
}
table
->
content
.
pData
=
tSkipListCreate
(
TSDB_SUPER_TABLE_SL_LEVEL
,
0
,
8
,
0
,
0
,
NULL
);
table
->
content
.
pData
=
tSkipListCreate
(
TSDB_SUPER_TABLE_SL_LEVEL
,
TSDB_DATA_TYPE_TIMESTAMP
,
TYPE_BYTES
[
TSDB_DATA_TYPE_TIMESTAMP
],
0
,
0
,
getTupleKey
);
if
(
newSuper
)
tsdbAddTableToMeta
(
pMeta
,
super
);
tsdbAddTableToMeta
(
pMeta
,
table
);
...
...
@@ -319,4 +320,10 @@ static int tsdbRemoveTableFromIndex(STsdbMeta *pMeta, STable *pTable) {
static
int
tsdbEstimateTableEncodeSize
(
STable
*
pTable
)
{
// TODO
return
0
;
}
static
char
*
getTupleKey
(
const
void
*
data
)
{
SDataRow
row
=
(
SDataRow
)
data
;
return
dataRowAt
(
row
,
TD_DATA_ROW_HEAD_SIZE
);
}
\ No newline at end of file
src/vnode/tsdb/src/tsdbMetaFile.c
浏览文件 @
f98efaf2
...
...
@@ -83,7 +83,7 @@ int32_t tsdbInsertMetaRecord(SMetaFile *mfh, int64_t uid, void *cont, int32_t co
}
// TODO: make below a function to implement
if
(
f
seek
(
mfh
->
fd
,
info
.
offset
,
SEEK_CUR
)
<
0
)
{
if
(
l
seek
(
mfh
->
fd
,
info
.
offset
,
SEEK_CUR
)
<
0
)
{
return
-
1
;
}
...
...
@@ -114,7 +114,7 @@ int32_t tsdbDeleteMetaRecord(SMetaFile *mfh, int64_t uid) {
// Remove record from file
info
.
offset
=
-
info
.
offset
;
if
(
f
seek
(
mfh
->
fd
,
-
info
.
offset
,
SEEK_CUR
)
<
0
)
{
if
(
l
seek
(
mfh
->
fd
,
-
info
.
offset
,
SEEK_CUR
)
<
0
)
{
return
-
1
;
}
...
...
@@ -149,7 +149,7 @@ int32_t tsdbUpdateMetaRecord(SMetaFile *mfh, int64_t uid, void *cont, int32_t co
mfh
->
size
+=
contLen
;
}
if
(
f
seek
(
mfh
->
fd
,
-
info
.
offset
,
SEEK_CUR
)
<
0
)
{
if
(
l
seek
(
mfh
->
fd
,
-
info
.
offset
,
SEEK_CUR
)
<
0
)
{
return
-
1
;
}
...
...
@@ -212,7 +212,7 @@ static int tsdbRestoreFromMetaFile(char *fname, SMetaFile *mfh) {
return
-
1
;
}
if
(
f
seek
(
fd
,
TSDB_META_FILE_HEADER_SIZE
,
SEEK_SET
)
<
0
)
{
if
(
l
seek
(
fd
,
TSDB_META_FILE_HEADER_SIZE
,
SEEK_SET
)
<
0
)
{
// TODO: deal with the error
close
(
fd
);
return
-
1
;
...
...
src/vnode/tsdb/tests/tsdbTests.cpp
浏览文件 @
f98efaf2
...
...
@@ -48,7 +48,7 @@ TEST(TsdbTest, createRepo) {
for
(
int
j
=
0
;
j
<
schemaNCols
(
schema
);
j
++
)
{
if
(
j
==
0
)
{
// Just for timestamp
tdAppendColVal
(
row
,
(
void
*
)(
&
time
),
schemaColAt
(
schema
,
j
));
tdAppendColVal
(
row
,
(
void
*
)(
&
t
t
ime
),
schemaColAt
(
schema
,
j
));
}
else
{
// For int
int
val
=
10
;
tdAppendColVal
(
row
,
(
void
*
)(
&
val
),
schemaColAt
(
schema
,
j
));
...
...
@@ -61,5 +61,7 @@ TEST(TsdbTest, createRepo) {
pMsg
->
length
=
pMsg
->
length
+
sizeof
(
SSubmitBlk
)
+
pBlock
->
len
;
tsdbInsertData
(
pRepo
,
pMsg
);
int
k
=
0
;
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录