Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
4f3867e2
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1183
Star
22014
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,发现更多精彩内容 >>
未验证
提交
4f3867e2
编写于
10月 08, 2022
作者:
S
Shengliang Guan
提交者:
GitHub
10月 08, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #17223 from taosdata/fix/TD-19394
fix: continue execute transactions after taosd restart
上级
a472bf16
4b92876f
变更
1
隐藏空白更改
内联
并排
Showing
1 changed file
with
31 addition
and
30 deletion
+31
-30
source/dnode/mnode/impl/src/mndTrans.c
source/dnode/mnode/impl/src/mndTrans.c
+31
-30
未找到文件。
source/dnode/mnode/impl/src/mndTrans.c
浏览文件 @
4f3867e2
...
...
@@ -17,9 +17,9 @@
#include "mndTrans.h"
#include "mndConsumer.h"
#include "mndDb.h"
#include "mndStb.h"
#include "mndPrivilege.h"
#include "mndShow.h"
#include "mndStb.h"
#include "mndSync.h"
#include "mndUser.h"
...
...
@@ -138,6 +138,7 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
SDB_SET_INT32
(
pRaw
,
dataPos
,
undoActionNum
,
_OVER
)
SDB_SET_INT32
(
pRaw
,
dataPos
,
commitActionNum
,
_OVER
)
int8_t
unused
=
0
;
for
(
int32_t
i
=
0
;
i
<
redoActionNum
;
++
i
)
{
STransAction
*
pAction
=
taosArrayGet
(
pTrans
->
redoActions
,
i
);
SDB_SET_INT32
(
pRaw
,
dataPos
,
pAction
->
id
,
_OVER
)
...
...
@@ -149,14 +150,14 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
reserved
,
_OVER
)
if
(
pAction
->
actionType
==
TRANS_ACTION_RAW
)
{
int32_t
len
=
sdbGetRawTotalSize
(
pAction
->
pRaw
);
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
rawWritten
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
unused
/*pAction->rawWritten*/
,
_OVER
)
SDB_SET_INT32
(
pRaw
,
dataPos
,
len
,
_OVER
)
SDB_SET_BINARY
(
pRaw
,
dataPos
,
(
void
*
)
pAction
->
pRaw
,
len
,
_OVER
)
}
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_INT8
(
pRaw
,
dataPos
,
unused
/*pAction->msgSent*/
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
unused
/*pAction->msgReceived*/
,
_OVER
)
SDB_SET_INT32
(
pRaw
,
dataPos
,
pAction
->
contLen
,
_OVER
)
SDB_SET_BINARY
(
pRaw
,
dataPos
,
pAction
->
pCont
,
pAction
->
contLen
,
_OVER
)
}
else
{
...
...
@@ -175,14 +176,14 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
reserved
,
_OVER
)
if
(
pAction
->
actionType
==
TRANS_ACTION_RAW
)
{
int32_t
len
=
sdbGetRawTotalSize
(
pAction
->
pRaw
);
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
rawWritten
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
unused
/*pAction->rawWritten*/
,
_OVER
)
SDB_SET_INT32
(
pRaw
,
dataPos
,
len
,
_OVER
)
SDB_SET_BINARY
(
pRaw
,
dataPos
,
(
void
*
)
pAction
->
pRaw
,
len
,
_OVER
)
}
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_INT8
(
pRaw
,
dataPos
,
unused
/*pAction->msgSent*/
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
unused
/*pAction->msgReceived*/
,
_OVER
)
SDB_SET_INT32
(
pRaw
,
dataPos
,
pAction
->
contLen
,
_OVER
)
SDB_SET_BINARY
(
pRaw
,
dataPos
,
pAction
->
pCont
,
pAction
->
contLen
,
_OVER
)
}
else
{
...
...
@@ -201,14 +202,14 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
reserved
,
_OVER
)
if
(
pAction
->
actionType
==
TRANS_ACTION_RAW
)
{
int32_t
len
=
sdbGetRawTotalSize
(
pAction
->
pRaw
);
SDB_SET_INT8
(
pRaw
,
dataPos
,
pAction
->
rawWritten
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
unused
/*pAction->rawWritten*/
,
_OVER
)
SDB_SET_INT32
(
pRaw
,
dataPos
,
len
,
_OVER
)
SDB_SET_BINARY
(
pRaw
,
dataPos
,
(
void
*
)
pAction
->
pRaw
,
len
,
_OVER
)
}
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_INT8
(
pRaw
,
dataPos
,
unused
/*pAction->msgSent*/
,
_OVER
)
SDB_SET_INT8
(
pRaw
,
dataPos
,
unused
/*pAction->msgReceived*/
,
_OVER
)
SDB_SET_INT32
(
pRaw
,
dataPos
,
pAction
->
contLen
,
_OVER
)
SDB_SET_BINARY
(
pRaw
,
dataPos
,
pAction
->
pCont
,
pAction
->
contLen
,
_OVER
)
}
else
{
...
...
@@ -305,6 +306,7 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
if
(
pTrans
->
undoActions
==
NULL
)
goto
_OVER
;
if
(
pTrans
->
commitActions
==
NULL
)
goto
_OVER
;
int8_t
unused
=
0
;
for
(
int32_t
i
=
0
;
i
<
redoActionNum
;
++
i
)
{
memset
(
&
action
,
0
,
sizeof
(
action
));
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
action
.
id
,
_OVER
)
...
...
@@ -317,7 +319,7 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
action
.
stage
=
stage
;
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
reserved
,
_OVER
)
if
(
action
.
actionType
==
TRANS_ACTION_RAW
)
{
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
rawWritten
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
unused
/*&action.rawWritten*/
,
_OVER
)
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
dataLen
,
_OVER
)
action
.
pRaw
=
taosMemoryMalloc
(
dataLen
);
if
(
action
.
pRaw
==
NULL
)
goto
_OVER
;
...
...
@@ -328,8 +330,8 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
}
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
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
msgReceived
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
unused
/*&action.msgSent*/
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
unused
/*&action.msgReceived*/
,
_OVER
)
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
action
.
contLen
,
_OVER
)
action
.
pCont
=
taosMemoryMalloc
(
action
.
contLen
);
if
(
action
.
pCont
==
NULL
)
goto
_OVER
;
...
...
@@ -353,7 +355,7 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
action
.
stage
=
stage
;
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
reserved
,
_OVER
)
if
(
action
.
actionType
==
TRANS_ACTION_RAW
)
{
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
rawWritten
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
unused
/*&action.rawWritten*/
,
_OVER
)
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
dataLen
,
_OVER
)
action
.
pRaw
=
taosMemoryMalloc
(
dataLen
);
if
(
action
.
pRaw
==
NULL
)
goto
_OVER
;
...
...
@@ -364,8 +366,8 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
}
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
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
msgReceived
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
unused
/*&action.msgSent*/
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
unused
/*&action.msgReceived*/
,
_OVER
)
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
action
.
contLen
,
_OVER
)
action
.
pCont
=
taosMemoryMalloc
(
action
.
contLen
);
if
(
action
.
pCont
==
NULL
)
goto
_OVER
;
...
...
@@ -389,7 +391,7 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
action
.
stage
=
stage
;
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
reserved
,
_OVER
)
if
(
action
.
actionType
)
{
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
rawWritten
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
unused
/*&action.rawWritten*/
,
_OVER
)
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
dataLen
,
_OVER
)
action
.
pRaw
=
taosMemoryMalloc
(
dataLen
);
if
(
action
.
pRaw
==
NULL
)
goto
_OVER
;
...
...
@@ -400,8 +402,8 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
}
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
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
action
.
msgReceived
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
unused
/*&action.msgSent*/
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
unused
/*&action.msgReceived*/
,
_OVER
)
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
action
.
contLen
,
_OVER
)
action
.
pCont
=
taosMemoryMalloc
(
action
.
contLen
);
if
(
action
.
pCont
==
NULL
)
goto
_OVER
;
...
...
@@ -818,7 +820,7 @@ static bool mndCheckTransConflict(SMnode *pMnode, STrans *pNew) {
if
(
mndCheckDbConflict
(
pNew
->
stbname
,
pTrans
))
conflict
=
true
;
}
if
(
pTrans
->
conflict
==
TRN_CONFLICT_DB_INSIDE
)
{
if
(
mndCheckDbConflict
(
pNew
->
stbname
,
pTrans
))
conflict
=
true
;
// for stb
if
(
mndCheckDbConflict
(
pNew
->
stbname
,
pTrans
))
conflict
=
true
;
// for stb
}
}
...
...
@@ -827,9 +829,8 @@ static bool mndCheckTransConflict(SMnode *pMnode, STrans *pNew) {
pNew
->
id
,
pNew
->
dbname
,
pNew
->
stbname
,
pNew
->
conflict
,
pTrans
->
id
,
pTrans
->
dbname
,
pTrans
->
stbname
,
pTrans
->
conflict
);
}
else
{
mInfo
(
"trans:%d, db:%s stb:%s type:%d, not conflict with trans:%d db:%s stb:%s type:%d"
,
pNew
->
id
,
pNew
->
dbname
,
pNew
->
stbname
,
pNew
->
conflict
,
pTrans
->
id
,
pTrans
->
dbname
,
pTrans
->
stbname
,
pTrans
->
conflict
);
mInfo
(
"trans:%d, db:%s stb:%s type:%d, not conflict with trans:%d db:%s stb:%s type:%d"
,
pNew
->
id
,
pNew
->
dbname
,
pNew
->
stbname
,
pNew
->
conflict
,
pTrans
->
id
,
pTrans
->
dbname
,
pTrans
->
stbname
,
pTrans
->
conflict
);
}
sdbRelease
(
pMnode
->
pSdb
,
pTrans
);
}
...
...
@@ -932,7 +933,7 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
SRpcHandleInfo
*
pInfo
=
taosArrayGet
(
pTrans
->
pRpcArray
,
i
);
if
(
pInfo
->
handle
!=
NULL
)
{
mInfo
(
"trans:%d, send rsp, code:0x%x stage:%s app:%p"
,
pTrans
->
id
,
code
,
mndTransStr
(
pTrans
->
stage
),
pInfo
->
ahandle
);
pInfo
->
ahandle
);
if
(
code
==
TSDB_CODE_RPC_NETWORK_UNAVAIL
)
{
code
=
TSDB_CODE_MND_TRANS_NETWORK_UNAVAILL
;
}
...
...
@@ -1015,8 +1016,8 @@ int32_t mndTransProcessRsp(SRpcMsg *pRsp) {
pAction
->
errCode
=
pRsp
->
code
;
}
mInfo
(
"trans:%d, %s:%d response is received, code:0x%x, accept:0x%x retry:0x%x"
,
transId
,
mndTransStr
(
pAction
->
stage
),
action
,
pRsp
->
code
,
pAction
->
acceptableCode
,
pAction
->
retryCode
);
mInfo
(
"trans:%d, %s:%d response is received, code:0x%x, accept:0x%x retry:0x%x"
,
transId
,
mndTransStr
(
pAction
->
stage
),
action
,
pRsp
->
code
,
pAction
->
acceptableCode
,
pAction
->
retryCode
);
mndTransExecute
(
pMnode
,
pTrans
);
_OVER:
...
...
@@ -1032,7 +1033,7 @@ static void mndTransResetAction(SMnode *pMnode, STrans *pTrans, STransAction *pA
pAction
->
errCode
==
TSDB_CODE_SYN_INTERNAL_ERROR
||
pAction
->
errCode
==
TSDB_CODE_SYN_NOT_LEADER
)
{
pAction
->
epSet
.
inUse
=
(
pAction
->
epSet
.
inUse
+
1
)
%
pAction
->
epSet
.
numOfEps
;
mInfo
(
"trans:%d, %s:%d execute status is reset and set epset inuse:%d"
,
pTrans
->
id
,
mndTransStr
(
pAction
->
stage
),
pAction
->
id
,
pAction
->
epSet
.
inUse
);
pAction
->
id
,
pAction
->
epSet
.
inUse
);
}
else
{
mInfo
(
"trans:%d, %s:%d execute status is reset"
,
pTrans
->
id
,
mndTransStr
(
pAction
->
stage
),
pAction
->
id
);
}
...
...
@@ -1062,7 +1063,7 @@ static int32_t mndTransWriteSingleLog(SMnode *pMnode, STrans *pTrans, STransActi
pAction
->
errCode
=
0
;
code
=
0
;
mInfo
(
"trans:%d, %s:%d write to sdb, type:%s status:%s"
,
pTrans
->
id
,
mndTransStr
(
pAction
->
stage
),
pAction
->
id
,
sdbTableName
(
pAction
->
pRaw
->
type
),
sdbStatusName
(
pAction
->
pRaw
->
status
));
sdbTableName
(
pAction
->
pRaw
->
type
),
sdbStatusName
(
pAction
->
pRaw
->
status
));
mndSetTransLastAction
(
pTrans
,
pAction
);
}
else
{
...
...
@@ -1263,7 +1264,7 @@ static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans)
pTrans
->
code
=
0
;
pTrans
->
redoActionPos
++
;
mInfo
(
"trans:%d, %s:%d is executed and need sync to other mnodes"
,
pTrans
->
id
,
mndTransStr
(
pAction
->
stage
),
pAction
->
id
);
pAction
->
id
);
code
=
mndTransSync
(
pMnode
,
pTrans
);
if
(
code
!=
0
)
{
pTrans
->
code
=
terrno
;
...
...
@@ -1282,7 +1283,7 @@ static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans)
terrno
=
code
;
pTrans
->
code
=
code
;
mInfo
(
"trans:%d, %s:%d receive code:0x%x and wait another schedule, failedTimes:%d"
,
pTrans
->
id
,
mndTransStr
(
pAction
->
stage
),
pAction
->
id
,
code
,
pTrans
->
failedTimes
);
mndTransStr
(
pAction
->
stage
),
pAction
->
id
,
code
,
pTrans
->
failedTimes
);
break
;
}
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录