Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
冷面1992
TDengine
提交
76d4cdf8
TDengine
项目概览
冷面1992
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
76d4cdf8
编写于
4月 08, 2020
作者:
S
slguan
浏览文件
操作
浏览文件
下载
差异文件
Merge remote-tracking branch 'origin/develop' into feature/balance
上级
cfd3eb2f
3c8d8bc4
变更
6
隐藏空白更改
内联
并排
Showing
6 changed file
with
83 addition
and
48 deletion
+83
-48
src/inc/mnode.h
src/inc/mnode.h
+4
-4
src/mnode/inc/mgmtTable.h
src/mnode/inc/mgmtTable.h
+1
-1
src/mnode/src/mgmtTable.c
src/mnode/src/mgmtTable.c
+6
-6
src/mnode/src/mgmtVgroup.c
src/mnode/src/mgmtVgroup.c
+1
-1
src/vnode/main/inc/vnodeInt.h
src/vnode/main/inc/vnodeInt.h
+6
-2
src/vnode/main/src/vnodeMain.c
src/vnode/main/src/vnodeMain.c
+65
-34
未找到文件。
src/inc/mnode.h
浏览文件 @
76d4cdf8
...
...
@@ -105,10 +105,10 @@ typedef struct {
typedef
struct
{
char
tableId
[
TSDB_TABLE_ID_LEN
+
1
];
int8_t
type
;
}
STable
Info
;
}
STable
Obj
;
typedef
struct
SSuperTableObj
{
STable
Info
info
;
STable
Obj
info
;
uint64_t
uid
;
int64_t
createdTime
;
int32_t
sversion
;
...
...
@@ -123,7 +123,7 @@ typedef struct SSuperTableObj {
}
SSuperTableObj
;
typedef
struct
{
STable
Info
info
;
STable
Obj
info
;
uint64_t
uid
;
int64_t
createdTime
;
int32_t
sversion
;
//used by normal table
...
...
@@ -255,7 +255,7 @@ typedef struct {
SUserObj
*
pUser
;
SDbObj
*
pDb
;
SVgObj
*
pVgroup
;
STable
Info
*
pTable
;
STable
Obj
*
pTable
;
}
SQueuedMsg
;
int32_t
mgmtInitSystem
();
...
...
src/mnode/inc/mgmtTable.h
浏览文件 @
76d4cdf8
...
...
@@ -27,7 +27,7 @@ extern "C" {
int32_t
mgmtInitTables
();
void
mgmtCleanUpTables
();
STable
Info
*
mgmtGetTable
(
char
*
tableId
);
STable
Obj
*
mgmtGetTable
(
char
*
tableId
);
void
mgmtIncTableRef
(
void
*
pTable
);
void
mgmtDecTableRef
(
void
*
pTable
);
void
mgmtDropAllChildTables
(
SDbObj
*
pDropDb
);
...
...
src/mnode/src/mgmtTable.c
浏览文件 @
76d4cdf8
...
...
@@ -486,8 +486,8 @@ static void *mgmtGetSuperTable(char *tableId) {
return
sdbGetRow
(
tsSuperTableSdb
,
tableId
);
}
STable
Info
*
mgmtGetTable
(
char
*
tableId
)
{
STable
Info
*
tableInfo
=
sdbGetRow
(
tsSuperTableSdb
,
tableId
);
STable
Obj
*
mgmtGetTable
(
char
*
tableId
)
{
STable
Obj
*
tableInfo
=
sdbGetRow
(
tsSuperTableSdb
,
tableId
);
if
(
tableInfo
!=
NULL
)
{
return
tableInfo
;
}
...
...
@@ -501,7 +501,7 @@ STableInfo *mgmtGetTable(char *tableId) {
}
void
mgmtIncTableRef
(
void
*
p1
)
{
STable
Info
*
pTable
=
(
STableInfo
*
)
p1
;
STable
Obj
*
pTable
=
(
STableObj
*
)
p1
;
if
(
pTable
->
type
==
TSDB_SUPER_TABLE
)
{
sdbIncRef
(
tsSuperTableSdb
,
pTable
);
}
else
{
...
...
@@ -512,7 +512,7 @@ void mgmtIncTableRef(void *p1) {
void
mgmtDecTableRef
(
void
*
p1
)
{
if
(
p1
==
NULL
)
return
;
STable
Info
*
pTable
=
(
STableInfo
*
)
p1
;
STable
Obj
*
pTable
=
(
STableObj
*
)
p1
;
if
(
pTable
->
type
==
TSDB_SUPER_TABLE
)
{
sdbDecRef
(
tsSuperTableSdb
,
pTable
);
}
else
{
...
...
@@ -1302,7 +1302,7 @@ static void mgmtProcessCreateChildTableMsg(SQueuedMsg *pMsg) {
return
;
}
pMsg
->
pTable
=
(
STable
Info
*
)
mgmtDoCreateChildTable
(
pCreate
,
pVgroup
,
sid
);
pMsg
->
pTable
=
(
STable
Obj
*
)
mgmtDoCreateChildTable
(
pCreate
,
pVgroup
,
sid
);
if
(
pMsg
->
pTable
==
NULL
)
{
mgmtSendSimpleResp
(
pMsg
->
thandle
,
terrno
);
return
;
...
...
@@ -1641,7 +1641,7 @@ static SChildTableObj* mgmtGetTableByPos(uint32_t dnodeId, int32_t vnode, int32_
}
SChildTableObj
*
pTable
=
pVgroup
->
tableList
[
sid
];
mgmtIncTableRef
((
STable
Info
*
)
pTable
);
mgmtIncTableRef
((
STable
Obj
*
)
pTable
);
mgmtDecVgroupRef
(
pVgroup
);
return
pTable
;
}
...
...
src/mnode/src/mgmtVgroup.c
浏览文件 @
76d4cdf8
...
...
@@ -301,7 +301,7 @@ int32_t mgmtGetVgroupMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
int32_t
maxReplica
=
0
;
SVgObj
*
pVgroup
=
NULL
;
STable
Info
*
pTable
=
NULL
;
STable
Obj
*
pTable
=
NULL
;
if
(
pShow
->
payloadLen
>
0
)
{
pTable
=
mgmtGetTable
(
pShow
->
payload
);
if
(
NULL
==
pTable
||
pTable
->
type
==
TSDB_SUPER_TABLE
)
{
...
...
src/vnode/main/inc/vnodeInt.h
浏览文件 @
76d4cdf8
...
...
@@ -20,6 +20,9 @@
extern
"C"
{
#endif
#include "tsync.h"
#include "twal.h"
typedef
enum
_VN_STATUS
{
VN_STATUS_INIT
,
VN_STATUS_CREATING
,
...
...
@@ -41,8 +44,9 @@ typedef struct {
void
*
sync
;
void
*
events
;
void
*
cq
;
// continuous query
int32_t
replicas
;
SVnodeDesc
vpeers
[
TSDB_MAX_MPEERS
];
STsdbCfg
tsdbCfg
;
SSyncCfg
syncCfg
;
SWalCfg
walCfg
;
}
SVnodeObj
;
int
vnodeWriteToQueue
(
void
*
param
,
SWalHead
*
pHead
,
int
type
);
...
...
src/vnode/main/src/vnodeMain.c
浏览文件 @
76d4cdf8
...
...
@@ -60,16 +60,6 @@ int32_t vnodeCreate(SMDCreateVnodeMsg *pVnodeCfg) {
return
TSDB_CODE_SUCCESS
;
}
STsdbCfg
tsdbCfg
=
{
0
};
tsdbCfg
.
precision
=
pVnodeCfg
->
cfg
.
precision
;
tsdbCfg
.
tsdbId
=
pVnodeCfg
->
cfg
.
vgId
;
tsdbCfg
.
maxTables
=
pVnodeCfg
->
cfg
.
maxSessions
;
tsdbCfg
.
daysPerFile
=
pVnodeCfg
->
cfg
.
daysPerFile
;
tsdbCfg
.
minRowsPerFileBlock
=
-
1
;
tsdbCfg
.
maxRowsPerFileBlock
=
-
1
;
tsdbCfg
.
keep
=
-
1
;
tsdbCfg
.
maxCacheSize
=
-
1
;
char
rootDir
[
TSDB_FILENAME_LEN
]
=
{
0
};
sprintf
(
rootDir
,
"%s/vnode%d"
,
tsVnodeDir
,
pVnodeCfg
->
cfg
.
vgId
);
if
(
mkdir
(
rootDir
,
0755
)
!=
0
)
{
...
...
@@ -89,6 +79,16 @@ int32_t vnodeCreate(SMDCreateVnodeMsg *pVnodeCfg) {
return
code
;
}
STsdbCfg
tsdbCfg
=
{
0
};
tsdbCfg
.
precision
=
pVnodeCfg
->
cfg
.
precision
;
tsdbCfg
.
tsdbId
=
pVnodeCfg
->
cfg
.
vgId
;
tsdbCfg
.
maxTables
=
pVnodeCfg
->
cfg
.
maxSessions
;
tsdbCfg
.
daysPerFile
=
pVnodeCfg
->
cfg
.
daysPerFile
;
tsdbCfg
.
minRowsPerFileBlock
=
-
1
;
tsdbCfg
.
maxRowsPerFileBlock
=
-
1
;
tsdbCfg
.
keep
=
-
1
;
tsdbCfg
.
maxCacheSize
=
-
1
;
char
tsdbDir
[
TSDB_FILENAME_LEN
]
=
{
0
};
sprintf
(
tsdbDir
,
"%s/vnode%d/tsdb"
,
tsVnodeDir
,
pVnodeCfg
->
cfg
.
vgId
);
code
=
tsdbCreateRepo
(
tsdbDir
,
&
tsdbCfg
,
NULL
);
...
...
@@ -140,7 +140,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
pVnode
->
rqueue
=
dnodeAllocateRqueue
(
pVnode
);
sprintf
(
temp
,
"%s/wal"
,
rootDir
);
pVnode
->
wal
=
walOpen
(
temp
,
3
,
tsC
ommitLog
);
pVnode
->
wal
=
walOpen
(
temp
,
pVnode
->
walCfg
.
wals
,
pVnode
->
walCfg
.
c
ommitLog
);
pVnode
->
sync
=
NULL
;
pVnode
->
events
=
NULL
;
pVnode
->
cq
=
NULL
;
...
...
@@ -293,9 +293,13 @@ static int32_t vnodeSaveCfg(SMDCreateVnodeMsg *pVnodeCfg) {
FILE
*
fp
=
fopen
(
cfgFile
,
"w"
);
if
(
!
fp
)
return
errno
;
fprintf
(
fp
,
"replicas %d
\n
"
,
pVnodeCfg
->
cfg
.
replications
);
fprintf
(
fp
,
"commitLog %d
\n
"
,
pVnodeCfg
->
cfg
.
commitLog
);
fprintf
(
fp
,
"wals %d
\n
"
,
3
);
fprintf
(
fp
,
"arbitratorIp %d
\n
"
,
pVnodeCfg
->
vpeerDesc
[
0
].
ip
);
fprintf
(
fp
,
"quorum %d
\n
"
,
1
);
fprintf
(
fp
,
"replica %d
\n
"
,
pVnodeCfg
->
cfg
.
replications
);
for
(
int32_t
i
=
0
;
i
<
pVnodeCfg
->
cfg
.
replications
;
i
++
)
{
fprintf
(
fp
,
"index%d
dnode %d ip %u
\n
"
,
i
,
pVnodeCfg
->
vpeerDesc
[
i
].
dnodeId
,
pVnodeCfg
->
vpeerDesc
[
i
].
ip
);
fprintf
(
fp
,
"index%d
nodeId %d nodeIp %u name n%d
\n
"
,
i
,
pVnodeCfg
->
vpeerDesc
[
i
].
dnodeId
,
pVnodeCfg
->
vpeerDesc
[
i
].
ip
,
pVnodeCfg
->
vpeerDesc
[
i
].
dnodeId
);
}
fclose
(
fp
);
...
...
@@ -306,37 +310,64 @@ static int32_t vnodeSaveCfg(SMDCreateVnodeMsg *pVnodeCfg) {
// TODO: this is a simple implement
static
int32_t
vnodeReadCfg
(
SVnodeObj
*
pVnode
)
{
char
option
[
3
][
16
]
=
{
0
};
char
cfgFile
[
TSDB_FILENAME_LEN
*
2
]
=
{
0
};
sprintf
(
cfgFile
,
"%s/vnode%d/config"
,
tsVnodeDir
,
pVnode
->
vgId
);
FILE
*
fp
=
fopen
(
cfgFile
,
"r"
);
if
(
!
fp
)
return
errno
;
char
option
[
3
][
32
]
=
{
0
};
int32_t
replicas
=
0
;
int32_t
num
=
fscanf
(
fp
,
"%s %d"
,
option
[
0
],
&
replicas
);
int32_t
commitLog
=
-
1
;
int32_t
num
=
fscanf
(
fp
,
"%s %d"
,
option
[
0
],
&
commitLog
);
if
(
num
!=
2
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
strcmp
(
option
[
0
],
"commitLog"
)
!=
0
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
commitLog
==
-
1
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
pVnode
->
walCfg
.
commitLog
=
(
int8_t
)
commitLog
;
int32_t
wals
=
-
1
;
num
=
fscanf
(
fp
,
"%s %d"
,
option
[
0
],
&
wals
);
if
(
num
!=
2
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
strcmp
(
option
[
0
],
"replicas"
)
!=
0
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
replicas
==
0
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
pVnode
->
replicas
=
replicas
;
for
(
int32_t
i
=
0
;
i
<
replicas
;
++
i
)
{
int32_t
dnodeId
=
0
;
uint32_t
dnodeIp
=
0
;
num
=
fscanf
(
fp
,
"%s %s %d %s %u"
,
option
[
0
],
option
[
1
],
&
dnodeId
,
option
[
2
],
&
dnodeIp
);
if
(
num
!=
5
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
strcmp
(
option
[
1
],
"dnode"
)
!=
0
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
strcmp
(
option
[
2
],
"ip"
)
!=
0
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
dnodeId
==
0
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
dnodeIp
==
0
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
pVnode
->
vpeers
[
i
].
dnodeId
=
dnodeId
;
pVnode
->
vpeers
[
i
].
ip
=
dnodeIp
;
pVnode
->
vpeers
[
i
].
vgId
=
pVnode
->
vgId
;
if
(
strcmp
(
option
[
0
],
"wals"
)
!=
0
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
wals
==
-
1
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
pVnode
->
walCfg
.
wals
=
(
int8_t
)
wals
;
int32_t
arbitratorIp
=
-
1
;
num
=
fscanf
(
fp
,
"%s %u"
,
option
[
0
],
&
arbitratorIp
);
if
(
num
!=
2
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
strcmp
(
option
[
0
],
"arbitratorIp"
)
!=
0
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
arbitratorIp
==
-
1
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
pVnode
->
syncCfg
.
arbitratorIp
=
arbitratorIp
;
int32_t
quorum
=
-
1
;
num
=
fscanf
(
fp
,
"%s %d"
,
option
[
0
],
&
quorum
);
if
(
num
!=
2
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
strcmp
(
option
[
0
],
"quorum"
)
!=
0
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
quorum
==
-
1
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
pVnode
->
syncCfg
.
quorum
=
(
int8_t
)
quorum
;
int32_t
replica
=
-
1
;
num
=
fscanf
(
fp
,
"%s %d"
,
option
[
0
],
&
replica
);
if
(
num
!=
2
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
strcmp
(
option
[
0
],
"replica"
)
!=
0
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
replica
==
-
1
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
pVnode
->
syncCfg
.
replica
=
(
int8_t
)
replica
;
for
(
int32_t
i
=
0
;
i
<
replica
;
++
i
)
{
int32_t
dnodeId
=
-
1
;
uint32_t
dnodeIp
=
-
1
;
num
=
fscanf
(
fp
,
"%s %s %d %s %u %s %s"
,
option
[
0
],
option
[
1
],
&
dnodeId
,
option
[
2
],
&
dnodeIp
,
option
[
3
],
pVnode
->
syncCfg
.
nodeInfo
[
i
].
name
);
if
(
num
!=
7
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
strcmp
(
option
[
1
],
"nodeId"
)
!=
0
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
strcmp
(
option
[
2
],
"nodeIp"
)
!=
0
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
strcmp
(
option
[
3
],
"name"
)
!=
0
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
dnodeId
==
-
1
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
if
(
dnodeIp
==
-
1
)
return
TSDB_CODE_INVALID_FILE_FORMAT
;
pVnode
->
syncCfg
.
nodeInfo
[
i
].
nodeId
=
dnodeId
;
pVnode
->
syncCfg
.
nodeInfo
[
i
].
nodeIp
=
dnodeIp
;
}
fclose
(
fp
);
dTrace
(
"pVnode:%p vgId:%d, read vnode cfg successed"
,
pVnode
,
pVnode
->
vgId
);
return
TSDB_CODE_SUCCESS
;
}
\ No newline at end of file
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录