Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
a4166b73
T
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
未验证
提交
a4166b73
编写于
6月 05, 2022
作者:
S
Shengliang Guan
提交者:
GitHub
6月 05, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #13464 from taosdata/fix/mnode
refactor: add null transaction action typ
上级
9b348a4a
41a1b0fc
变更
5
隐藏空白更改
内联
并排
Showing
5 changed file
with
73 addition
and
52 deletion
+73
-52
include/common/tmsgdef.h
include/common/tmsgdef.h
+1
-2
source/dnode/mgmt/mgmt_mnode/src/mmHandle.c
source/dnode/mgmt/mgmt_mnode/src/mmHandle.c
+0
-1
source/dnode/mnode/impl/inc/mndTrans.h
source/dnode/mnode/impl/inc/mndTrans.h
+9
-2
source/dnode/mnode/impl/src/mndMnode.c
source/dnode/mnode/impl/src/mndMnode.c
+5
-22
source/dnode/mnode/impl/src/mndTrans.c
source/dnode/mnode/impl/src/mndTrans.c
+58
-25
未找到文件。
include/common/tmsgdef.h
浏览文件 @
a4166b73
...
...
@@ -152,8 +152,7 @@ enum {
TD_DEF_MSG_TYPE
(
TDMT_MND_SYSTABLE_RETRIEVE
,
"retrieve"
,
NULL
,
NULL
)
TD_DEF_MSG_TYPE
(
TDMT_MND_GRANT
,
"grant"
,
NULL
,
NULL
)
TD_DEF_MSG_TYPE
(
TDMT_MND_AUTH
,
"auth"
,
NULL
,
NULL
)
TD_DEF_MSG_TYPE
(
TDMT_MND_CONFIRM_WRITE
,
"mnode-confirm-write"
,
NULL
,
NULL
)
TD_DEF_MSG_TYPE
(
TDMT_MND_APPLY_MSG
,
"mnode-apply-msg"
,
NULL
,
NULL
)
TD_DEF_MSG_TYPE
(
TDMT_MND_APPLY_MSG
,
"mnode-apply"
,
NULL
,
NULL
)
TD_NEW_MSG_SEG
(
TDMT_VND_MSG
)
TD_DEF_MSG_TYPE
(
TDMT_VND_SUBMIT
,
"submit"
,
SSubmitReq
,
SSubmitRsp
)
...
...
source/dnode/mgmt/mgmt_mnode/src/mmHandle.c
浏览文件 @
a4166b73
...
...
@@ -197,7 +197,6 @@ SArray *mmGetMsgHandles() {
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_MND_SYSTABLE_RETRIEVE
,
mmPutNodeMsgToReadQueue
,
0
)
==
NULL
)
goto
_OVER
;
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_MND_GRANT
,
mmPutNodeMsgToWriteQueue
,
0
)
==
NULL
)
goto
_OVER
;
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_MND_AUTH
,
mmPutNodeMsgToReadQueue
,
0
)
==
NULL
)
goto
_OVER
;
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_MND_CONFIRM_WRITE
,
mmPutNodeMsgToWriteQueue
,
0
)
==
NULL
)
goto
_OVER
;
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_VND_QUERY
,
mmPutNodeMsgToQueryQueue
,
1
)
==
NULL
)
goto
_OVER
;
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_VND_QUERY_CONTINUE
,
mmPutNodeMsgToQueryQueue
,
1
)
==
NULL
)
goto
_OVER
;
...
...
source/dnode/mnode/impl/inc/mndTrans.h
浏览文件 @
a4166b73
...
...
@@ -29,12 +29,18 @@ typedef enum {
TRANS_STOP_FUNC_MQ_REB
=
4
,
}
ETrnFunc
;
typedef
enum
{
TRANS_ACTION_NULL
=
0
,
TRANS_ACTION_MSG
=
1
,
TRANS_ACTION_RAW
=
2
,
}
ETrnAct
;
typedef
struct
{
int32_t
id
;
int32_t
errCode
;
int32_t
acceptableCode
;
int8_t
stage
;
int8_t
actionType
;
// 0-msg, 1-raw
ETrnAct
actionType
;
int8_t
rawWritten
;
int8_t
msgSent
;
int8_t
msgReceived
;
...
...
@@ -57,6 +63,7 @@ void mndTransDrop(STrans *pTrans);
int32_t
mndTransAppendRedolog
(
STrans
*
pTrans
,
SSdbRaw
*
pRaw
);
int32_t
mndTransAppendUndolog
(
STrans
*
pTrans
,
SSdbRaw
*
pRaw
);
int32_t
mndTransAppendCommitlog
(
STrans
*
pTrans
,
SSdbRaw
*
pRaw
);
int32_t
mndTransAppendNullLog
(
STrans
*
pTrans
);
int32_t
mndTransAppendRedoAction
(
STrans
*
pTrans
,
STransAction
*
pAction
);
int32_t
mndTransAppendUndoAction
(
STrans
*
pTrans
,
STransAction
*
pAction
);
void
mndTransSetRpcRsp
(
STrans
*
pTrans
,
void
*
pCont
,
int32_t
contLen
);
...
...
@@ -65,7 +72,7 @@ void mndTransSetDbName(STrans *pTrans, const char *dbname);
void
mndTransSetSerial
(
STrans
*
pTrans
);
int32_t
mndTransPrepare
(
SMnode
*
pMnode
,
STrans
*
pTrans
);
void
mndTransProcessRsp
(
SRpcMsg
*
pRsp
);
int32_t
mndTransProcessRsp
(
SRpcMsg
*
pRsp
);
void
mndTransPullup
(
SMnode
*
pMnode
);
int32_t
mndKillTrans
(
SMnode
*
pMnode
,
STrans
*
pTrans
);
...
...
source/dnode/mnode/impl/src/mndMnode.c
浏览文件 @
a4166b73
...
...
@@ -34,9 +34,6 @@ static int32_t mndMnodeActionUpdate(SSdb *pSdb, SMnodeObj *pOld, SMnodeObj *pNe
static
int32_t
mndProcessCreateMnodeReq
(
SRpcMsg
*
pReq
);
static
int32_t
mndProcessAlterMnodeReq
(
SRpcMsg
*
pReq
);
static
int32_t
mndProcessDropMnodeReq
(
SRpcMsg
*
pReq
);
static
int32_t
mndProcessCreateMnodeRsp
(
SRpcMsg
*
pRsp
);
static
int32_t
mndProcessAlterMnodeRsp
(
SRpcMsg
*
pRsp
);
static
int32_t
mndProcessDropMnodeRsp
(
SRpcMsg
*
pRsp
);
static
int32_t
mndRetrieveMnodes
(
SRpcMsg
*
pReq
,
SShowObj
*
pShow
,
SSDataBlock
*
pBlock
,
int32_t
rows
);
static
void
mndCancelGetNextMnode
(
SMnode
*
pMnode
,
void
*
pIter
);
...
...
@@ -53,11 +50,11 @@ int32_t mndInitMnode(SMnode *pMnode) {
};
mndSetMsgHandle
(
pMnode
,
TDMT_MND_CREATE_MNODE
,
mndProcessCreateMnodeReq
);
mndSetMsgHandle
(
pMnode
,
TDMT_DND_CREATE_MNODE_RSP
,
mndTransProcessRsp
);
mndSetMsgHandle
(
pMnode
,
TDMT_MND_ALTER_MNODE
,
mndProcessAlterMnodeReq
);
mndSetMsgHandle
(
pMnode
,
TDMT_MND_ALTER_MNODE_RSP
,
mndTransProcessRsp
);
mndSetMsgHandle
(
pMnode
,
TDMT_MND_DROP_MNODE
,
mndProcessDropMnodeReq
);
mndSetMsgHandle
(
pMnode
,
TDMT_DND_CREATE_MNODE_RSP
,
mndProcessCreateMnodeRsp
);
mndSetMsgHandle
(
pMnode
,
TDMT_MND_ALTER_MNODE_RSP
,
mndProcessAlterMnodeRsp
);
mndSetMsgHandle
(
pMnode
,
TDMT_DND_DROP_MNODE_RSP
,
mndProcessDropMnodeRsp
);
mndSetMsgHandle
(
pMnode
,
TDMT_DND_DROP_MNODE_RSP
,
mndTransProcessRsp
);
mndAddShowRetrieveHandle
(
pMnode
,
TSDB_MGMT_TABLE_MNODE
,
mndRetrieveMnodes
);
mndAddShowFreeIterHandle
(
pMnode
,
TSDB_MGMT_TABLE_MNODE
,
mndCancelGetNextMnode
);
...
...
@@ -367,7 +364,7 @@ static int32_t mndCreateMnode(SMnode *pMnode, SRpcMsg *pReq, SDnodeObj *pDnode,
if
(
mndSetCreateMnodeRedoLogs
(
pMnode
,
pTrans
,
&
mnodeObj
)
!=
0
)
goto
_OVER
;
if
(
mndSetCreateMnodeCommitLogs
(
pMnode
,
pTrans
,
&
mnodeObj
)
!=
0
)
goto
_OVER
;
if
(
mndSetCreateMnodeRedoActions
(
pMnode
,
pTrans
,
pDnode
,
&
mnodeObj
)
!=
0
)
goto
_OVER
;
if
(
mndTransAppendNullLog
(
pTrans
)
!=
0
)
goto
_OVER
;
if
(
mndTransPrepare
(
pMnode
,
pTrans
)
!=
0
)
goto
_OVER
;
code
=
0
;
...
...
@@ -549,6 +546,7 @@ static int32_t mndDropMnode(SMnode *pMnode, SRpcMsg *pReq, SMnodeObj *pObj) {
if
(
mndSetDropMnodeRedoLogs
(
pMnode
,
pTrans
,
pObj
)
!=
0
)
goto
_OVER
;
if
(
mndSetDropMnodeCommitLogs
(
pMnode
,
pTrans
,
pObj
)
!=
0
)
goto
_OVER
;
if
(
mndSetDropMnodeRedoActions
(
pMnode
,
pTrans
,
pObj
->
pDnode
,
pObj
)
!=
0
)
goto
_OVER
;
if
(
mndTransAppendNullLog
(
pTrans
)
!=
0
)
goto
_OVER
;
if
(
mndTransPrepare
(
pMnode
,
pTrans
)
!=
0
)
goto
_OVER
;
code
=
0
;
...
...
@@ -616,21 +614,6 @@ _OVER:
return
code
;
}
static
int32_t
mndProcessCreateMnodeRsp
(
SRpcMsg
*
pRsp
)
{
mndTransProcessRsp
(
pRsp
);
return
0
;
}
static
int32_t
mndProcessAlterMnodeRsp
(
SRpcMsg
*
pRsp
)
{
mndTransProcessRsp
(
pRsp
);
return
0
;
}
static
int32_t
mndProcessDropMnodeRsp
(
SRpcMsg
*
pRsp
)
{
mndTransProcessRsp
(
pRsp
);
return
0
;
}
static
int32_t
mndRetrieveMnodes
(
SRpcMsg
*
pReq
,
SShowObj
*
pShow
,
SSDataBlock
*
pBlock
,
int32_t
rows
)
{
SMnode
*
pMnode
=
pReq
->
info
.
node
;
SSdb
*
pSdb
=
pMnode
->
pSdb
;
...
...
source/dnode/mnode/impl/src/mndTrans.c
浏览文件 @
a4166b73
...
...
@@ -88,12 +88,14 @@ static int32_t mndTransGetActionsSize(SArray *pArray) {
for
(
int32_t
i
=
0
;
i
<
actionNum
;
++
i
)
{
STransAction
*
pAction
=
taosArrayGet
(
pArray
,
i
);
if
(
pAction
->
actionType
)
{
if
(
pAction
->
actionType
==
TRANS_ACTION_RAW
)
{
rawDataLen
+=
(
sdbGetRawTotalSize
(
pAction
->
pRaw
)
+
sizeof
(
int32_t
));
}
else
{
}
else
if
(
pAction
->
actionType
==
TRANS_ACTION_MSG
)
{
rawDataLen
+=
(
sizeof
(
STransAction
)
+
pAction
->
contLen
);
}
else
{
// empty
}
rawDataLen
+=
sizeof
(
pAction
->
actionType
);
rawDataLen
+=
sizeof
(
int8_t
);
}
return
rawDataLen
;
...
...
@@ -137,18 +139,20 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
SDB_SET_INT32
(
pRaw
,
dataPos
,
pAction
->
acceptableCode
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
actionType
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
stage
,
_OVER
)
if
(
pAction
->
actionType
)
{
if
(
pAction
->
actionType
==
TRANS_ACTION_RAW
)
{
int32_t
len
=
sdbGetRawTotalSize
(
pAction
->
pRaw
);
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
rawWritten
,
_OVER
)
SDB_SET_INT32
(
pRaw
,
dataPos
,
len
,
_OVER
)
SDB_SET_BINARY
(
pRaw
,
dataPos
,
(
void
*
)
pAction
->
pRaw
,
len
,
_OVER
)
}
else
{
}
else
if
(
pAction
->
actionType
==
TRANS_ACTION_MSG
)
{
SDB_SET_BINARY
(
pRaw
,
dataPos
,
(
void
*
)
&
pAction
->
epSet
,
sizeof
(
SEpSet
),
_OVER
)
SDB_SET_INT16
(
pRaw
,
dataPos
,
pAction
->
msgType
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
msgSent
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
msgReceived
,
_OVER
)
SDB_SET_INT32
(
pRaw
,
dataPos
,
pAction
->
contLen
,
_OVER
)
SDB_SET_BINARY
(
pRaw
,
dataPos
,
pAction
->
pCont
,
pAction
->
contLen
,
_OVER
)
}
else
{
// nothing
}
}
...
...
@@ -159,18 +163,20 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
SDB_SET_INT32
(
pRaw
,
dataPos
,
pAction
->
acceptableCode
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
actionType
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
stage
,
_OVER
)
if
(
pAction
->
actionType
)
{
if
(
pAction
->
actionType
==
TRANS_ACTION_RAW
)
{
int32_t
len
=
sdbGetRawTotalSize
(
pAction
->
pRaw
);
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
rawWritten
,
_OVER
)
SDB_SET_INT32
(
pRaw
,
dataPos
,
len
,
_OVER
)
SDB_SET_BINARY
(
pRaw
,
dataPos
,
(
void
*
)
pAction
->
pRaw
,
len
,
_OVER
)
}
else
{
}
else
if
(
pAction
->
actionType
==
TRANS_ACTION_MSG
)
{
SDB_SET_BINARY
(
pRaw
,
dataPos
,
(
void
*
)
&
pAction
->
epSet
,
sizeof
(
SEpSet
),
_OVER
)
SDB_SET_INT16
(
pRaw
,
dataPos
,
pAction
->
msgType
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
msgSent
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
msgReceived
,
_OVER
)
SDB_SET_INT32
(
pRaw
,
dataPos
,
pAction
->
contLen
,
_OVER
)
SDB_SET_BINARY
(
pRaw
,
dataPos
,
pAction
->
pCont
,
pAction
->
contLen
,
_OVER
)
}
else
{
// nothing
}
}
...
...
@@ -181,18 +187,20 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
SDB_SET_INT32
(
pRaw
,
dataPos
,
pAction
->
acceptableCode
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
actionType
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
stage
,
_OVER
)
if
(
pAction
->
actionType
)
{
if
(
pAction
->
actionType
==
TRANS_ACTION_RAW
)
{
int32_t
len
=
sdbGetRawTotalSize
(
pAction
->
pRaw
);
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
rawWritten
,
_OVER
)
SDB_SET_INT32
(
pRaw
,
dataPos
,
len
,
_OVER
)
SDB_SET_BINARY
(
pRaw
,
dataPos
,
(
void
*
)
pAction
->
pRaw
,
len
,
_OVER
)
}
else
{
}
else
if
(
pAction
->
actionType
==
TRANS_ACTION_MSG
)
{
SDB_SET_BINARY
(
pRaw
,
dataPos
,
(
void
*
)
&
pAction
->
epSet
,
sizeof
(
SEpSet
),
_OVER
)
SDB_SET_INT16
(
pRaw
,
dataPos
,
pAction
->
msgType
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
msgSent
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
msgReceived
,
_OVER
)
SDB_SET_INT32
(
pRaw
,
dataPos
,
pAction
->
contLen
,
_OVER
)
SDB_SET_BINARY
(
pRaw
,
dataPos
,
pAction
->
pCont
,
pAction
->
contLen
,
_OVER
)
}
else
{
// nothing
}
}
...
...
@@ -252,6 +260,7 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
int16_t
policy
=
0
;
int16_t
conflict
=
0
;
int16_t
exec
=
0
;
int8_t
actionType
=
0
;
SDB_GET_INT16
(
pRaw
,
dataPos
,
&
stage
,
_OVER
)
SDB_GET_INT16
(
pRaw
,
dataPos
,
&
policy
,
_OVER
)
SDB_GET_INT16
(
pRaw
,
dataPos
,
&
conflict
,
_OVER
)
...
...
@@ -279,9 +288,10 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
action
.
id
,
_OVER
)
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
action
.
errCode
,
_OVER
)
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
action
.
acceptableCode
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
actionType
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
actionType
,
_OVER
)
action
.
actionType
=
actionType
;
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
stage
,
_OVER
)
if
(
action
.
actionType
)
{
if
(
action
.
actionType
==
TRANS_ACTION_RAW
)
{
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
rawWritten
,
_OVER
)
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
dataLen
,
_OVER
)
action
.
pRaw
=
taosMemoryMalloc
(
dataLen
);
...
...
@@ -290,7 +300,7 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
SDB_GET_BINARY
(
pRaw
,
dataPos
,
(
void
*
)
action
.
pRaw
,
dataLen
,
_OVER
);
if
(
taosArrayPush
(
pTrans
->
redoActions
,
&
action
)
==
NULL
)
goto
_OVER
;
action
.
pRaw
=
NULL
;
}
else
{
}
else
if
(
action
.
actionType
==
TRANS_ACTION_MSG
)
{
SDB_GET_BINARY
(
pRaw
,
dataPos
,
(
void
*
)
&
action
.
epSet
,
sizeof
(
SEpSet
),
_OVER
);
SDB_GET_INT16
(
pRaw
,
dataPos
,
&
action
.
msgType
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
msgSent
,
_OVER
)
...
...
@@ -301,6 +311,8 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
SDB_GET_BINARY
(
pRaw
,
dataPos
,
action
.
pCont
,
action
.
contLen
,
_OVER
);
if
(
taosArrayPush
(
pTrans
->
redoActions
,
&
action
)
==
NULL
)
goto
_OVER
;
action
.
pCont
=
NULL
;
}
else
{
if
(
taosArrayPush
(
pTrans
->
redoActions
,
&
action
)
==
NULL
)
goto
_OVER
;
}
}
...
...
@@ -308,9 +320,10 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
action
.
id
,
_OVER
)
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
action
.
errCode
,
_OVER
)
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
action
.
acceptableCode
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
actionType
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
actionType
,
_OVER
)
action
.
actionType
=
actionType
;
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
stage
,
_OVER
)
if
(
action
.
actionType
)
{
if
(
action
.
actionType
==
TRANS_ACTION_RAW
)
{
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
rawWritten
,
_OVER
)
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
dataLen
,
_OVER
)
action
.
pRaw
=
taosMemoryMalloc
(
dataLen
);
...
...
@@ -319,7 +332,7 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
SDB_GET_BINARY
(
pRaw
,
dataPos
,
(
void
*
)
action
.
pRaw
,
dataLen
,
_OVER
);
if
(
taosArrayPush
(
pTrans
->
undoActions
,
&
action
)
==
NULL
)
goto
_OVER
;
action
.
pRaw
=
NULL
;
}
else
{
}
else
if
(
action
.
actionType
==
TRANS_ACTION_MSG
)
{
SDB_GET_BINARY
(
pRaw
,
dataPos
,
(
void
*
)
&
action
.
epSet
,
sizeof
(
SEpSet
),
_OVER
);
SDB_GET_INT16
(
pRaw
,
dataPos
,
&
action
.
msgType
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
msgSent
,
_OVER
)
...
...
@@ -330,6 +343,8 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
SDB_GET_BINARY
(
pRaw
,
dataPos
,
action
.
pCont
,
action
.
contLen
,
_OVER
);
if
(
taosArrayPush
(
pTrans
->
undoActions
,
&
action
)
==
NULL
)
goto
_OVER
;
action
.
pCont
=
NULL
;
}
else
{
if
(
taosArrayPush
(
pTrans
->
redoActions
,
&
action
)
==
NULL
)
goto
_OVER
;
}
}
...
...
@@ -337,7 +352,8 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
action
.
id
,
_OVER
)
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
action
.
errCode
,
_OVER
)
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
action
.
acceptableCode
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
actionType
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
actionType
,
_OVER
)
action
.
actionType
=
actionType
;
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
stage
,
_OVER
)
if
(
action
.
actionType
)
{
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
rawWritten
,
_OVER
)
...
...
@@ -348,7 +364,7 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
SDB_GET_BINARY
(
pRaw
,
dataPos
,
(
void
*
)
action
.
pRaw
,
dataLen
,
_OVER
);
if
(
taosArrayPush
(
pTrans
->
commitActions
,
&
action
)
==
NULL
)
goto
_OVER
;
action
.
pRaw
=
NULL
;
}
else
{
}
else
if
(
action
.
actionType
==
TRANS_ACTION_MSG
)
{
SDB_GET_BINARY
(
pRaw
,
dataPos
,
(
void
*
)
&
action
.
epSet
,
sizeof
(
SEpSet
),
_OVER
);
SDB_GET_INT16
(
pRaw
,
dataPos
,
&
action
.
msgType
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
msgSent
,
_OVER
)
...
...
@@ -359,6 +375,8 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
SDB_GET_BINARY
(
pRaw
,
dataPos
,
action
.
pCont
,
action
.
contLen
,
_OVER
);
if
(
taosArrayPush
(
pTrans
->
commitActions
,
&
action
)
==
NULL
)
goto
_OVER
;
action
.
pCont
=
NULL
;
}
else
{
if
(
taosArrayPush
(
pTrans
->
redoActions
,
&
action
)
==
NULL
)
goto
_OVER
;
}
}
...
...
@@ -552,10 +570,12 @@ static void mndTransDropActions(SArray *pArray) {
int32_t
size
=
taosArrayGetSize
(
pArray
);
for
(
int32_t
i
=
0
;
i
<
size
;
++
i
)
{
STransAction
*
pAction
=
taosArrayGet
(
pArray
,
i
);
if
(
pAction
->
actionType
)
{
if
(
pAction
->
actionType
==
TRANS_ACTION_RAW
)
{
taosMemoryFreeClear
(
pAction
->
pRaw
);
}
else
{
}
else
if
(
pAction
->
actionType
==
TRANS_ACTION_MSG
)
{
taosMemoryFreeClear
(
pAction
->
pCont
);
}
else
{
// nothing
}
}
...
...
@@ -583,27 +603,34 @@ static int32_t mndTransAppendAction(SArray *pArray, STransAction *pAction) {
}
int32_t
mndTransAppendRedolog
(
STrans
*
pTrans
,
SSdbRaw
*
pRaw
)
{
STransAction
action
=
{.
stage
=
TRN_STAGE_REDO_ACTION
,
.
actionType
=
true
,
.
pRaw
=
pRaw
};
STransAction
action
=
{.
stage
=
TRN_STAGE_REDO_ACTION
,
.
actionType
=
TRANS_ACTION_RAW
,
.
pRaw
=
pRaw
};
return
mndTransAppendAction
(
pTrans
->
redoActions
,
&
action
);
}
int32_t
mndTransAppendNullLog
(
STrans
*
pTrans
)
{
STransAction
action
=
{.
stage
=
TRN_STAGE_REDO_ACTION
,
.
actionType
=
TRANS_ACTION_NULL
};
return
mndTransAppendAction
(
pTrans
->
redoActions
,
&
action
);
}
int32_t
mndTransAppendUndolog
(
STrans
*
pTrans
,
SSdbRaw
*
pRaw
)
{
STransAction
action
=
{.
stage
=
TRN_STAGE_UNDO_ACTION
,
.
actionType
=
true
,
.
pRaw
=
pRaw
};
STransAction
action
=
{.
stage
=
TRN_STAGE_UNDO_ACTION
,
.
actionType
=
TRANS_ACTION_RAW
,
.
pRaw
=
pRaw
};
return
mndTransAppendAction
(
pTrans
->
undoActions
,
&
action
);
}
int32_t
mndTransAppendCommitlog
(
STrans
*
pTrans
,
SSdbRaw
*
pRaw
)
{
STransAction
action
=
{.
stage
=
TRN_STAGE_COMMIT_ACTION
,
.
actionType
=
true
,
.
pRaw
=
pRaw
};
STransAction
action
=
{.
stage
=
TRN_STAGE_COMMIT_ACTION
,
.
actionType
=
TRANS_ACTION_RAW
,
.
pRaw
=
pRaw
};
return
mndTransAppendAction
(
pTrans
->
commitActions
,
&
action
);
}
int32_t
mndTransAppendRedoAction
(
STrans
*
pTrans
,
STransAction
*
pAction
)
{
pAction
->
stage
=
TRN_STAGE_REDO_ACTION
;
pAction
->
actionType
=
TRANS_ACTION_MSG
;
return
mndTransAppendAction
(
pTrans
->
redoActions
,
pAction
);
}
int32_t
mndTransAppendUndoAction
(
STrans
*
pTrans
,
STransAction
*
pAction
)
{
pAction
->
stage
=
TRN_STAGE_UNDO_ACTION
;
pAction
->
actionType
=
TRANS_ACTION_MSG
;
return
mndTransAppendAction
(
pTrans
->
undoActions
,
pAction
);
}
...
...
@@ -782,7 +809,7 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
}
}
void
mndTransProcessRsp
(
SRpcMsg
*
pRsp
)
{
int32_t
mndTransProcessRsp
(
SRpcMsg
*
pRsp
)
{
SMnode
*
pMnode
=
pRsp
->
info
.
node
;
int64_t
signature
=
(
int64_t
)(
pRsp
->
info
.
ahandle
);
int32_t
transId
=
(
int32_t
)(
signature
>>
32
);
...
...
@@ -827,6 +854,7 @@ void mndTransProcessRsp(SRpcMsg *pRsp) {
_OVER:
mndReleaseTrans
(
pMnode
,
pTrans
);
return
0
;
}
static
void
mndTransResetActions
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
SArray
*
pArray
)
{
...
...
@@ -899,10 +927,15 @@ static int32_t mndTransSendSingleMsg(SMnode *pMnode, STrans *pTrans, STransActio
}
static
int32_t
mndTransExecSingleAction
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
STransAction
*
pAction
)
{
if
(
pAction
->
actionType
)
{
if
(
pAction
->
actionType
==
TRANS_ACTION_RAW
)
{
return
mndTransWriteSingleLog
(
pMnode
,
pTrans
,
pAction
);
}
else
{
}
else
if
(
pAction
->
actionType
==
TRANS_ACTION_MSG
)
{
return
mndTransSendSingleMsg
(
pMnode
,
pTrans
,
pAction
);
}
else
{
pAction
->
rawWritten
=
0
;
pAction
->
errCode
=
0
;
mDebug
(
"trans:%d, %s:%d null action executed"
,
pTrans
->
id
,
mndTransStr
(
pAction
->
stage
),
pAction
->
id
);
return
0
;
}
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录