@@ -667,6 +667,29 @@ MtmBeginTransaction(MtmCurrentTrans* x)
667
667
}
668
668
}
669
669
670
+
671
+ static MtmTransState *
672
+ MtmCreateTransState (MtmCurrentTrans * x )
673
+ {
674
+ bool found ;
675
+ MtmTransState * ts = hash_search (MtmXid2State , & x -> xid , HASH_ENTER , & found );
676
+ if (!found ) {
677
+ ts -> status = TRANSACTION_STATUS_IN_PROGRESS ;
678
+ ts -> snapshot = x -> snapshot ;
679
+ if (TransactionIdIsValid (x -> gtid .xid )) {
680
+ Assert (x -> gtid .node != MtmNodeId );
681
+ ts -> gtid = x -> gtid ;
682
+ } else {
683
+ /* I am coordinator of transaction */
684
+ ts -> gtid .xid = x -> xid ;
685
+ ts -> gtid .node = MtmNodeId ;
686
+ }
687
+ }
688
+ return ts ;
689
+ }
690
+
691
+
692
+
670
693
/*
671
694
* Prepare transaction for two-phase commit.
672
695
* This code is executed by PRE_PREPARE hook before PREPARE message is sent to replicas by logical replication
@@ -675,7 +698,7 @@ static void
675
698
MtmPrePrepareTransaction (MtmCurrentTrans * x )
676
699
{
677
700
MtmTransState * ts ;
678
- TransactionId * subxids ;
701
+ TransactionId * subxids ;
679
702
680
703
if (!x -> isDistributed ) {
681
704
return ;
@@ -703,14 +726,12 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
703
726
MtmCheckClusterLock ();
704
727
}
705
728
706
- ts = hash_search (MtmXid2State , & x -> xid , HASH_ENTER , NULL );
707
- ts -> status = TRANSACTION_STATUS_IN_PROGRESS ;
729
+ ts = MtmCreateTransState (x );
708
730
/*
709
731
* Invalid CSN prevent replication of transaction by logical replication
710
732
*/
711
733
ts -> snapshot = x -> isReplicated || !x -> containsDML ? INVALID_CSN : x -> snapshot ;
712
734
ts -> csn = MtmAssignCSN ();
713
- ts -> gtid = x -> gtid ;
714
735
ts -> procno = MyProc -> pgprocno ;
715
736
ts -> nVotes = 1 ; /* I am voted myself */
716
737
ts -> votingCompleted = false;
@@ -722,15 +743,6 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
722
743
x -> csn = ts -> csn ;
723
744
724
745
Mtm -> transCount += 1 ;
725
-
726
- if (TransactionIdIsValid (x -> gtid .xid )) {
727
- Assert (x -> gtid .node != MtmNodeId );
728
- ts -> gtid = x -> gtid ;
729
- } else {
730
- /* I am coordinator of transaction */
731
- ts -> gtid .xid = x -> xid ;
732
- ts -> gtid .node = MtmNodeId ;
733
- }
734
746
MtmTransactionListAppend (ts );
735
747
MtmAddSubtransactions (ts , subxids , ts -> nSubxids );
736
748
MTM_TRACE ("%d: MtmPrePrepareTransaction prepare commit of %d (gtid.xid=%d, gtid.node=%d, CSN=%ld)\n" ,
@@ -844,7 +856,9 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
844
856
MtmTransactionListAppend (ts );
845
857
}
846
858
MtmSendNotificationMessage (ts , MSG_ABORTED ); /* send notification to coordinator */
847
- }
859
+ } else if (x -> status == TRANSACTION_STATUS_ABORTED && x -> isReplicated && !x -> isPrepared ) {
860
+ hash_search (MtmXid2State , & x -> xid , HASH_REMOVE , NULL );
861
+ }
848
862
MtmUnlock ();
849
863
}
850
864
MtmResetTransaction (x );
@@ -868,28 +882,32 @@ void MtmSendNotificationMessage(MtmTransState* ts, MtmMessageCode cmd)
868
882
869
883
void MtmJoinTransaction (GlobalTransactionId * gtid , csn_t globalSnapshot )
870
884
{
885
+ MtmTx .gtid = * gtid ;
886
+ MtmTx .xid = GetCurrentTransactionId ();
887
+ MtmTx .isReplicated = true;
888
+ MtmTx .isDistributed = true;
889
+ MtmTx .containsDML = true;
890
+
871
891
if (globalSnapshot != INVALID_CSN ) {
872
892
MtmLock (LW_EXCLUSIVE );
873
893
MtmSyncClock (globalSnapshot );
894
+ MtmTx .snapshot = globalSnapshot ;
895
+ if (Mtm -> status != MTM_RECOVERY ) {
896
+ MtmCreateTransState (& MtmTx ); /* we need local->remote xid mapping for deadlock detection */
897
+ }
874
898
MtmUnlock ();
875
899
} else {
876
900
globalSnapshot = MtmTx .snapshot ;
877
901
}
878
902
if (!TransactionIdIsValid (gtid -> xid )) {
879
903
/* In case of recovery InvalidTransactionId is passed */
880
904
if (Mtm -> status != MTM_RECOVERY ) {
881
- elog (PANIC , "Node %d tries to recover node %d which is in %s mode" , MtmReplicationNodeId , MtmNodeId , MtmNodeStatusMnem [Mtm -> status ]);
905
+ elog (PANIC , "Node %d tries to recover node %d which is in %s mode" , gtid -> node , MtmNodeId , MtmNodeStatusMnem [Mtm -> status ]);
882
906
}
883
907
} else if (Mtm -> status == MTM_RECOVERY ) {
884
908
/* When recovery is completed we get normal transaction ID and switch to normal mode */
885
909
MtmRecoveryCompleted ();
886
910
}
887
- MtmTx .gtid = * gtid ;
888
- MtmTx .xid = GetCurrentTransactionId ();
889
- MtmTx .snapshot = globalSnapshot ;
890
- MtmTx .isReplicated = true;
891
- MtmTx .isDistributed = true;
892
- MtmTx .containsDML = true;
893
911
}
894
912
895
913
void MtmSetCurrentTransactionGID (char const * gid )
0 commit comments