Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
f88e536b
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1184
Star
22015
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
f88e536b
编写于
5月 15, 2022
作者:
S
Shengliang Guan
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
refactor: adjust dnode logs
上级
21a9367d
变更
6
隐藏空白更改
内联
并排
Showing
6 changed file
with
39 addition
and
35 deletion
+39
-35
source/dnode/mgmt/exe/dmMain.c
source/dnode/mgmt/exe/dmMain.c
+2
-2
source/dnode/mgmt/mgmt_dnode/src/dmInt.c
source/dnode/mgmt/mgmt_dnode/src/dmInt.c
+0
-4
source/dnode/mgmt/node_mgmt/src/dmMgmt.c
source/dnode/mgmt/node_mgmt/src/dmMgmt.c
+6
-6
source/dnode/mgmt/node_mgmt/src/dmNodes.c
source/dnode/mgmt/node_mgmt/src/dmNodes.c
+12
-4
source/dnode/mgmt/node_mgmt/src/dmProc.c
source/dnode/mgmt/node_mgmt/src/dmProc.c
+16
-16
source/dnode/mgmt/node_mgmt/src/dmTransport.c
source/dnode/mgmt/node_mgmt/src/dmTransport.c
+3
-3
未找到文件。
source/dnode/mgmt/exe/dmMain.c
浏览文件 @
f88e536b
...
...
@@ -191,9 +191,9 @@ static int32_t dmRunDnode() {
dmSetSignalHandle
();
}
dInfo
(
"start t
he servic
e"
);
dInfo
(
"start t
o run dnod
e"
);
int32_t
code
=
dmRun
(
pDnode
);
dInfo
(
"s
tart s
hutting down the service"
);
dInfo
(
"shutting down the service"
);
global
.
pDnode
=
NULL
;
dmClose
(
pDnode
);
...
...
source/dnode/mgmt/mgmt_dnode/src/dmInt.c
浏览文件 @
f88e536b
...
...
@@ -33,7 +33,6 @@ static void dmStopMgmt(SDnodeMgmt *pMgmt) {
}
static
int32_t
dmOpenMgmt
(
SMgmtInputOpt
*
pInput
,
SMgmtOutputOpt
*
pOutput
)
{
dInfo
(
"dnode-mgmt start to init"
);
SDnodeMgmt
*
pMgmt
=
taosMemoryCalloc
(
1
,
sizeof
(
SDnodeMgmt
));
if
(
pMgmt
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
...
...
@@ -58,15 +57,12 @@ static int32_t dmOpenMgmt(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
}
pOutput
->
pMgmt
=
pMgmt
;
dInfo
(
"dnode-mgmt is initialized"
);
return
0
;
}
static
void
dmCloseMgmt
(
SDnodeMgmt
*
pMgmt
)
{
dInfo
(
"dnode-mgmt start to clean up"
);
dmStopWorker
(
pMgmt
);
taosMemoryFree
(
pMgmt
);
dInfo
(
"dnode-mgmt is cleaned up"
);
}
static
int32_t
dmRequireMgmt
(
const
SMgmtInputOpt
*
pInput
,
bool
*
required
)
{
...
...
source/dnode/mgmt/node_mgmt/src/dmMgmt.c
浏览文件 @
f88e536b
...
...
@@ -212,7 +212,7 @@ SDnode *dmCreate(const SDnodeOpt *pOption) {
}
dmReportStartup
(
pDnode
,
"dnode-transport"
,
"initialized"
);
dInfo
(
"dnode is created,
data
:%p"
,
pDnode
);
dInfo
(
"dnode is created,
ptr
:%p"
,
pDnode
);
code
=
0
;
_OVER:
...
...
@@ -231,7 +231,7 @@ void dmClose(SDnode *pDnode) {
dmCleanupClient
(
pDnode
);
dmCleanupServer
(
pDnode
);
dmClearVars
(
pDnode
);
dInfo
(
"dnode is closed,
data
:%p"
,
pDnode
);
dInfo
(
"dnode is closed,
ptr
:%p"
,
pDnode
);
}
void
dmSetStatus
(
SDnode
*
pDnode
,
EDndRunStatus
status
)
{
...
...
@@ -254,7 +254,7 @@ SMgmtWrapper *dmAcquireWrapper(SDnode *pDnode, EDndNodeType ntype) {
taosRLockLatch
(
&
pWrapper
->
latch
);
if
(
pWrapper
->
deployed
)
{
int32_t
refCount
=
atomic_add_fetch_32
(
&
pWrapper
->
refCount
,
1
);
dTrace
(
"node:%s, is acquired, ref
Count
:%d"
,
pWrapper
->
name
,
refCount
);
dTrace
(
"node:%s, is acquired, ref:%d"
,
pWrapper
->
name
,
refCount
);
}
else
{
terrno
=
TSDB_CODE_NODE_NOT_DEPLOYED
;
pRetWrapper
=
NULL
;
...
...
@@ -270,7 +270,7 @@ int32_t dmMarkWrapper(SMgmtWrapper *pWrapper) {
taosRLockLatch
(
&
pWrapper
->
latch
);
if
(
pWrapper
->
deployed
||
(
InParentProc
(
pWrapper
->
proc
.
ptype
)
&&
pWrapper
->
required
))
{
int32_t
refCount
=
atomic_add_fetch_32
(
&
pWrapper
->
refCount
,
1
);
dTrace
(
"node:%s, is marked, ref
Count
:%d"
,
pWrapper
->
name
,
refCount
);
dTrace
(
"node:%s, is marked, ref:%d"
,
pWrapper
->
name
,
refCount
);
}
else
{
terrno
=
TSDB_CODE_NODE_NOT_DEPLOYED
;
code
=
-
1
;
...
...
@@ -286,14 +286,14 @@ void dmReleaseWrapper(SMgmtWrapper *pWrapper) {
taosRLockLatch
(
&
pWrapper
->
latch
);
int32_t
refCount
=
atomic_sub_fetch_32
(
&
pWrapper
->
refCount
,
1
);
taosRUnLockLatch
(
&
pWrapper
->
latch
);
dTrace
(
"node:%s, is released, ref
Count
:%d"
,
pWrapper
->
name
,
refCount
);
dTrace
(
"node:%s, is released, ref:%d"
,
pWrapper
->
name
,
refCount
);
}
void
dmReportStartup
(
SDnode
*
pDnode
,
const
char
*
pName
,
const
char
*
pDesc
)
{
SStartupInfo
*
pStartup
=
&
pDnode
->
startup
;
tstrncpy
(
pStartup
->
name
,
pName
,
TSDB_STEP_NAME_LEN
);
tstrncpy
(
pStartup
->
desc
,
pDesc
,
TSDB_STEP_DESC_LEN
);
d
Info
(
"step:%s, %s"
,
pStartup
->
name
,
pStartup
->
desc
);
d
Debug
(
"step:%s, %s"
,
pStartup
->
name
,
pStartup
->
desc
);
}
void
dmReportStartupByWrapper
(
SMgmtWrapper
*
pWrapper
,
const
char
*
pName
,
const
char
*
pDesc
)
{
...
...
source/dnode/mgmt/node_mgmt/src/dmNodes.c
浏览文件 @
f88e536b
...
...
@@ -99,15 +99,17 @@ int32_t dmOpenNode(SMgmtWrapper *pWrapper) {
}
if
(
OnlyInSingleProc
(
pWrapper
->
proc
.
ptype
))
{
dInfo
(
"node:%s, start to open"
,
pWrapper
->
name
);
if
((
*
pWrapper
->
func
.
openFp
)(
&
input
,
&
output
)
!=
0
)
{
dError
(
"node:%s, failed to open since %s"
,
pWrapper
->
name
,
terrstr
());
return
-
1
;
}
d
Debug
(
"node:%s, has been opened"
,
pWrapper
->
name
);
d
Info
(
"node:%s, has been opened"
,
pWrapper
->
name
);
pWrapper
->
deployed
=
true
;
}
if
(
InChildProc
(
pWrapper
->
proc
.
ptype
))
{
dDebug
(
"node:%s, start to open"
,
pWrapper
->
name
);
if
((
*
pWrapper
->
func
.
openFp
)(
&
input
,
&
output
)
!=
0
)
{
dError
(
"node:%s, failed to open since %s"
,
pWrapper
->
name
,
terrstr
());
return
-
1
;
...
...
@@ -123,6 +125,7 @@ int32_t dmOpenNode(SMgmtWrapper *pWrapper) {
}
if
(
InParentProc
(
pWrapper
->
proc
.
ptype
))
{
dDebug
(
"node:%s, start to open"
,
pWrapper
->
name
);
if
(
dmInitParentProc
(
pWrapper
)
!=
0
)
{
return
-
1
;
}
...
...
@@ -145,9 +148,13 @@ int32_t dmOpenNode(SMgmtWrapper *pWrapper) {
int32_t
dmStartNode
(
SMgmtWrapper
*
pWrapper
)
{
if
(
OnlyInParentProc
(
pWrapper
->
proc
.
ptype
))
return
0
;
if
(
pWrapper
->
func
.
startFp
!=
NULL
&&
(
*
pWrapper
->
func
.
startFp
)(
pWrapper
->
pMgmt
)
!=
0
)
{
dError
(
"node:%s, failed to start since %s"
,
pWrapper
->
name
,
terrstr
());
return
-
1
;
if
(
pWrapper
->
func
.
startFp
!=
NULL
)
{
dDebug
(
"node:%s, start to start"
,
pWrapper
->
name
);
if
((
*
pWrapper
->
func
.
startFp
)(
pWrapper
->
pMgmt
)
!=
0
)
{
dError
(
"node:%s, failed to start since %s"
,
pWrapper
->
name
,
terrstr
());
return
-
1
;
}
dDebug
(
"node:%s, has been started"
,
pWrapper
->
name
);
}
dmReportStartup
(
pWrapper
->
pDnode
,
pWrapper
->
name
,
"started"
);
...
...
@@ -156,6 +163,7 @@ int32_t dmStartNode(SMgmtWrapper *pWrapper) {
void
dmStopNode
(
SMgmtWrapper
*
pWrapper
)
{
if
(
pWrapper
->
func
.
stopFp
!=
NULL
&&
pWrapper
->
pMgmt
!=
NULL
)
{
dDebug
(
"node:%s, start to stop"
,
pWrapper
->
name
);
(
*
pWrapper
->
func
.
stopFp
)(
pWrapper
->
pMgmt
);
dDebug
(
"node:%s, has been stopped"
,
pWrapper
->
name
);
}
...
...
source/dnode/mgmt/node_mgmt/src/dmProc.c
浏览文件 @
f88e536b
...
...
@@ -185,8 +185,8 @@ static int32_t dmPushToProcQueue(SProc *proc, SProcQueue *queue, const char *pHe
taosThreadMutexUnlock
(
&
queue
->
mutex
);
tsem_post
(
&
queue
->
sem
);
dTrace
(
"node:%s, push
proc msg at pos:%d ftype:%d remain:%d handle:%p ref:%"
PRId64
", head:%d %p body:%d %p"
,
queue
->
name
,
pos
,
ftype
,
queue
->
items
,
(
void
*
)
handle
,
handleRef
,
headLen
,
pHead
,
bodyLen
,
pBody
);
dTrace
(
"node:%s, push
msg:%p:%d cont:%p%d handle:%p, ftype:%d pos:%d remain:%d"
,
queue
->
name
,
pHead
,
headLen
,
pBody
,
bodyLen
,
(
void
*
)
handle
,
ftype
,
pos
,
queue
->
items
);
return
0
;
}
...
...
@@ -315,31 +315,31 @@ static void *dmConsumChildQueue(void *param) {
EProcFuncType
ftype
=
PROC_FUNC_REQ
;
SNodeMsg
*
pReq
=
NULL
;
dDebug
(
"node:%s, start to consume from c
hild
queue"
,
proc
->
name
);
dDebug
(
"node:%s, start to consume from cqueue"
,
proc
->
name
);
do
{
numOfMsgs
=
dmPopFromProcQueue
(
queue
,
&
pHead
,
&
headLen
,
&
pBody
,
&
bodyLen
,
&
ftype
);
if
(
numOfMsgs
==
0
)
{
dDebug
(
"node:%s, get no msg from c
hild q
ueue and exit thread"
,
proc
->
name
);
dDebug
(
"node:%s, get no msg from cueue and exit thread"
,
proc
->
name
);
break
;
}
if
(
numOfMsgs
<
0
)
{
dError
(
"node:%s, get no msg from c
hild
queue since %s"
,
proc
->
name
,
terrstr
());
dError
(
"node:%s, get no msg from cqueue since %s"
,
proc
->
name
,
terrstr
());
taosMsleep
(
1
);
continue
;
}
if
(
ftype
!=
PROC_FUNC_REQ
)
{
dFatal
(
"node:%s, msg:%p from c
hild
queue, invalid ftype:%d"
,
proc
->
name
,
pHead
,
ftype
);
dFatal
(
"node:%s, msg:%p from cqueue, invalid ftype:%d"
,
proc
->
name
,
pHead
,
ftype
);
taosFreeQitem
(
pHead
);
rpcFreeCont
(
pBody
);
}
else
{
dTrace
(
"node:%s, msg:%p from c
hild q
ueue"
,
proc
->
name
,
pHead
);
dTrace
(
"node:%s, msg:%p from cueue"
,
proc
->
name
,
pHead
);
pReq
=
pHead
;
pReq
->
rpcMsg
.
pCont
=
pBody
;
code
=
dmProcessNodeMsg
(
pWrapper
,
pReq
);
if
(
code
!=
0
)
{
dError
(
"node:%s, failed to process msg:%p since %s, put into p
arent
queue"
,
proc
->
name
,
pReq
,
terrstr
());
dError
(
"node:%s, failed to process msg:%p since %s, put into pqueue"
,
proc
->
name
,
pReq
,
terrstr
());
SRpcMsg
rspMsg
=
{
.
handle
=
pReq
->
rpcMsg
.
handle
,
.
ahandle
=
pReq
->
rpcMsg
.
ahandle
,
...
...
@@ -371,16 +371,16 @@ static void *dmConsumParentQueue(void *param) {
EProcFuncType
ftype
=
PROC_FUNC_REQ
;
SRpcMsg
*
pRsp
=
NULL
;
dDebug
(
"node:%s, start to consume from p
arent
queue"
,
proc
->
name
);
dDebug
(
"node:%s, start to consume from pqueue"
,
proc
->
name
);
do
{
numOfMsgs
=
dmPopFromProcQueue
(
queue
,
&
pHead
,
&
headLen
,
&
pBody
,
&
bodyLen
,
&
ftype
);
if
(
numOfMsgs
==
0
)
{
dDebug
(
"node:%s, get no msg from p
arent
queue and exit thread"
,
proc
->
name
);
dDebug
(
"node:%s, get no msg from pqueue and exit thread"
,
proc
->
name
);
break
;
}
if
(
numOfMsgs
<
0
)
{
dError
(
"node:%s, get no msg from p
arent
queue since %s"
,
proc
->
name
,
terrstr
());
dError
(
"node:%s, get no msg from pqueue since %s"
,
proc
->
name
,
terrstr
());
taosMsleep
(
1
);
continue
;
}
...
...
@@ -388,22 +388,22 @@ static void *dmConsumParentQueue(void *param) {
if
(
ftype
==
PROC_FUNC_RSP
)
{
pRsp
=
pHead
;
pRsp
->
pCont
=
pBody
;
dTrace
(
"node:%s, rsp msg:%p from p
arent
queue, code:0x%04x handle:%p"
,
proc
->
name
,
pRsp
,
code
,
pRsp
->
handle
);
dTrace
(
"node:%s, rsp msg:%p from pqueue, code:0x%04x handle:%p"
,
proc
->
name
,
pRsp
,
code
,
pRsp
->
handle
);
dmRemoveProcRpcHandle
(
proc
,
pRsp
->
handle
);
rpcSendResponse
(
pRsp
);
}
else
if
(
ftype
==
PROC_FUNC_REGIST
)
{
pRsp
=
pHead
;
dTrace
(
"node:%s, regist msg:%p from p
arent
queue, code:0x%04x handle:%p"
,
proc
->
name
,
pRsp
,
code
,
pRsp
->
handle
);
dTrace
(
"node:%s, regist msg:%p from pqueue, code:0x%04x handle:%p"
,
proc
->
name
,
pRsp
,
code
,
pRsp
->
handle
);
rpcRegisterBrokenLinkArg
(
pRsp
);
rpcFreeCont
(
pBody
);
}
else
if
(
ftype
==
PROC_FUNC_RELEASE
)
{
pRsp
=
pHead
;
dTrace
(
"node:%s, release msg:%p from p
arent
queue, code:0x%04x handle:%p"
,
proc
->
name
,
pRsp
,
code
,
pRsp
->
handle
);
dTrace
(
"node:%s, release msg:%p from pqueue, code:0x%04x handle:%p"
,
proc
->
name
,
pRsp
,
code
,
pRsp
->
handle
);
dmRemoveProcRpcHandle
(
proc
,
pRsp
->
handle
);
rpcReleaseHandle
(
pRsp
->
handle
,
(
int8_t
)
pRsp
->
code
);
rpcFreeCont
(
pBody
);
}
else
{
dFatal
(
"node:%s, msg:%p get from p
arent
queue, invalid ftype:%d"
,
proc
->
name
,
pHead
,
ftype
);
dFatal
(
"node:%s, msg:%p get from pqueue, invalid ftype:%d"
,
proc
->
name
,
pHead
,
ftype
);
rpcFreeCont
(
pBody
);
}
...
...
@@ -502,7 +502,7 @@ void dmPutToProcPQueue(SProc *proc, const void *pHead, int16_t headLen, const vo
EProcFuncType
ftype
)
{
int32_t
retry
=
0
;
while
(
dmPushToProcQueue
(
proc
,
proc
->
pqueue
,
pHead
,
headLen
,
pBody
,
bodyLen
,
0
,
0
,
ftype
)
!=
0
)
{
dWarn
(
"node:%s, failed to put msg:%p to p
arent
queue since %s, retry:%d"
,
proc
->
name
,
pHead
,
terrstr
(),
retry
);
dWarn
(
"node:%s, failed to put msg:%p to pqueue since %s, retry:%d"
,
proc
->
name
,
pHead
,
terrstr
(),
retry
);
retry
++
;
taosMsleep
(
retry
);
}
...
...
source/dnode/mgmt/node_mgmt/src/dmTransport.c
浏览文件 @
f88e536b
...
...
@@ -121,13 +121,13 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) {
}
}
dTrace
(
"msg:%s is received, handle:%p app:%p"
,
TMSG_INFO
(
msgType
),
pRpc
->
handle
,
pRpc
->
ahandle
);
if
(
dmMarkWrapper
(
pWrapper
)
!=
0
)
{
goto
_OVER
;
}
else
{
needRelease
=
true
;
}
dTrace
(
"msg:%s is received, handle:%p app:%p"
,
TMSG_INFO
(
msgType
),
pRpc
->
handle
,
pRpc
->
ahandle
);
pMsg
=
taosAllocateQitem
(
sizeof
(
SNodeMsg
),
RPC_QITEM
);
if
(
pMsg
==
NULL
)
{
goto
_OVER
;
...
...
@@ -138,7 +138,7 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) {
}
if
(
InParentProc
(
pWrapper
->
proc
.
ptype
))
{
dTrace
(
"msg:%p, put into c
hild queue, handle:%p"
,
pMsg
,
pRpc
->
handle
);
dTrace
(
"msg:%p, put into c
queue, handle:%p ref:%"
PRId64
,
pMsg
,
pRpc
->
handle
,
pRpc
->
refId
);
code
=
dmPutToProcCQueue
(
&
pWrapper
->
proc
,
pMsg
,
sizeof
(
SNodeMsg
),
pRpc
->
pCont
,
pRpc
->
contLen
,
(
isReq
&&
(
pRpc
->
code
==
0
))
?
pRpc
->
handle
:
NULL
,
pRpc
->
refId
,
PROC_FUNC_REQ
);
}
else
{
...
...
@@ -148,7 +148,7 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) {
_OVER:
if
(
code
==
0
)
{
if
(
pWrapper
!=
NULL
&&
InParentProc
(
pWrapper
->
proc
.
ptype
))
{
dTrace
(
"msg:%p,
freed in parent process
"
,
pMsg
);
dTrace
(
"msg:%p,
is freed after push to cqueue
"
,
pMsg
);
taosFreeQitem
(
pMsg
);
rpcFreeCont
(
pRpc
->
pCont
);
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录