[PGPRO-4074] Additional mm XACT_EVENTs and distributed deadlock hook.

This is finely selected cherry pick of dec7a31cadc9048. I've removed xtm because
we don't need almost all fields. ATX save-resume hook is not ported but will be
in the future once ATX lands itself. Parallel workers serialize-deserialize hook
is not ported because currently I see no value for mm in it; long time ago it
was used for importing csn snapshots.

Unused XACT_EVENTs are not ported as well.

tags: multimaster
(cherry picked from commit ec1bf5004833e95a18da8e05443a4f6e308d1fb6)

Author: arssher

contrib/postgres_fdw/connection.c

@@ -713,6 +713,16 @@ pgfdw_xact_callback(XactEvent event, void *arg)
 	HASH_SEQ_STATUS scan;
 	ConnCacheEntry *entry;
 
+	/* Do nothing for these (multimaster support) events */
+	switch (event)
+	{
+		case XACT_EVENT_START:
+		case XACT_EVENT_COMMIT_COMMAND:
+			return;
+		default:
+			break;
+	}
+
 	/* Quick exit if no connections were touched in this transaction. */
 	if (!xact_got_connection)
 		return;
@@ -861,6 +871,9 @@ pgfdw_xact_callback(XactEvent event, void *arg)
 					/* Disarm changing_xact_state if it all worked. */
 					entry->changing_xact_state = abort_cleanup_failure;
 					break;
+				case XACT_EVENT_START:
+				case XACT_EVENT_COMMIT_COMMAND:
+					Assert(false);
 			}
 		}
 

src/backend/access/transam/xact.c

@@ -310,7 +310,6 @@ static void AtCommit_Memory(void);
 static void AtStart_Cache(void);
 static void AtStart_Memory(void);
 static void AtStart_ResourceOwner(void);
-static void CallXactCallbacks(XactEvent event);
 static void CallSubXactCallbacks(SubXactEvent event,
 								 SubTransactionId mySubid,
 								 SubTransactionId parentSubid);
@@ -2827,6 +2826,7 @@ StartTransactionCommand(void)
 		case TBLOCK_DEFAULT:
 			StartTransaction();
 			s->blockState = TBLOCK_STARTED;
+			CallXactCallbacks(XACT_EVENT_START);
 			break;
 
 			/*
@@ -2919,6 +2919,8 @@ CommitTransactionCommand(void)
 {
 	TransactionState s = CurrentTransactionState;
 
+	CallXactCallbacks(XACT_EVENT_COMMIT_COMMAND);
+
 	if (s->chain)
 		SaveTransactionCharacteristics();
 
@@ -3536,7 +3538,7 @@ UnregisterXactCallback(XactCallback callback, void *arg)
 	}
 }
 
-static void
+void
 CallXactCallbacks(XactEvent event)
 {
 	XactCallbackItem *item;
@@ -5283,6 +5285,7 @@ StartParallelWorkerTransaction(char *tstatespace)
 
 	Assert(CurrentTransactionState->blockState == TBLOCK_DEFAULT);
 	StartTransaction();
+	CallXactCallbacks(XACT_EVENT_START);
 
 	tstate = (SerializedTransactionState *) tstatespace;
 	XactIsoLevel = tstate->xactIsoLevel;

src/backend/storage/lmgr/deadlock.c

@@ -32,6 +32,7 @@
 #include "storage/proc.h"
 #include "utils/memutils.h"
 
+bool (*DetectGlobalDeadLock) (PGPROC *proc) = NULL;
 
 /*
  * One edge in the waits-for graph.
@@ -280,6 +281,8 @@ DeadLockCheck(PGPROC *proc)
 		return DS_SOFT_DEADLOCK;
 	else if (blocking_autovacuum_proc != NULL)
 		return DS_BLOCKED_BY_AUTOVACUUM;
+	else if (DetectGlobalDeadLock && DetectGlobalDeadLock(proc))
+		return DS_DISTRIBUTED_DEADLOCK;
 	else
 		return DS_NO_DEADLOCK;
 }

src/backend/storage/lmgr/proc.c

@@ -1486,6 +1486,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 a 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,
@@ -1510,7 +1526,8 @@ 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),
@@ -1727,7 +1744,8 @@ 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.

src/include/access/xact.h

@@ -108,14 +108,16 @@ extern int	MyXactFlags;
  */
 typedef enum
 {
+	XACT_EVENT_START,
 	XACT_EVENT_COMMIT,
 	XACT_EVENT_PARALLEL_COMMIT,
 	XACT_EVENT_ABORT,
 	XACT_EVENT_PARALLEL_ABORT,
 	XACT_EVENT_PREPARE,
 	XACT_EVENT_PRE_COMMIT,
 	XACT_EVENT_PARALLEL_PRE_COMMIT,
-	XACT_EVENT_PRE_PREPARE
+	XACT_EVENT_PRE_PREPARE,
+	XACT_EVENT_COMMIT_COMMAND
 } XactEvent;
 
 typedef void (*XactCallback) (XactEvent event, void *arg);
@@ -459,5 +461,6 @@ extern void ParsePrepareRecord(uint8 info, xl_xact_prepare *xlrec, xl_xact_parse
 extern void EnterParallelMode(void);
 extern void ExitParallelMode(void);
 extern bool IsInParallelMode(void);
+extern void CallXactCallbacks(XactEvent event);
 
 #endif							/* XACT_H */

src/include/storage/lock.h

@@ -500,8 +500,9 @@ typedef enum
 	DS_NO_DEADLOCK,				/* no deadlock detected */
 	DS_SOFT_DEADLOCK,			/* deadlock avoided by queue rearrangement */
 	DS_HARD_DEADLOCK,			/* deadlock, no way out but ERROR */
-	DS_BLOCKED_BY_AUTOVACUUM	/* no deadlock; queue blocked by autovacuum
+	DS_BLOCKED_BY_AUTOVACUUM,	/* no deadlock; queue blocked by autovacuum
 								 * worker */
+	DS_DISTRIBUTED_DEADLOCK		/* distributed deadlock detected by DTM */
 } DeadLockState;
 
 /*

src/include/storage/proc.h

@@ -296,6 +296,8 @@ extern PGDLLIMPORT int LockTimeout;
 extern PGDLLIMPORT int IdleInTransactionSessionTimeout;
 extern bool log_lock_waits;
 
+/* hook for extensions to catch distributed deadlock */
+extern PGDLLIMPORT bool (*DetectGlobalDeadLock) (PGPROC *proc);
 
 /*
  * Function Prototypes