Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
dgr6878
TDengine
提交
d34cabf2
T
TDengine
项目概览
dgr6878
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
8
Star
0
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,发现更多精彩内容 >>
提交
d34cabf2
编写于
5月 29, 2020
作者:
B
Bomin Zhang
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
TD-449: fix several bugs
上级
fe300f3c
变更
10
隐藏空白更改
内联
并排
Showing
10 changed file
with
50 addition
and
13 deletion
+50
-13
src/cq/src/cqMain.c
src/cq/src/cqMain.c
+4
-2
src/dnode/inc/dnodeModule.h
src/dnode/inc/dnodeModule.h
+2
-1
src/dnode/src/dnodeMain.c
src/dnode/src/dnodeMain.c
+1
-0
src/dnode/src/dnodeMgmt.c
src/dnode/src/dnodeMgmt.c
+17
-1
src/inc/tcq.h
src/inc/tcq.h
+1
-0
src/inc/tsdb.h
src/inc/tsdb.h
+1
-0
src/inc/vnode.h
src/inc/vnode.h
+1
-0
src/tsdb/src/tsdbMain.c
src/tsdb/src/tsdbMain.c
+12
-0
src/tsdb/src/tsdbMeta.c
src/tsdb/src/tsdbMeta.c
+0
-8
src/vnode/src/vnodeMain.c
src/vnode/src/vnodeMain.c
+11
-1
未找到文件。
src/cq/src/cqMain.c
浏览文件 @
d34cabf2
...
...
@@ -38,6 +38,7 @@ typedef struct {
int
vgId
;
char
user
[
TSDB_USER_LEN
];
char
pass
[
TSDB_PASSWORD_LEN
];
char
db
[
TSDB_DB_NAME_LEN
];
FCqWrite
cqWrite
;
void
*
ahandle
;
int
num
;
// number of continuous streams
...
...
@@ -73,6 +74,7 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
strcpy
(
pContext
->
user
,
pCfg
->
user
);
strcpy
(
pContext
->
pass
,
pCfg
->
pass
);
strcpy
(
pContext
->
db
,
pCfg
->
db
);
pContext
->
vgId
=
pCfg
->
vgId
;
pContext
->
cqWrite
=
pCfg
->
cqWrite
;
pContext
->
ahandle
=
ahandle
;
...
...
@@ -207,9 +209,8 @@ void cqDrop(void *handle) {
}
static
void
cqCreateStream
(
SCqContext
*
pContext
,
SCqObj
*
pObj
)
{
if
(
pContext
->
dbConn
==
NULL
)
{
pContext
->
dbConn
=
taos_connect
(
"localhost"
,
pContext
->
user
,
pContext
->
pass
,
NULL
,
0
);
pContext
->
dbConn
=
taos_connect
(
"localhost"
,
pContext
->
user
,
pContext
->
pass
,
pContext
->
db
,
0
);
if
(
pContext
->
dbConn
==
NULL
)
{
cError
(
"vgId:%d, failed to connect to TDengine(%s)"
,
pContext
->
vgId
,
tstrerror
(
terrno
));
}
...
...
@@ -217,6 +218,7 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
}
int64_t
lastKey
=
0
;
pObj
->
pContext
=
pContext
;
pObj
->
pStream
=
taos_open_stream
(
pContext
->
dbConn
,
pObj
->
sqlStr
,
cqProcessStreamRes
,
lastKey
,
pObj
,
NULL
);
if
(
pObj
->
pStream
)
{
pContext
->
num
++
;
...
...
src/dnode/inc/dnodeModule.h
浏览文件 @
d34cabf2
...
...
@@ -22,7 +22,8 @@ extern "C" {
int32_t
dnodeInitModules
();
void
dnodeStartModules
();
void
dnodeCleanupModules
();
void
dnodeStartStream
();
void
dnodeCleanUpModules
();
void
dnodeProcessModuleStatus
(
uint32_t
moduleStatus
);
#ifdef __cplusplus
...
...
src/dnode/src/dnodeMain.c
浏览文件 @
d34cabf2
...
...
@@ -124,6 +124,7 @@ int32_t dnodeInitSystem() {
dnodeStartModules
();
dnodeSetRunStatus
(
TSDB_DNODE_RUN_STATUS_RUNING
);
dnodeStartStream
();
dPrint
(
"TDengine is initialized successfully"
);
...
...
src/dnode/src/dnodeMgmt.c
浏览文件 @
d34cabf2
...
...
@@ -260,11 +260,27 @@ static int32_t dnodeOpenVnodes() {
}
free
(
vnodeList
);
dPrint
(
"there are total vnodes:%d, openned:%d failed:%d"
,
numOfVnodes
,
numOfVnodes
-
failed
,
failed
);
return
TSDB_CODE_SUCCESS
;
}
void
dnodeStartStream
()
{
int32_t
vnodeList
[
TSDB_MAX_VNODES
];
int32_t
numOfVnodes
=
0
;
int32_t
status
=
dnodeGetVnodeList
(
vnodeList
,
&
numOfVnodes
);
if
(
status
!=
TSDB_CODE_SUCCESS
)
{
dPrint
(
"Get dnode list failed"
);
return
;
}
for
(
int32_t
i
=
0
;
i
<
numOfVnodes
;
++
i
)
{
vnodeStartStream
(
vnodeList
[
i
]);
}
dPrint
(
"streams started"
);
}
static
void
dnodeCloseVnodes
()
{
int32_t
*
vnodeList
=
(
int32_t
*
)
malloc
(
sizeof
(
int32_t
)
*
TSDB_MAX_VNODES
);
int32_t
numOfVnodes
;
...
...
src/inc/tcq.h
浏览文件 @
d34cabf2
...
...
@@ -27,6 +27,7 @@ typedef struct {
int
vgId
;
char
user
[
TSDB_USER_LEN
];
char
pass
[
TSDB_PASSWORD_LEN
];
char
db
[
TSDB_DB_NAME_LEN
];
FCqWrite
cqWrite
;
}
SCqCfg
;
...
...
src/inc/tsdb.h
浏览文件 @
d34cabf2
...
...
@@ -118,6 +118,7 @@ int tsdbDropTable(TsdbRepoT *pRepo, STableId tableId);
int
tsdbAlterTable
(
TsdbRepoT
*
repo
,
STableCfg
*
pCfg
);
int
tsdbUpdateTagValue
(
TsdbRepoT
*
repo
,
SUpdateTableTagValMsg
*
pMsg
);
TSKEY
tsdbGetTableLastKey
(
TsdbRepoT
*
repo
,
uint64_t
uid
);
void
tsdbStartStream
(
TsdbRepoT
*
repo
);
uint32_t
tsdbGetFileInfo
(
TsdbRepoT
*
repo
,
char
*
name
,
uint32_t
*
index
,
uint32_t
eindex
,
int32_t
*
size
);
...
...
src/inc/vnode.h
浏览文件 @
d34cabf2
...
...
@@ -38,6 +38,7 @@ typedef struct {
int32_t
vnodeCreate
(
SMDCreateVnodeMsg
*
pVnodeCfg
);
int32_t
vnodeDrop
(
int32_t
vgId
);
int32_t
vnodeOpen
(
int32_t
vgId
,
char
*
rootDir
);
int32_t
vnodeStartStream
(
int32_t
vgId
);
int32_t
vnodeAlter
(
void
*
pVnode
,
SMDCreateVnodeMsg
*
pVnodeCfg
);
int32_t
vnodeClose
(
int32_t
vgId
);
...
...
src/tsdb/src/tsdbMain.c
浏览文件 @
d34cabf2
...
...
@@ -474,6 +474,18 @@ TSKEY tsdbGetTableLastKey(TsdbRepoT *repo, uint64_t uid) {
return
TSDB_GET_TABLE_LAST_KEY
(
pTable
);
}
void
tsdbStartStream
(
TsdbRepoT
*
repo
)
{
STsdbRepo
*
pRepo
=
(
STsdbRepo
*
)
repo
;
STsdbMeta
*
pMeta
=
pRepo
->
tsdbMeta
;
for
(
int
i
=
0
;
i
<
pRepo
->
config
.
maxTables
;
i
++
)
{
STable
*
pTable
=
pMeta
->
tables
[
i
];
if
(
pTable
&&
pTable
->
type
==
TSDB_STREAM_TABLE
)
{
pTable
->
cqhandle
=
(
*
pRepo
->
appH
.
cqCreateFunc
)(
pRepo
->
appH
.
cqH
,
pTable
->
tableId
.
tid
,
pTable
->
sql
,
tsdbGetTableSchema
(
pMeta
,
pTable
));
}
}
}
STableInfo
*
tsdbGetTableInfo
(
TsdbRepoT
*
pRepo
,
STableId
tableId
)
{
// TODO
return
NULL
;
...
...
src/tsdb/src/tsdbMeta.c
浏览文件 @
d34cabf2
...
...
@@ -150,7 +150,6 @@ int tsdbRestoreTable(void *pHandle, void *cont, int contLen) {
void
tsdbOrgMeta
(
void
*
pHandle
)
{
STsdbMeta
*
pMeta
=
(
STsdbMeta
*
)
pHandle
;
STsdbRepo
*
pRepo
=
(
STsdbRepo
*
)
pMeta
->
pRepo
;
for
(
int
i
=
1
;
i
<
pMeta
->
maxTables
;
i
++
)
{
STable
*
pTable
=
pMeta
->
tables
[
i
];
...
...
@@ -158,13 +157,6 @@ void tsdbOrgMeta(void *pHandle) {
tsdbAddTableIntoIndex
(
pMeta
,
pTable
);
}
}
for
(
int
i
=
0
;
i
<
pMeta
->
maxTables
;
i
++
)
{
STable
*
pTable
=
pMeta
->
tables
[
i
];
if
(
pTable
&&
pTable
->
type
==
TSDB_STREAM_TABLE
)
{
pTable
->
cqhandle
=
(
*
pRepo
->
appH
.
cqCreateFunc
)(
pRepo
->
appH
.
cqH
,
i
,
pTable
->
sql
,
tsdbGetTableSchema
(
pMeta
,
pTable
));
}
}
}
/**
...
...
src/vnode/src/vnodeMain.c
浏览文件 @
d34cabf2
...
...
@@ -208,8 +208,9 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
}
SCqCfg
cqCfg
=
{
0
};
sprintf
(
cqCfg
.
user
,
"root"
);
sprintf
(
cqCfg
.
user
,
"
_
root"
);
strcpy
(
cqCfg
.
pass
,
tsInternalPass
);
strcpy
(
cqCfg
.
db
,
"s1_db0"
);
// TODO: replace hard coded db name
cqCfg
.
vgId
=
vnode
;
cqCfg
.
cqWrite
=
vnodeWriteToQueue
;
pVnode
->
cq
=
cqOpen
(
pVnode
,
&
cqCfg
);
...
...
@@ -277,6 +278,15 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
return
TSDB_CODE_SUCCESS
;
}
int32_t
vnodeStartStream
(
int32_t
vnode
)
{
SVnodeObj
*
pVnode
=
vnodeAccquireVnode
(
vnode
);
if
(
pVnode
!=
NULL
)
{
tsdbStartStream
(
pVnode
->
tsdb
);
vnodeRelease
(
pVnode
);
}
return
TSDB_CODE_SUCCESS
;
}
int32_t
vnodeClose
(
int32_t
vgId
)
{
SVnodeObj
**
ppVnode
=
(
SVnodeObj
**
)
taosHashGet
(
tsDnodeVnodesHash
,
(
const
char
*
)
&
vgId
,
sizeof
(
int32_t
));
if
(
ppVnode
==
NULL
||
*
ppVnode
==
NULL
)
return
0
;
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录