Make it possible to abort transaction from XTM TransactionIdSetStatus call

Author: knizhnik

src/backend/access/transam/clog.c

@@ -93,11 +93,11 @@ static void TransactionIdSetStatusBit(TransactionId xid, XidStatus status,
 static void set_status_by_pages(int nsubxids, TransactionId *subxids,
 					XidStatus status, XLogRecPtr lsn);
 
-void
+bool
 TransactionIdSetTreeStatus(TransactionId xid, int nsubxids,
-					TransactionId *subxids, XidStatus status, XLogRecPtr lsn)
+                           TransactionId *subxids, XidStatus status, XLogRecPtr lsn)
 {
-	TM->SetTransactionStatus(xid, nsubxids, subxids, status, lsn);
+	return TM->SetTransactionStatus(xid, nsubxids, subxids, status, lsn);
 }
 
 /*
@@ -151,7 +151,7 @@ TransactionIdSetTreeStatus(TransactionId xid, int nsubxids,
  * but aren't yet in cache, as well as hinting pages not to fall out of
  * cache yet.
  */
-void
+bool
 PgTransactionIdSetTreeStatus(TransactionId xid, int nsubxids,
 					TransactionId *subxids, XidStatus status, XLogRecPtr lsn)
 {
@@ -217,6 +217,7 @@ PgTransactionIdSetTreeStatus(TransactionId xid, int nsubxids,
 							subxids + nsubxids_on_first_page,
 							status, lsn);
 	}
+    return true;
 }
 
 /*

src/backend/access/transam/transam.c

@@ -256,24 +256,24 @@ TransactionIdIsKnownCompleted(TransactionId transactionId)
  * This commit operation is not guaranteed to be atomic, but if not, subxids
  * are correctly marked subcommit first.
  */
-void
+bool 
 TransactionIdCommitTree(TransactionId xid, int nxids, TransactionId *xids)
 {
-	TransactionIdSetTreeStatus(xid, nxids, xids,
-							   TRANSACTION_STATUS_COMMITTED,
-							   InvalidXLogRecPtr);
+	return TransactionIdSetTreeStatus(xid, nxids, xids,
+                                      TRANSACTION_STATUS_COMMITTED,
+                                      InvalidXLogRecPtr);
 }
 
 /*
  * TransactionIdAsyncCommitTree
  *		Same as above, but for async commits.  The commit record LSN is needed.
  */
-void
+bool
 TransactionIdAsyncCommitTree(TransactionId xid, int nxids, TransactionId *xids,
 							 XLogRecPtr lsn)
 {
-	TransactionIdSetTreeStatus(xid, nxids, xids,
-							   TRANSACTION_STATUS_COMMITTED, lsn);
+	return TransactionIdSetTreeStatus(xid, nxids, xids,
+                                      TRANSACTION_STATUS_COMMITTED, lsn);
 }
 
 /*

src/backend/access/transam/xact.c

@@ -1135,6 +1135,7 @@ RecordTransactionCommit(void)
 	SharedInvalidationMessage *invalMessages = NULL;
 	bool		RelcacheInitFileInval = false;
 	bool		wrote_xlog;
+    bool        committed = false;
 
 	/* Get data needed for commit record */
 	nrels = smgrGetPendingDeletes(true, &rels);
@@ -1273,8 +1274,9 @@ RecordTransactionCommit(void)
 		/*
 		 * Now we may update the CLOG, if we wrote a COMMIT record above
 		 */
-		if (markXidCommitted)
-			TransactionIdCommitTree(xid, nchildren, children);
+		if (markXidCommitted) {
+			committed = TransactionIdCommitTree(xid, nchildren, children);
+        }
 	}
 	else
 	{
@@ -1296,8 +1298,9 @@ RecordTransactionCommit(void)
 		 * XLOG. Instead, we store the LSN up to which the XLOG must be
 		 * flushed before the CLOG may be updated.
 		 */
-		if (markXidCommitted)
-			TransactionIdAsyncCommitTree(xid, nchildren, children, XactLastRecEnd);
+		if (markXidCommitted) {
+			committed = TransactionIdAsyncCommitTree(xid, nchildren, children, XactLastRecEnd);
+        }
 	}
 
 	/*
@@ -1308,6 +1311,9 @@ RecordTransactionCommit(void)
 	{
 		MyPgXact->delayChkpt = false;
 		END_CRIT_SECTION();
+        if (!committed) {
+            elog(ERROR, "Transaction commit rejected by XTM");
+        }
 	}
 
 	/* Compute latestXid while we have the child XIDs handy */
@@ -5310,6 +5316,7 @@ xact_redo_commit(xl_xact_parsed_commit *parsed,
 	TransactionId max_xid;
 	int			i;
 	TimestampTz commit_time;
+    bool committed;
 
 	max_xid = TransactionIdLatest(xid, parsed->nsubxacts, parsed->subxacts);
 
@@ -5345,7 +5352,7 @@ xact_redo_commit(xl_xact_parsed_commit *parsed,
 		/*
 		 * Mark the transaction committed in pg_clog.
 		 */
-		TransactionIdCommitTree(xid, parsed->nsubxacts, parsed->subxacts);
+		committed = TransactionIdCommitTree(xid, parsed->nsubxacts, parsed->subxacts);
 	}
 	else
 	{
@@ -5369,8 +5376,8 @@ xact_redo_commit(xl_xact_parsed_commit *parsed,
 		 * bits set on changes made by transactions that haven't yet
 		 * recovered. It's unlikely but it's good to be safe.
 		 */
-		TransactionIdAsyncCommitTree(
-							  xid, parsed->nsubxacts, parsed->subxacts, lsn);
+		committed = TransactionIdAsyncCommitTree(
+            xid, parsed->nsubxacts, parsed->subxacts, lsn);
 
 		/*
 		 * We must mark clog before we update the ProcArray.
@@ -5397,7 +5404,9 @@ xact_redo_commit(xl_xact_parsed_commit *parsed,
 		 */
 		StandbyReleaseLockTree(xid, 0, NULL);
 	}
-
+    if (!committed) {
+        elog(NOTICE, "XTM rejected recovert of tran saction %u", xid);
+    }
 	if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
 	{
 		/* recover apply progress */

src/include/access/clog.h

@@ -30,8 +30,8 @@ typedef int XidStatus;
 #define TRANSACTION_STATUS_UNKNOWN 	        0x03
 
 
-extern void TransactionIdSetTreeStatus(TransactionId xid, int nsubxids,
-				   TransactionId *subxids, XidStatus status, XLogRecPtr lsn);
+extern bool TransactionIdSetTreeStatus(TransactionId xid, int nsubxids,
+                                       TransactionId *subxids, XidStatus status, XLogRecPtr lsn);
 extern XidStatus TransactionIdGetStatus(TransactionId xid, XLogRecPtr *lsn);
 
 extern Size CLOGShmemBuffers(void);

src/include/access/transam.h

@@ -157,8 +157,8 @@ extern bool TransactionIdDidCommit(TransactionId transactionId);
 extern bool TransactionIdDidAbort(TransactionId transactionId);
 extern bool TransactionIdIsKnownCompleted(TransactionId transactionId);
 extern void TransactionIdAbort(TransactionId transactionId);
-extern void TransactionIdCommitTree(TransactionId xid, int nxids, TransactionId *xids);
-extern void TransactionIdAsyncCommitTree(TransactionId xid, int nxids, TransactionId *xids, XLogRecPtr lsn);
+extern bool TransactionIdCommitTree(TransactionId xid, int nxids, TransactionId *xids);
+extern bool TransactionIdAsyncCommitTree(TransactionId xid, int nxids, TransactionId *xids, XLogRecPtr lsn);
 extern void TransactionIdAbortTree(TransactionId xid, int nxids, TransactionId *xids);
 extern bool TransactionIdPrecedes(TransactionId id1, TransactionId id2);
 extern bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2);

src/include/access/xtm.h

@@ -21,7 +21,7 @@ typedef struct
 	XidStatus (*GetTransactionStatus)(TransactionId xid, XLogRecPtr *lsn);
 
 	/* Set current transaction status (encapsulation of TransactionIdGetStatus in clog.c) */
-	void (*SetTransactionStatus)(TransactionId xid, int nsubxids, TransactionId *subxids, XidStatus status, XLogRecPtr lsn);
+	bool (*SetTransactionStatus)(TransactionId xid, int nsubxids, TransactionId *subxids, XidStatus status, XLogRecPtr lsn);
 
 	/* Get current transaction snaphot (encapsulation of GetSnapshotData in procarray.c) */
 	Snapshot (*GetSnapshot)(Snapshot snapshot);
@@ -51,8 +51,8 @@ extern TransactionManager PgTM; /* Standard PostgreSQL transaction manager */
 /* Standard PostgreSQL function implementing TM interface */
 extern bool PgXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot);
 
-extern void PgTransactionIdSetTreeStatus(TransactionId xid, int nsubxids,
-										TransactionId *subxids, XidStatus status, XLogRecPtr lsn);
+extern bool PgTransactionIdSetTreeStatus(TransactionId xid, int nsubxids,
+                                         TransactionId *subxids, XidStatus status, XLogRecPtr lsn);
 extern XidStatus PgTransactionIdGetStatus(TransactionId xid, XLogRecPtr *lsn);
 
 extern Snapshot PgGetSnapshotData(Snapshot snapshot);