Add WaitForLockers in lmgr, refactoring index.c code

This is in support of a future REINDEX CONCURRENTLY feature.

Michael Paquier

src/backend/catalog/index.c

@@ -1323,7 +1323,6 @@ index_drop(Oid indexId, bool concurrent)
 				indexrelid;
 	LOCKTAG		heaplocktag;
 	LOCKMODE	lockmode;
-	VirtualTransactionId *old_lockholders;
 
 	/*
 	 * To drop an index safely, we must grab exclusive lock on its parent
@@ -1445,11 +1444,8 @@ index_drop(Oid indexId, bool concurrent)
 
 		/*
 		 * Now we must wait until no running transaction could be using the
-		 * index for a query.  To do this, inquire which xacts currently would
+		 * index for a query. Note we do not need to worry about xacts that
-		 * conflict with AccessExclusiveLock on the table -- ie, which ones
+		 * open the table for reading after this point; they will see the
-		 * have a lock of any kind on the table. Then wait for each of these
-		 * xacts to commit or abort. Note we do not need to worry about xacts
-		 * that open the table for reading after this point; they will see the
 		 * index as invalid when they open the relation.
 		 *
 		 * Note: the reason we use actual lock acquisition here, rather than
@@ -1457,18 +1453,8 @@ index_drop(Oid indexId, bool concurrent)
 		 * possible if one of the transactions in question is blocked trying
 		 * to acquire an exclusive lock on our table.  The lock code will
 		 * detect deadlock and error out properly.
-		 *
-		 * Note: GetLockConflicts() never reports our own xid, hence we need
-		 * not check for that.	Also, prepared xacts are not reported, which
-		 * is fine since they certainly aren't going to do anything more.
 		 */
-		old_lockholders = GetLockConflicts(&heaplocktag, AccessExclusiveLock);
+		WaitForLockers(heaplocktag, AccessExclusiveLock);
-
-		while (VirtualTransactionIdIsValid(*old_lockholders))
-		{
-			VirtualXactLock(*old_lockholders, true);
-			old_lockholders++;
-		}
 
 		/*
 		 * No more predicate locks will be acquired on this index, and we're
@@ -1510,15 +1496,9 @@ index_drop(Oid indexId, bool concurrent)
 
 		/*
 		 * Wait till every transaction that saw the old index state has
-		 * finished.  The logic here is the same as above.
+		 * finished.
 		 */
-		old_lockholders = GetLockConflicts(&heaplocktag, AccessExclusiveLock);
+		WaitForLockers(heaplocktag, AccessExclusiveLock);
-
-		while (VirtualTransactionIdIsValid(*old_lockholders))
-		{
-			VirtualXactLock(*old_lockholders, true);
-			old_lockholders++;
-		}
 
 		/*
 		 * Re-open relations to allow us to complete our actions.

src/backend/commands/indexcmds.c

@@ -321,7 +321,6 @@ DefineIndex(IndexStmt *stmt,
 	IndexInfo  *indexInfo;
 	int			numberOfAttributes;
 	TransactionId limitXmin;
-	VirtualTransactionId *old_lockholders;
 	VirtualTransactionId *old_snapshots;
 	int			n_old_snapshots;
 	LockRelId	heaprelid;
@@ -652,30 +651,17 @@ DefineIndex(IndexStmt *stmt,
 	 * for an overview of how this works)
 	 *
 	 * Now we must wait until no running transaction could have the table open
-	 * with the old list of indexes.  To do this, inquire which xacts
+	 * with the old list of indexes. Note we do not need to worry about xacts
-	 * currently would conflict with ShareLock on the table -- ie, which ones
+	 * that open the table for writing after this point; they will see the new
-	 * have a lock that permits writing the table.	Then wait for each of
+	 * index when they open it.
-	 * these xacts to commit or abort.	Note we do not need to worry about
-	 * xacts that open the table for writing after this point; they will see
-	 * the new index when they open it.
 	 *
 	 * Note: the reason we use actual lock acquisition here, rather than just
 	 * checking the ProcArray and sleeping, is that deadlock is possible if
 	 * one of the transactions in question is blocked trying to acquire an
 	 * exclusive lock on our table.  The lock code will detect deadlock and
 	 * error out properly.
-	 *
-	 * Note: GetLockConflicts() never reports our own xid, hence we need not
-	 * check for that.	Also, prepared xacts are not reported, which is fine
-	 * since they certainly aren't going to do anything more.
 	 */
-	old_lockholders = GetLockConflicts(&heaplocktag, ShareLock);
+	WaitForLockers(heaplocktag, ShareLock);
-
-	while (VirtualTransactionIdIsValid(*old_lockholders))
-	{
-		VirtualXactLock(*old_lockholders, true);
-		old_lockholders++;
-	}
 
 	/*
 	 * At this moment we are sure that there are no transactions with the
@@ -739,13 +725,7 @@ DefineIndex(IndexStmt *stmt,
 	 * We once again wait until no transaction can have the table open with
 	 * the index marked as read-only for updates.
 	 */
-	old_lockholders = GetLockConflicts(&heaplocktag, ShareLock);
+	WaitForLockers(heaplocktag, ShareLock);
-
-	while (VirtualTransactionIdIsValid(*old_lockholders))
-	{
-		VirtualXactLock(*old_lockholders, true);
-		old_lockholders++;
-	}
 
 	/*
 	 * Now take the "reference snapshot" that will be used by validate_index()

src/backend/storage/lmgr/lmgr.c

@@ -533,6 +533,73 @@ ConditionalXactLockTableWait(TransactionId xid)
 	return true;
 }
 
+/*
+ * WaitForLockersMultiple
+ *		Wait until no transaction holds locks that conflict with the given
+ *		locktags at the given lockmode.
+ *
+ * To do this, obtain the current list of lockers, and wait on their VXIDs
+ * until they are finished.
+ *
+ * Note we don't try to acquire the locks on the given locktags, only the VXIDs
+ * of its lock holders; if somebody grabs a conflicting lock on the objects
+ * after we obtained our initial list of lockers, we will not wait for them.
+ */
+void
+WaitForLockersMultiple(List *locktags, LOCKMODE lockmode)
+{
+	List	   *holders = NIL;
+	ListCell   *lc;
+
+	/* Done if no locks to wait for */
+	if (list_length(locktags) == 0)
+		return;
+
+	/* Collect the transactions we need to wait on */
+	foreach(lc, locktags)
+	{
+		LOCKTAG    *locktag = lfirst(lc);
+
+		holders = lappend(holders, GetLockConflicts(locktag, lockmode));
+	}
+
+	/*
+	 * Note: GetLockConflicts() never reports our own xid, hence we need not
+	 * check for that.	Also, prepared xacts are not reported, which is fine
+	 * since they certainly aren't going to do anything anymore.
+	 */
+
+	/* Finally wait for each such transaction to complete */
+	foreach(lc, holders)
+	{
+		VirtualTransactionId *lockholders = lfirst(lc);
+
+		while (VirtualTransactionIdIsValid(*lockholders))
+		{
+			VirtualXactLock(*lockholders, true);
+			lockholders++;
+		}
+	}
+
+	list_free_deep(holders);
+}
+
+/*
+ * WaitForLockers
+ *
+ * Same as WaitForLockersMultiple, for a single lock tag.
+ */
+void
+WaitForLockers(LOCKTAG heaplocktag, LOCKMODE lockmode)
+{
+	List   *l;
+
+	l = list_make1(&heaplocktag);
+	WaitForLockersMultiple(l, lockmode);
+	list_free(l);
+}
+
+
 /*
  *		LockDatabaseObject
  *

src/include/storage/lmgr.h

@@ -57,6 +57,10 @@ extern void XactLockTableDelete(TransactionId xid);
 extern void XactLockTableWait(TransactionId xid);
 extern bool ConditionalXactLockTableWait(TransactionId xid);
 
+/* Lock VXIDs, specified by conflicting locktags */
+extern void WaitForLockers(LOCKTAG heaplocktag, LOCKMODE lockmode);
+extern void WaitForLockersMultiple(List *locktags, LOCKMODE lockmode);
+
 /* Lock a general object (other than a relation) of the current database */
 extern void LockDatabaseObject(Oid classid, Oid objid, uint16 objsubid,
 				   LOCKMODE lockmode);