port xtm to 11-current

Author: kelvich

src/backend/access/transam/Makefile

@@ -15,7 +15,7 @@ include $(top_builddir)/src/Makefile.global
 OBJS = clog.o commit_ts.o generic_xlog.o multixact.o parallel.o rmgr.o slru.o \
 	subtrans.o timeline.o transam.o twophase.o twophase_rmgr.o varsup.o \
 	xact.o xlog.o xlogarchive.o xlogfuncs.o \
-	xloginsert.o xlogreader.o xlogutils.o
+	xloginsert.o xlogreader.o xlogutils.o xtm.o
 
 include $(top_srcdir)/src/backend/common.mk
 

src/backend/access/transam/clog.c

@@ -38,6 +38,7 @@
 #include "access/xlog.h"
 #include "access/xloginsert.h"
 #include "access/xlogutils.h"
+#include "access/xtm.h"
 #include "miscadmin.h"
 #include "pg_trace.h"
 
@@ -79,6 +80,12 @@ static SlruCtlData ClogCtlData;
 
 #define ClogCtl (&ClogCtlData)
 
+void
+TransactionIdSetTreeStatus(TransactionId xid, int nsubxids,
+					TransactionId *subxids, XidStatus status, XLogRecPtr lsn)
+{
+	return TM->SetTransactionStatus(xid, nsubxids, subxids, status, lsn);
+}
 
 static int	ZeroCLOGPage(int pageno, bool writeXlog);
 static bool CLOGPagePrecedes(int page1, int page2);
@@ -146,7 +153,7 @@ static void set_status_by_pages(int nsubxids, TransactionId *subxids,
  * cache yet.
  */
 void
-TransactionIdSetTreeStatus(TransactionId xid, int nsubxids,
+PgTransactionIdSetTreeStatus(TransactionId xid, int nsubxids,
 						   TransactionId *subxids, XidStatus status, XLogRecPtr lsn)
 {
 	int			pageno = TransactionIdToPage(xid);	/* get page of parent */
@@ -352,10 +359,10 @@ TransactionIdSetStatusBit(TransactionId xid, XidStatus status, XLogRecPtr lsn, i
 	 * Current state change should be from 0 or subcommitted to target state
 	 * or we should already be there when replaying changes during recovery.
 	 */
-	Assert(curval == 0 ||
-		   (curval == TRANSACTION_STATUS_SUB_COMMITTED &&
-			status != TRANSACTION_STATUS_IN_PROGRESS) ||
-		   curval == status);
+/*	Assert(curval == 0 || */
+/*		   (curval == TRANSACTION_STATUS_SUB_COMMITTED && */
+/*			status != TRANSACTION_STATUS_IN_PROGRESS) || */
+/*		   curval == status); */
 
 	/* note this assumes exclusive access to the clog page */
 	byteval = *byteptr;
@@ -397,6 +404,12 @@ TransactionIdSetStatusBit(TransactionId xid, XidStatus status, XLogRecPtr lsn, i
  */
 XidStatus
 TransactionIdGetStatus(TransactionId xid, XLogRecPtr *lsn)
+{
+	return TM->GetTransactionStatus(xid, lsn);
+}
+
+XidStatus
+PgTransactionIdGetStatus(TransactionId xid, XLogRecPtr *lsn)
 {
 	int			pageno = TransactionIdToPage(xid);
 	int			byteno = TransactionIdToByte(xid);

src/backend/access/transam/subtrans.c

@@ -115,7 +115,7 @@ SubTransGetParent(TransactionId xid)
 	TransactionId parent;
 
 	/* Can't ask about stuff that might not be around anymore */
-	Assert(TransactionIdFollowsOrEquals(xid, TransactionXmin));
+	//Assert(TransactionIdFollowsOrEquals(xid, TransactionXmin));
 
 	/* Bootstrap and frozen XIDs have no parent */
 	if (!TransactionIdIsNormal(xid))
@@ -153,7 +153,10 @@ SubTransGetTopmostTransaction(TransactionId xid)
 				previousXid = xid;
 
 	/* Can't ask about stuff that might not be around anymore */
-	Assert(TransactionIdFollowsOrEquals(xid, TransactionXmin));
+	if (!TransactionIdFollowsOrEquals(xid, TransactionXmin)) 
+	{
+		return xid;
+	}
 
 	while (TransactionIdIsValid(parentXid))
 	{

src/backend/access/transam/varsup.c

@@ -19,6 +19,7 @@
 #include "access/transam.h"
 #include "access/xact.h"
 #include "access/xlog.h"
+#include "access/xtm.h"
 #include "commands/dbcommands.h"
 #include "miscadmin.h"
 #include "postmaster/autovacuum.h"
@@ -33,6 +34,11 @@
 /* pointer to "variable cache" in shared memory (set up by shmem.c) */
 VariableCache ShmemVariableCache = NULL;
 
+TransactionId
+GetNewTransactionId(bool isSubXact)
+{
+	return TM->GetNewTransactionId(isSubXact);
+}
 
 /*
  * Allocate the next XID for a new transaction or subtransaction.
@@ -45,7 +51,7 @@ VariableCache ShmemVariableCache = NULL;
  * issue a warning about XID wrap.
  */
 TransactionId
-GetNewTransactionId(bool isSubXact)
+PgGetNewTransactionId(bool isSubXact)
 {
 	TransactionId xid;
 

src/backend/access/transam/xact.c

@@ -27,6 +27,7 @@
 #include "access/transam.h"
 #include "access/twophase.h"
 #include "access/xact.h"
+#include "access/xtm.h"
 #include "access/xlog.h"
 #include "access/xloginsert.h"
 #include "access/xlogutils.h"
@@ -4897,6 +4898,7 @@ SerializeTransactionState(Size maxsize, char *start_address)
 		Assert(maxsize >= (nParallelCurrentXids + c) * sizeof(TransactionId));
 		memcpy(&result[c], ParallelCurrentXids,
 			   nParallelCurrentXids * sizeof(TransactionId));
+		TM->SerializeTransactionState(&result[c + nParallelCurrentXids]);
 		return;
 	}
 
@@ -4930,6 +4932,7 @@ SerializeTransactionState(Size maxsize, char *start_address)
 	/* Copy data into output area. */
 	result[c++] = (TransactionId) nxids;
 	memcpy(&result[c], workspace, nxids * sizeof(TransactionId));
+	TM->SerializeTransactionState(&result[c + nxids]);
 }
 
 /*
@@ -4952,6 +4955,7 @@ StartParallelWorkerTransaction(char *tstatespace)
 	currentCommandId = tstate[4];
 	nParallelCurrentXids = (int) tstate[5];
 	ParallelCurrentXids = &tstate[6];
+	TM->DeserializeTransactionState(&tstate[nParallelCurrentXids + 6]);
 
 	CurrentTransactionState->blockState = TBLOCK_PARALLEL_INPROGRESS;
 }

src/backend/access/transam/xtm.c

@@ -0,0 +1,85 @@
+/*-------------------------------------------------------------------------
+ *
+ * xtm.c
+ *		PostgreSQL implementation of transaction manager protocol
+ *
+ * This module defines default iplementaiton of PostgreSQL transaction manager protocol
+ *
+ * Portions Copyright (c) 1996-2015, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/backend/access/transam/xtm.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/transam.h"
+#include "access/xtm.h"
+
+TransactionId
+PgGetGlobalTransactionId(void)
+{
+	return InvalidTransactionId;
+}
+
+bool
+PgDetectGlobalDeadLock(PGPROC *proc)
+{
+	return false;
+}
+
+char const *
+PgGetTransactionManagerName(void)
+{
+	return "postgres";
+}
+
+size_t 
+PgGetTransactionStateSize(void)
+{
+	return 0;
+}
+
+void
+PgSerializeTransactionState(void* ctx)
+{
+}
+
+void
+PgDeserializeTransactionState(void* ctx)
+{
+}
+
+void PgInitializeSequence(int64* init, int64* step)
+{
+	*init = 1;
+	*step = 1;
+}
+
+
+TransactionManager PgTM = {
+	PgTransactionIdGetStatus,
+	PgTransactionIdSetTreeStatus,
+	PgGetSnapshotData,
+	PgGetNewTransactionId,
+	PgGetOldestXmin,
+	PgTransactionIdIsInProgress,
+	PgGetGlobalTransactionId,
+	PgXidInMVCCSnapshot,
+	PgDetectGlobalDeadLock,
+	PgGetTransactionManagerName,
+	PgGetTransactionStateSize,
+	PgSerializeTransactionState,
+	PgDeserializeTransactionState,
+	PgInitializeSequence
+};
+
+TransactionManager *TM = &PgTM;
+
+TransactionManager *
+GetTransactionManager(void)
+{
+	return TM;
+}

src/backend/storage/ipc/procarray.c

@@ -51,6 +51,7 @@
 #include "access/twophase.h"
 #include "access/xact.h"
 #include "access/xlog.h"
+#include "access/xtm.h"
 #include "catalog/catalog.h"
 #include "miscadmin.h"
 #include "pgstat.h"
@@ -969,6 +970,12 @@ ProcArrayApplyXidAssignment(TransactionId topxid,
 	LWLockRelease(ProcArrayLock);
 }
 
+bool
+TransactionIdIsInProgress(TransactionId xid)
+{
+	return TM->IsInProgress(xid);
+}
+
 /*
  * TransactionIdIsInProgress -- is given transaction running in some backend
  *
@@ -996,7 +1003,7 @@ ProcArrayApplyXidAssignment(TransactionId topxid,
  * PGXACT again anyway; see GetNewTransactionId).
  */
 bool
-TransactionIdIsInProgress(TransactionId xid)
+PgTransactionIdIsInProgress(TransactionId xid)
 {
 	static TransactionId *xids = NULL;
 	int			nxids = 0;
@@ -1256,6 +1263,11 @@ TransactionIdIsActive(TransactionId xid)
 	return result;
 }
 
+TransactionId
+GetOldestXmin(Relation rel, int flags)
+{
+	return TM->GetOldestXmin(rel, flags);
+}
 
 /*
  * GetOldestXmin -- returns oldest transaction that was running
@@ -1312,7 +1324,7 @@ TransactionIdIsActive(TransactionId xid)
  * GetOldestXmin() move backwards, with no consequences for data integrity.
  */
 TransactionId
-GetOldestXmin(Relation rel, int flags)
+PgGetOldestXmin(Relation rel, int flags)
 {
 	ProcArrayStruct *arrayP = procArray;
 	TransactionId result;
@@ -1469,6 +1481,12 @@ GetMaxSnapshotSubxidCount(void)
 	return TOTAL_MAX_CACHED_SUBXIDS;
 }
 
+Snapshot
+GetSnapshotData(Snapshot snapshot)
+{
+	return TM->GetSnapshot(snapshot);
+}
+
 /*
  * GetSnapshotData -- returns information about running transactions.
  *
@@ -1506,7 +1524,7 @@ GetMaxSnapshotSubxidCount(void)
  * not statically allocated (see xip allocation below).
  */
 Snapshot
-GetSnapshotData(Snapshot snapshot)
+PgGetSnapshotData(Snapshot snapshot)
 {
 	ProcArrayStruct *arrayP = procArray;
 	TransactionId xmin;

src/backend/storage/lmgr/deadlock.c

@@ -30,6 +30,7 @@
 #include "pgstat.h"
 #include "storage/lmgr.h"
 #include "storage/proc.h"
+#include "access/xtm.h"
 #include "utils/memutils.h"
 
 
@@ -281,7 +282,7 @@ DeadLockCheck(PGPROC *proc)
 	else if (blocking_autovacuum_proc != NULL)
 		return DS_BLOCKED_BY_AUTOVACUUM;
 	else
-		return DS_NO_DEADLOCK;
+		return TM->DetectGlobalDeadLock(proc) ? DS_DISTRIBUTED_DEADLOCK : DS_NO_DEADLOCK;
 }
 
 /*

src/backend/storage/lmgr/lock.c

@@ -3849,6 +3849,20 @@ GetLockmodeName(LOCKMETHODID lockmethodid, LOCKMODE mode)
 	return LockMethods[lockmethodid]->lockModeNames[mode];
 }
 
+void
+EnumerateLocks(LockIterator iterator, void* arg)
+{
+	PROCLOCK   *proclock;
+	HASH_SEQ_STATUS status;
+
+	hash_seq_init(&status, LockMethodProcLockHash);
+
+	while ((proclock = (PROCLOCK *) hash_seq_search(&status)) != NULL)
+	{
+		iterator(proclock, arg);
+	}
+}
+
 #ifdef LOCK_DEBUG
 /*
  * Dump all locks in the given proc's myProcLocks lists.

src/backend/storage/lmgr/predicate.c

@@ -4011,7 +4011,7 @@ CheckForSerializableConflictOut(bool visible, Relation relation,
 			xid = InvalidTransactionId;
 	}
 	Assert(TransactionIdIsValid(xid));
-	Assert(TransactionIdFollowsOrEquals(xid, TransactionXmin));
+	//Assert(TransactionIdFollowsOrEquals(xid, TransactionXmin));
 
 	/*
 	 * Find top level xid.  Bail out if xid is too early to be a conflict, or

src/backend/storage/lmgr/proc.c

@@ -1449,6 +1449,22 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
 											   "Processes holding the lock: %s. Wait queue: %s.",
 											   lockHoldersNum, lock_holders_sbuf.data, lock_waiters_sbuf.data))));
 			}
+			else if (deadlock_state == DS_DISTRIBUTED_DEADLOCK)
+			{
+				/*
+				 * This message is a bit redundant with the error that will be
+				 * reported subsequently, but in some cases the error report
+				 * might not make it to the log (eg, if it's caught by an
+				 * exception handler), and we want to ensure all long-wait
+				 * events get logged.
+				 */
+				ereport(LOG,
+						(errmsg("process %d detected distributed deadlock while waiting for %s on %s after %ld.%03d ms",
+								MyProcPid, modename, buf.data, msecs, usecs),
+						 (errdetail_log_plural("Process holding the lock: %s. Wait queue: %s.",
+						   "Processes holding the lock: %s. Wait queue: %s.",
+											   lockHoldersNum, lock_holders_sbuf.data, lock_waiters_sbuf.data))));
+			}
 
 			if (myWaitStatus == STATUS_WAITING)
 				ereport(LOG,
@@ -1473,7 +1489,7 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
 				 * future-proofing, print a message if it looks like someone
 				 * else kicked us off the lock.
 				 */
-				if (deadlock_state != DS_HARD_DEADLOCK)
+				if (deadlock_state != DS_HARD_DEADLOCK && deadlock_state != DS_DISTRIBUTED_DEADLOCK)
 					ereport(LOG,
 							(errmsg("process %d failed to acquire %s on %s after %ld.%03d ms",
 									MyProcPid, modename, buf.data, msecs, usecs),
@@ -1693,7 +1709,7 @@ CheckDeadLock(void)
 	/* Run the deadlock check, and set deadlock_state for use by ProcSleep */
 	deadlock_state = DeadLockCheck(MyProc);
 
-	if (deadlock_state == DS_HARD_DEADLOCK)
+	if (deadlock_state == DS_HARD_DEADLOCK || deadlock_state == DS_DISTRIBUTED_DEADLOCK)
 	{
 		/*
 		 * Oops.  We have a deadlock.