Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commitb82ebd0

Browse files
knizhnikkelvich
authored andcommitted
Abort voting backends on node disconnect
1 parent671f689 commitb82ebd0

File tree

3 files changed

+36
-18
lines changed

3 files changed

+36
-18
lines changed

‎arbiter.c

Lines changed: 2 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -584,12 +584,6 @@ static void MtmTransSender(Datum arg)
584584
}
585585
}
586586

587-
staticvoidMtmWakeUpBackend(MtmTransState*ts)
588-
{
589-
MTM_TRACE("Wakeup backed procno=%d, pid=%d\n",ts->procno,ProcGlobal->allProcs[ts->procno].pid);
590-
ts->votingCompleted= true;
591-
SetLatch(&ProcGlobal->allProcs[ts->procno].procLatch);
592-
}
593587

594588
#if !USE_EPOLL
595589
staticboolMtmRecovery()
@@ -715,9 +709,7 @@ static void MtmTransReceiver(Datum arg)
715709
commit on smaller subset of nodes */
716710
elog(WARNING,"Coordinator of distributed transaction see less nodes than node %d: %lx instead of %lx",
717711
msg->node,Mtm->disabledNodeMask,msg->disabledNodeMask);
718-
ts->status=TRANSACTION_STATUS_ABORTED;
719-
MtmAdjustSubtransactions(ts);
720-
Mtm->nActiveTransactions-=1;
712+
MtmAbortTransaction(ts);
721713
}
722714

723715
if (++ts->nVotes==Mtm->nNodes) {
@@ -735,9 +727,7 @@ static void MtmTransReceiver(Datum arg)
735727
Assert(ts->nVotes<Mtm->nNodes);
736728
if (ts->status!=TRANSACTION_STATUS_ABORTED) {
737729
Assert(ts->status==TRANSACTION_STATUS_IN_PROGRESS);
738-
ts->status=TRANSACTION_STATUS_ABORTED;
739-
MtmAdjustSubtransactions(ts);
740-
Mtm->nActiveTransactions-=1;
730+
MtmAbortTransaction(ts);
741731
}
742732
if (++ts->nVotes==Mtm->nNodes) {
743733
MtmWakeUpBackend(ts);

‎multimaster.c

Lines changed: 32 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -552,6 +552,7 @@ static void MtmAddSubtransactions(MtmTransState* ts, TransactionId* subxids, int
552552
Assert(!found);
553553
sts->status=ts->status;
554554
sts->csn=ts->csn;
555+
sts->votingCompleted= true;
555556
MtmTransactionListInsertAfter(ts,sts);
556557
}
557558
}
@@ -744,7 +745,8 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
744745
if (!MtmIsCoordinator(ts)||Mtm->status==MTM_RECOVERY) {
745746
MtmTransMap*tm= (MtmTransMap*)hash_search(MtmGid2State,x->gid,HASH_ENTER,NULL);
746747
Assert(x->gid[0]);
747-
tm->state=ts;
748+
tm->state=ts;
749+
ts->votingCompleted= true;
748750
if (Mtm->status!=MTM_RECOVERY) {
749751
MtmSendNotificationMessage(ts,MSG_READY);/* send notification to coordinator */
750752
}else {
@@ -776,9 +778,7 @@ MtmAbortPreparedTransaction(MtmCurrentTrans* x)
776778
MtmLock(LW_EXCLUSIVE);
777779
tm= (MtmTransMap*)hash_search(MtmGid2State,x->gid,HASH_REMOVE,NULL);
778780
Assert(tm!=NULL);
779-
tm->state->status=TRANSACTION_STATUS_ABORTED;
780-
MtmAdjustSubtransactions(tm->state);
781-
Mtm->nActiveTransactions-=1;
781+
MtmAbortTransaction(tm->state);
782782
MtmUnlock();
783783
x->status=TRANSACTION_STATUS_ABORTED;
784784
}
@@ -834,6 +834,7 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
834834
ts->gtid=x->gtid;
835835
ts->nSubxids=0;
836836
ts->cmd=MSG_INVALID;
837+
ts->votingCompleted= true;
837838
MtmTransactionListAppend(ts);
838839
}
839840
MtmSendNotificationMessage(ts,MSG_ABORTED);/* send notification to coordinator */
@@ -936,6 +937,20 @@ csn_t MtmGetTransactionCSN(TransactionId xid)
936937
returncsn;
937938
}
938939

940+
voidMtmWakeUpBackend(MtmTransState*ts)
941+
{
942+
MTM_TRACE("Wakeup backed procno=%d, pid=%d\n",ts->procno,ProcGlobal->allProcs[ts->procno].pid);
943+
ts->votingCompleted= true;
944+
SetLatch(&ProcGlobal->allProcs[ts->procno].procLatch);
945+
}
946+
947+
voidMtmAbortTransaction(MtmTransState*ts)
948+
{
949+
ts->status=TRANSACTION_STATUS_ABORTED;
950+
MtmAdjustSubtransactions(ts);
951+
Mtm->nActiveTransactions-=1;
952+
}
953+
939954
/*
940955
* -------------------------------------------
941956
* HA functions
@@ -1212,9 +1227,10 @@ void MtmCheckQuorum(void)
12121227
}
12131228
}
12141229

1215-
12161230
voidMtmOnNodeDisconnect(intnodeId)
1217-
{
1231+
{
1232+
MtmTransState*ts;
1233+
12181234
BIT_SET(Mtm->connectivityMask,nodeId-1);
12191235
BIT_SET(Mtm->reconnectMask,nodeId-1);
12201236
RaftableSet(psprintf("node-mask-%d",MtmNodeId),&Mtm->connectivityMask,sizeofMtm->connectivityMask, false);
@@ -1228,6 +1244,16 @@ void MtmOnNodeDisconnect(int nodeId)
12281244
BIT_SET(Mtm->disabledNodeMask,nodeId-1);
12291245
Mtm->nNodes-=1;
12301246
MtmCheckQuorum();
1247+
/* Interrupt voting for active transaction and abort them */
1248+
for (ts=Mtm->transListHead;ts!=NULL;ts=ts->next) {
1249+
if (!ts->votingCompleted) {
1250+
if (ts->status!=TRANSACTION_STATUS_ABORTED) {
1251+
elog(WARNING,"Rollback active transaction %d:%d",ts->gtid.node,ts->gtid.xid);
1252+
MtmAbortTransaction(ts);
1253+
}
1254+
MtmWakeUpBackend(ts);
1255+
}
1256+
}
12311257
}
12321258
MtmUnlock();
12331259
}

‎multimaster.h

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -199,8 +199,10 @@ extern void MtmDropNode(int nodeId, bool dropSlot);
199199
externvoidMtmRecoverNode(intnodeId);
200200
externvoidMtmOnNodeDisconnect(intnodeId);
201201
externvoidMtmOnNodeConnect(intnodeId);
202+
externvoidMtmWakeUpBackend(MtmTransState*ts);
202203
externtimestamp_tMtmGetCurrentTime(void);
203204
externvoidMtmSleep(timestamp_tinterval);
205+
externvoidMtmAbortTransaction(MtmTransState*ts);
204206
externvoidMtmSetCurrentTransactionGID(charconst*gid);
205207
externcsn_tMtmGetTransactionCSN(TransactionIdxid);
206208
externvoidMtmSetCurrentTransactionCSN(csn_tcsn);

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp