From f6350fe3f7a8b989c8fa06d47251cd0930ca9b8e Mon Sep 17 00:00:00 2001
From: Robert Haas <rhaas@postgresql.org>
Date: Tue, 28 Apr 2015 17:23:47 -0400
Subject: [PATCH 2/2] Lock management changes for parallelism.

This part still needs more work.
---
 src/backend/access/transam/README.parallel |   70 ++++++++-
 src/backend/access/transam/parallel.c      |   22 ++-
 src/backend/access/transam/xact.c          |    4 +
 src/backend/storage/ipc/standby.c          |    6 +-
 src/backend/storage/lmgr/lock.c            |  233 +++++++++++++++++++++++++++-
 src/include/storage/lock.h                 |   10 +-
 6 files changed, 333 insertions(+), 12 deletions(-)

diff --git a/src/backend/access/transam/README.parallel b/src/backend/access/transam/README.parallel
index 1005186..2257e4c 100644
--- a/src/backend/access/transam/README.parallel
+++ b/src/backend/access/transam/README.parallel
@@ -122,9 +122,8 @@ worker.  This includes:
     values are restored, this incidentally sets SessionUserId and OuterUserId
     to the correct values.  This final step restores CurrentUserId.
 
-To prevent undetected or unprincipled deadlocks when running in parallel mode,
-this could should eventually handle heavyweight locks in some way.  This is
-not implemented yet.
+  - Locks on relations and database objects, and any lock held on the
+    transaction's XID.  See lock management, below.
 
 Transaction Integration
 =======================
@@ -188,6 +187,71 @@ importantly:
     safely access the initiating backend's pg_temp namespace, and should
     not create one of their own.
 
+Lock Management
+===============
+
+Certain heavyweight locks that the initiating backend holds at the beginning
+of parallelism are copied to each worker, which unconditionally acquires them.
+The parallel backends acquire - without waiting - each such lock that the
+leader holds, even if that lock is self-exclusive.  This creates the unusual
+situation that a lock which could normally only be held by a single backend
+can be shared among several backends in a parallel group.
+
+Obviously, this presents significant hazards that would not be present in
+normal execution.  If, for example, a backend were to initiate parallelism
+while ReindexIsProcessingIndex() were true for some index, the parallel
+backends launched at that time would neither share this state nor be excluded
+from accessing the index via the heavyweight lock mechanism.  It is therefore
+imperative that backends only initiate parallelism from places where it will
+be safe for parallel workers to access the relations on which they hold locks.
+It is also important that they not afterwards do anything which causes access
+to those relations to become unsafe, or at least not until after parallelism
+has concluded.  The fact that parallelism is strictly read-only means that the
+opportunities for such mishaps are few and far between; furthermore, most
+operations which present these hazards are DDL operations, which will be
+rejected by CheckTableNotInUse() if parallel mode is active.
+
+Only relation locks, locks on database or shared objects, and any lock on
+our transaction ID are copied to workers.  Advisory locks are not copied,
+but the leader may hold them at the start of parallelism; they cannot
+subsequently be manipulated while parallel mode is active.  It is not safe
+to attempt parallelism while holding a lock of any other type, such as a
+page, tuple, or relation extension lock, and such attempts will fail.  Such
+locks can be taken and released during parallel mode; they merely cannot be
+held at the start of parallel mode, since we would then fail to provide
+necessary mutual exclusion.
+
+Copying locks to workers is important for the avoidance of undetected
+deadlock between the initiating process and its parallel workers.  If the
+initiating process holds a lock on an object at the start of parallelism,
+and the worker subsequently attempts to acquire a lock on that same object
+and blocks, this will result in an undetected deadlock, because the
+initiating process cannot finish the transaction (thus releasing the lock)
+until the worker terminates, and the worker cannot acquire the lock while
+the initiating process holds it.  Locks which the processes involved acquire
+and then release during parallelism do not present this hazard; they simply
+force the processes involved to take turns accessing the protected resource.
+
+Locks which the processes involved acquire and retain *do* present this
+hazard.  In general, we can avoid this problem by forbidding any of the
+group of parallel processes from holding any heavyweight locks at the end of
+parallelism which they did not hold at the start of parallelism.  For now,
+we take this approach; if it proves to be onerous in the future, we can look
+for ways to relax it then.
+
+Copying locks to workers is also important for the avoidance of undetected
+deadlock involving both the parallel processes and other processes.  For
+example, suppose processes A1 and A2 are cooperating parallel processes and
+B is an unrelated process.  Suppose A1 holds a lock L1 and waits for A2 to
+finish; B holds a lock L2 and blocks waiting for L1; and A2 blocks waiting
+for L2.  Without lock copying, the waits-for graph is A2 -> B -> A1; there
+is no cycle.  With lock copying, A2 will also hold the lock on L1 and the
+deadlock detector can find the cycle A2 -> B -> A2.  As in the case of
+deadlocks within the parallel group, undetected deadlock occur if either A1
+or A2 acquired a lock after the start of parallelism and attempted to
+retain it beyond the end of parallelism.  The prohibitions discussed above
+protect us against this case.
+
 Coding Conventions
 ===================
 
diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c
index 8ed7314..d2cec8e 100644
--- a/src/backend/access/transam/parallel.c
+++ b/src/backend/access/transam/parallel.c
@@ -56,7 +56,8 @@
 #define PARALLEL_KEY_TRANSACTION_SNAPSHOT	UINT64CONST(0xFFFFFFFFFFFF0006)
 #define PARALLEL_KEY_ACTIVE_SNAPSHOT		UINT64CONST(0xFFFFFFFFFFFF0007)
 #define PARALLEL_KEY_TRANSACTION_STATE		UINT64CONST(0xFFFFFFFFFFFF0008)
-#define PARALLEL_KEY_EXTENSION_TRAMPOLINE	UINT64CONST(0xFFFFFFFFFFFF0009)
+#define PARALLEL_KEY_LOCK					UINT64CONST(0xFFFFFFFFFFFF0009)
+#define PARALLEL_KEY_EXTENSION_TRAMPOLINE	UINT64CONST(0xFFFFFFFFFFFF000A)
 
 /* Fixed-size parallel state. */
 typedef struct FixedParallelState
@@ -191,6 +192,7 @@ InitializeParallelDSM(ParallelContext *pcxt)
 	Size	tsnaplen = 0;
 	Size	asnaplen = 0;
 	Size	tstatelen = 0;
+	Size	lockstatelen = 0;
 	Size	segsize = 0;
 	int		i;
 	FixedParallelState *fps;
@@ -223,8 +225,10 @@ InitializeParallelDSM(ParallelContext *pcxt)
 		shm_toc_estimate_chunk(&pcxt->estimator, asnaplen);
 		tstatelen = EstimateTransactionStateSpace();
 		shm_toc_estimate_chunk(&pcxt->estimator, tstatelen);
+		lockstatelen = EstimateLockStateSpace();
+		shm_toc_estimate_chunk(&pcxt->estimator, lockstatelen);
 		/* If you add more chunks here, you probably need to add keys. */
-		shm_toc_estimate_keys(&pcxt->estimator, 6);
+		shm_toc_estimate_keys(&pcxt->estimator, 7);
 
 		/* Estimate space need for error queues. */
 		StaticAssertStmt(BUFFERALIGN(PARALLEL_ERROR_QUEUE_SIZE) ==
@@ -295,6 +299,7 @@ InitializeParallelDSM(ParallelContext *pcxt)
 		char   *tsnapspace;
 		char   *asnapspace;
 		char   *tstatespace;
+		char   *lockstatespace;
 		char   *error_queue_space;
 
 		/* Serialize shared libraries we have loaded. */
@@ -326,6 +331,11 @@ InitializeParallelDSM(ParallelContext *pcxt)
 		SerializeTransactionState(tstatelen, tstatespace);
 		shm_toc_insert(pcxt->toc, PARALLEL_KEY_TRANSACTION_STATE, tstatespace);
 
+		/* Serialize heavyweight locks. */
+		lockstatespace = shm_toc_allocate(pcxt->toc, lockstatelen);
+		SerializeLockState(lockstatelen, lockstatespace);
+		shm_toc_insert(pcxt->toc, PARALLEL_KEY_LOCK, lockstatespace);
+
 		/* Allocate space for worker information. */
 		pcxt->worker = palloc0(sizeof(ParallelWorkerInfo) * pcxt->nworkers);
 
@@ -452,6 +462,8 @@ LaunchParallelWorkers(ParallelContext *pcxt)
 void
 WaitForParallelWorkersToFinish(ParallelContext *pcxt)
 {
+	CheckForRetainedParallelLocks();
+
 	for (;;)
 	{
 		bool	anyone_alive = false;
@@ -807,6 +819,7 @@ ParallelWorkerMain(Datum main_arg)
 	char   *tsnapspace;
 	char   *asnapspace;
 	char   *tstatespace;
+	char   *lockstatespace;
 	StringInfoData	msgbuf;
 
 	/* Establish signal handlers. */
@@ -926,6 +939,11 @@ ParallelWorkerMain(Datum main_arg)
 	/* Restore user ID and security context. */
 	SetUserIdAndSecContext(fps->current_user_id, fps->sec_context);
 
+	/* Restore locks. */
+	lockstatespace = shm_toc_lookup(toc, PARALLEL_KEY_LOCK);
+	Assert(lockstatespace != NULL);
+	RestoreLockState(fps->parallel_master_pgproc, lockstatespace);
+
 	/*
 	 * We've initialized all of our state now; nothing should change hereafter.
 	 */
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index ebc232a..f8b529e 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -893,6 +893,8 @@ ExitParallelMode(void)
 	Assert(s->parallelModeLevel > 1 || !ParallelContextActive());
 
 	--s->parallelModeLevel;
+
+	CheckForRetainedParallelLocks();
 }
 
 /*
@@ -1950,6 +1952,7 @@ CommitTransaction(void)
 	/* If we might have parallel workers, clean them up now. */
 	if (IsInParallelMode())
 	{
+		CheckForRetainedParallelLocks();
 		AtEOXact_Parallel(true);
 		s->parallelModeLevel = 0;
 	}
@@ -2483,6 +2486,7 @@ AbortTransaction(void)
 	/* If in parallel mode, clean up workers and exit parallel mode. */
 	if (IsInParallelMode())
 	{
+		ForgetParallelLocks();
 		AtEOXact_Parallel(false);
 		s->parallelModeLevel = 0;
 	}
diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c
index 292bed5..6002d51 100644
--- a/src/backend/storage/ipc/standby.c
+++ b/src/backend/storage/ipc/standby.c
@@ -366,7 +366,8 @@ ResolveRecoveryConflictWithLock(Oid dbOid, Oid relOid)
 		ResolveRecoveryConflictWithVirtualXIDs(backends,
 											 PROCSIG_RECOVERY_CONFLICT_LOCK);
 
-		if (LockAcquireExtended(&locktag, AccessExclusiveLock, true, true, false)
+		if (LockAcquireExtended(&locktag, AccessExclusiveLock,
+								true, true, false, false)
 			!= LOCKACQUIRE_NOT_AVAIL)
 			lock_acquired = true;
 	}
@@ -592,7 +593,8 @@ StandbyAcquireAccessExclusiveLock(TransactionId xid, Oid dbOid, Oid relOid)
 	 */
 	SET_LOCKTAG_RELATION(locktag, newlock->dbOid, newlock->relOid);
 
-	if (LockAcquireExtended(&locktag, AccessExclusiveLock, true, true, false)
+	if (LockAcquireExtended(&locktag, AccessExclusiveLock,
+							true, true, false, false)
 		== LOCKACQUIRE_NOT_AVAIL)
 		ResolveRecoveryConflictWithLock(newlock->dbOid, newlock->relOid);
 }
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index 1eb2d4b..b19bffa 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -35,6 +35,7 @@
 #include "access/transam.h"
 #include "access/twophase.h"
 #include "access/twophase_rmgr.h"
+#include "access/xact.h"
 #include "access/xlog.h"
 #include "miscadmin.h"
 #include "pg_trace.h"
@@ -258,6 +259,13 @@ static LOCALLOCK *StrongLockInProgress;
 static LOCALLOCK *awaitedLock;
 static ResourceOwner awaitedOwner;
 
+/*
+ * Track locks acquired in parallel mode.  Any such locks must be released
+ * before exiting parallel mode; see src/backend/access/transam/README.parallel
+ */
+static LOCALLOCKTAG *LocksAcquiredInParallelMode;
+static int nLocksAcquiredInParallelMode;
+static int maxLocksAcquiredInParallelMode;
 
 #ifdef LOCK_DEBUG
 
@@ -669,7 +677,8 @@ LockAcquire(const LOCKTAG *locktag,
 			bool sessionLock,
 			bool dontWait)
 {
-	return LockAcquireExtended(locktag, lockmode, sessionLock, dontWait, true);
+	return LockAcquireExtended(locktag, lockmode, sessionLock, dontWait, true,
+							   NULL);
 }
 
 /*
@@ -680,13 +689,20 @@ LockAcquire(const LOCKTAG *locktag,
  * caller to note that the lock table is full and then begin taking
  * extreme action to reduce the number of other lock holders before
  * retrying the action.
+ *
+ * leader_proc should be false except for the case of a parallel worker
+ * reacquiring locks already held by the parallel group leader.  In that
+ * case, we never log the lock acquisition since the parent has already
+ * done it; and more importantly and surprisingly, we ignore lock conflicts.
+ * See src/backend/access/transam/README.parallel for further discussion.
  */
 LockAcquireResult
 LockAcquireExtended(const LOCKTAG *locktag,
 					LOCKMODE lockmode,
 					bool sessionLock,
 					bool dontWait,
-					bool reportMemoryError)
+					bool reportMemoryError,
+					PGPROC *leader_proc)
 {
 	LOCKMETHODID lockmethodid = locktag->locktag_lockmethodid;
 	LockMethod	lockMethodTable;
@@ -700,6 +716,7 @@ LockAcquireExtended(const LOCKTAG *locktag,
 	LWLock	   *partitionLock;
 	int			status;
 	bool		log_lock = false;
+	bool		remember_parallel_lock = false;
 
 	if (lockmethodid <= 0 || lockmethodid >= lengthof(LockMethods))
 		elog(ERROR, "unrecognized lock method: %d", lockmethodid);
@@ -797,13 +814,42 @@ LockAcquireExtended(const LOCKTAG *locktag,
 	if (lockmode >= AccessExclusiveLock &&
 		locktag->locktag_type == LOCKTAG_RELATION &&
 		!RecoveryInProgress() &&
-		XLogStandbyInfoActive())
+		XLogStandbyInfoActive() && leader_proc == NULL)
 	{
 		LogAccessExclusiveLockPrepare();
 		log_lock = true;
 	}
 
 	/*
+	 * If we're in parallel mode, ensure that there will be space to
+	 * remember this lock, so that we can later check that it got released
+	 * before the end of parallelism.  We can skip this when reacquiring the
+	 * parallel leader's locks, as those don't have to be released before
+	 * parallelism ends.  See src/backend/access/transam/README.parallel for
+	 * more details.
+	 */
+	if (leader_proc == NULL && IsInParallelMode())
+	{
+		if (maxLocksAcquiredInParallelMode == 0)
+		{
+			LocksAcquiredInParallelMode =
+				MemoryContextAlloc(TopMemoryContext, 8 * sizeof(LOCALLOCKTAG));
+			maxLocksAcquiredInParallelMode = 8;
+		}
+		else if (nLocksAcquiredInParallelMode >=
+					maxLocksAcquiredInParallelMode)
+		{
+			LocksAcquiredInParallelMode =
+				repalloc(LocksAcquiredInParallelMode,
+						 2 * maxLocksAcquiredInParallelMode *
+						 sizeof(LOCALLOCKTAG));
+			maxLocksAcquiredInParallelMode *= 2;
+		}
+		Assert(nLocksAcquiredInParallelMode < maxLocksAcquiredInParallelMode);
+		remember_parallel_lock = true;
+	}
+
+	/*
 	 * Attempt to take lock via fast path, if eligible.  But if we remember
 	 * having filled up the fast path array, we don't attempt to make any
 	 * further use of it until we release some locks.  It's possible that some
@@ -842,6 +888,9 @@ LockAcquireExtended(const LOCKTAG *locktag,
 			locallock->lock = NULL;
 			locallock->proclock = NULL;
 			GrantLockLocal(locallock, owner);
+			if (remember_parallel_lock)
+				LocksAcquiredInParallelMode[nLocksAcquiredInParallelMode++] =
+					localtag;
 			return LOCKACQUIRE_OK;
 		}
 	}
@@ -910,7 +959,8 @@ LockAcquireExtended(const LOCKTAG *locktag,
 	/*
 	 * If lock requested conflicts with locks requested by waiters, must join
 	 * wait queue.  Otherwise, check for conflict with already-held locks.
-	 * (That's last because most complex check.)
+	 * (That's last because most complex check.)  Parallel reacquire never
+	 * conflicts.
 	 */
 	if (lockMethodTable->conflictTab[lockmode] & lock->waitMask)
 		status = STATUS_FOUND;
@@ -918,6 +968,26 @@ LockAcquireExtended(const LOCKTAG *locktag,
 		status = LockCheckConflicts(lockMethodTable, lockmode,
 									lock, proclock);
 
+	/*
+	 * When reacquiring the master's locks in a parallel worker, we ignore
+	 * lock conflicts; in effect, the parallel leader and the worker are
+	 * sharing the lock.  However, if the leader somehow manages to die before
+	 * we reach this point, then we wouldn't be sharing the lock with the
+	 * parallel group leader - we'd just be grabbing it in the face of
+	 * conflicts.  To make sure that can't happen, check that the leader
+	 * still has the lock.
+	 */
+	if (status == STATUS_FOUND && leader_proc != NULL)
+	{
+		PROCLOCKTAG	leadertag;
+
+		leadertag.myLock = lock;
+		leadertag.myProc = MyProc;
+		if (hash_search(LockMethodProcLockHash, (void *) &leadertag,
+						HASH_FIND, NULL))
+			status = STATUS_OK;
+	}
+
 	if (status == STATUS_OK)
 	{
 		/* No conflict with held or previously requested locks */
@@ -1034,6 +1104,8 @@ LockAcquireExtended(const LOCKTAG *locktag,
 							   locktag->locktag_field2);
 	}
 
+	if (remember_parallel_lock)
+		LocksAcquiredInParallelMode[nLocksAcquiredInParallelMode++] = localtag;
 	return LOCKACQUIRE_OK;
 }
 
@@ -1818,6 +1890,38 @@ LockRelease(const LOCKTAG *locktag, LOCKMODE lockmode, bool sessionLock)
 	}
 
 	/*
+	 * If we're tracking locks acquired in parallel mode and we're about
+	 * to release the last copy of this lock, remove it from our tracking
+	 * table.  We search back-to-front since we'll probably release the last
+	 * lock acquired first.
+	 */
+	if (IsInParallelMode() && locallock->nLocks == 1)
+	{
+		int		n;
+		bool	found = false;
+
+		for (n = nLocksAcquiredInParallelMode - 1; n >= 0; --n)
+		{
+			LOCALLOCKTAG	*ptag = &LocksAcquiredInParallelMode[n];
+
+			if (memcmp(ptag, &localtag, sizeof(LOCALLOCKTAG)) == 0)
+			{
+				--nLocksAcquiredInParallelMode;
+				if (n < nLocksAcquiredInParallelMode)
+					memmove(ptag, ptag + 1, sizeof(LOCALLOCKTAG) *
+							(nLocksAcquiredInParallelMode - n));
+				found = true;
+				break;
+			}
+		}
+
+		if (!found)
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_TRANSACTION_STATE),
+					 errmsg("cannot release locks acquired before the start of parallelism while parallelism is active")));
+	}
+
+	/*
 	 * Decrease the total local count.  If we're still holding the lock, we're
 	 * done.
 	 */
@@ -1943,6 +2047,7 @@ LockReleaseAll(LOCKMETHODID lockmethodid, bool allLocks)
 	if (lockmethodid <= 0 || lockmethodid >= lengthof(LockMethods))
 		elog(ERROR, "unrecognized lock method: %d", lockmethodid);
 	lockMethodTable = LockMethods[lockmethodid];
+	nLocksAcquiredInParallelMode = 0;
 
 #ifdef LOCK_DEBUG
 	if (*(lockMethodTable->trace_flag))
@@ -3557,6 +3662,126 @@ GetLockmodeName(LOCKMETHODID lockmethodid, LOCKMODE mode)
 	return LockMethods[lockmethodid]->lockModeNames[mode];
 }
 
+/*
+ * Estimate the amount of space required to record information on locks that
+ * need to be copied to parallel workers.
+ */
+Size
+EstimateLockStateSpace(void)
+{
+	return add_size(sizeof(long),
+					mul_size(hash_get_num_entries(LockMethodLocalHash),
+							 sizeof(LOCALLOCKTAG)));
+}
+
+/*
+ * Serialize relevant heavyweight lock state into the memory beginning at
+ * start_address.  maxsize should be at least as large as the value returned
+ * by EstimateLockStateSpace.
+ */
+void
+SerializeLockState(Size maxsize, char *start_address)
+{
+	char	   *endptr = start_address + maxsize;
+	char	   *curptr = start_address + sizeof(long);
+	HASH_SEQ_STATUS status;
+	LOCALLOCK  *locallock;
+	long		count = 0;
+
+	hash_seq_init(&status, LockMethodLocalHash);
+
+	while ((locallock = (LOCALLOCK *) hash_seq_search(&status)) != NULL)
+	{
+		LockTagType	type;
+
+		if (locallock->nLocks == 0)
+			continue;
+
+		/*
+		 * We only copy ordinary heavyweight locks; advisory lock operations
+		 * are prohibited while in parallel mode.
+		 */
+		if (locallock->tag.lock.locktag_lockmethodid != DEFAULT_LOCKMETHOD)
+			continue;
+
+		type = locallock->tag.lock.locktag_type;
+
+		/*
+		 * Let's make sure that the parallel leader isn't holding some kind
+		 * of lock we're not expecting, like a relation extension lock, or
+		 * a page or tuple lock.  If we are, this is probably not a safe point
+		 * from which to initiate parallelism.
+		 */
+		if (type == LOCKTAG_TRANSACTION)
+		{
+			if (locallock->tag.mode != ExclusiveLock)
+				elog(ERROR, "unexpected non-exclusive transaction lock");
+		}
+		else if (type != LOCKTAG_RELATION && type != LOCKTAG_OBJECT)
+			elog(ERROR, "unexpected lock tag type: %d", (int) type);
+
+		/* Double-check that we're not going to overrun the space. */
+		if (curptr >= endptr)
+			elog(ERROR, "not enough space to serialize lock state");
+
+		memcpy(curptr, &locallock->tag, sizeof(LOCALLOCKTAG));
+		curptr += sizeof(LOCALLOCKTAG);
+		count++;
+	}
+
+	memcpy(start_address, &count, sizeof(long));
+}
+
+/*
+ * Retake the locals specified by the serialized lock state.
+ */
+void
+RestoreLockState(PGPROC *leader_proc, char *start_address)
+{
+	char	   *curptr = start_address + sizeof(long);
+	long		count;
+
+	Assert(leader_proc != NULL);
+	memcpy(&count, start_address, sizeof(long));
+
+	while (count > 0)
+	{
+		LOCALLOCKTAG	locallocktag;
+		LockAcquireResult	result;
+
+		memcpy(&locallocktag, curptr, sizeof(LOCALLOCKTAG));
+		curptr += sizeof(LOCALLOCKTAG);
+		--count;
+
+		result = LockAcquireExtended(&locallocktag.lock, locallocktag.mode,
+									 false, true, true, leader_proc);
+		if (result != LOCKACQUIRE_OK)
+			ereport(ERROR,
+					(errmsg("parallel worker lock not available")));
+	}
+}
+
+/*
+ * Verify that no locks acquired while in parallel mode are still held.
+ */
+void
+CheckForRetainedParallelLocks(void)
+{
+	if (nLocksAcquiredInParallelMode > 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_TRANSACTION_STATE),
+				 errmsg("cannot retain locks acquired while in parallel mode")));
+}
+
+/*
+ * Stop tracking locks acquired in parallel mode.
+ */
+void
+ForgetParallelLocks(void)
+{
+	nLocksAcquiredInParallelMode = 0;
+}
+
 #ifdef LOCK_DEBUG
 /*
  * Dump all locks in the given proc's myProcLocks lists.
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index dae517f..80144b9 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -503,7 +503,8 @@ extern LockAcquireResult LockAcquireExtended(const LOCKTAG *locktag,
 					LOCKMODE lockmode,
 					bool sessionLock,
 					bool dontWait,
-					bool report_memory_error);
+					bool report_memory_error,
+					PGPROC *leader_proc);
 extern void AbortStrongLockAcquire(void);
 extern bool LockRelease(const LOCKTAG *locktag,
 			LOCKMODE lockmode, bool sessionLock);
@@ -564,4 +565,11 @@ extern void VirtualXactLockTableInsert(VirtualTransactionId vxid);
 extern void VirtualXactLockTableCleanup(void);
 extern bool VirtualXactLock(VirtualTransactionId vxid, bool wait);
 
+/* Parallel worker state sharing. */
+extern Size EstimateLockStateSpace(void);
+extern void SerializeLockState(Size maxsize, char *start_address);
+extern void RestoreLockState(PGPROC *leader_proc, char *start_address);
+extern void CheckForRetainedParallelLocks(void);
+extern void ForgetParallelLocks(void);
+
 #endif   /* LOCK_H */
-- 
1.7.9.6 (Apple Git-31.1)

