Add WaitForLockers in lmgr, refactoring index.c code

This is in support of a future REINDEX CONCURRENTLY feature.

Michael Paquier

diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index b73ee4f2d19f877b30806ce5a092caa4b8b4ed88..826e504bb4796abe656d16ed4d354356675297e5 100644 (file)
--- a/src/backend/catalog/index.c
+++ b/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
-        * conflict with AccessExclusiveLock on the table -- ie, which ones
-        * 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 for a query. 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);
-
-       while (VirtualTransactionIdIsValid(*old_lockholders))
-       {
-           VirtualXactLock(*old_lockholders, true);
-           old_lockholders++;
-       }
+       WaitForLockers(heaplocktag, AccessExclusiveLock);
 
        /*
         * 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);
-
-       while (VirtualTransactionIdIsValid(*old_lockholders))
-       {
-           VirtualXactLock(*old_lockholders, true);
-           old_lockholders++;
-       }
+       WaitForLockers(heaplocktag, AccessExclusiveLock);
 
        /*
         * Re-open relations to allow us to complete our actions.

diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index 902daa079461def237c7d16a78aee853da9a92d2..2155252e4ad27563b56ccee5b56704ee5b2f47d6 100644 (file)
--- a/src/backend/commands/indexcmds.c
+++ b/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
-    * currently would conflict with ShareLock on the table -- ie, which ones
-    * have a lock that permits writing 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 writing after this point; they will see
-    * the new index when they open it.
+    * with the old list of indexes. 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);
-
-   while (VirtualTransactionIdIsValid(*old_lockholders))
-   {
-       VirtualXactLock(*old_lockholders, true);
-       old_lockholders++;
-   }
+   WaitForLockers(heaplocktag, ShareLock);
 
    /*
     * 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);
-
-   while (VirtualTransactionIdIsValid(*old_lockholders))
-   {
-       VirtualXactLock(*old_lockholders, true);
-       old_lockholders++;
-   }
+   WaitForLockers(heaplocktag, ShareLock);
 
    /*
     * Now take the "reference snapshot" that will be used by validate_index()

diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 2b7a1db3ebaae86957354109d635ff1f17b8059e..a978172796d6b5363e1c7b606c7b010481b26381 100644 (file)
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/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
  *

diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index 9b1fb93462a68c90758f787cbd4c706e2e1b6035..1a8c018a1eed11ebb7b3653eaabb7168a331bd63 100644 (file)
--- a/src/include/storage/lmgr.h
+++ b/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);