Improve error detection capability in proclists.

Previously, although the initial state of a proclist_node is expected
to be next == prev == 0, proclist_delete_offset would reset nodes to
next == prev == INVALID_PGPROCNO when removing them from a list.
This is the same state that a node in a singleton list has, so that
it's impossible to distinguish not-in-a-list from in-a-list.  Change
proclist_delete_offset to reset removed nodes to next == prev == 0,
making it possible to distinguish those cases, and then add Asserts
to the list add and delete functions that the supplied node isn't
or is in a list at entry.  Also tighten assertions about the node
being in the particular list (not some other one) where it is possible
to check that in O(1) time.

In ConditionVariablePrepareToSleep, since we don't expect the process's
cvWaitLink to already be in a list, remove the more-or-less-useless
proclist_contains check; we'd rather have proclist_push_tail's new
assertion fire if that happens.

Improve various comments related to proclists, too.

Patch by me, reviewed by Thomas Munro.  This isn't back-patchable, since
there could theoretically be inlined copies of proclist_delete_offset in
third-party modules.  But it's only improving debuggability anyway.

Discussion: https://postgr.es/m/CAEepm=0NWKehYw7NDoUSf8juuKOPRnCyY3vuaSvhrEWsOTAa3w@mail.gmail.com

diff --git a/src/backend/storage/lmgr/condition_variable.c b/src/backend/storage/lmgr/condition_variable.c
index 60234db4cd0884eb1ac69e44bb79f74e658b4fda..e3bc034de45c764c6c104f3ab7764a3b64c69647 100644 (file)
--- a/src/backend/storage/lmgr/condition_variable.c
+++ b/src/backend/storage/lmgr/condition_variable.c
@@ -86,8 +86,7 @@ ConditionVariablePrepareToSleep(ConditionVariable *cv)
 
    /* Add myself to the wait queue. */
    SpinLockAcquire(&cv->mutex);
-   if (!proclist_contains(&cv->wakeup, pgprocno, cvWaitLink))
-       proclist_push_tail(&cv->wakeup, pgprocno, cvWaitLink);
+   proclist_push_tail(&cv->wakeup, pgprocno, cvWaitLink);
    SpinLockRelease(&cv->mutex);
 }
 

diff --git a/src/include/storage/proclist.h b/src/include/storage/proclist.h
index 4e25b47593fe5c1b95df3aad5134283520c04bee..59a478e1f632b02c6d895b8c10de781110ba3021 100644 (file)
--- a/src/include/storage/proclist.h
+++ b/src/include/storage/proclist.h
@@ -42,7 +42,7 @@ proclist_is_empty(proclist_head *list)
 
 /*
  * Get a pointer to a proclist_node inside a given PGPROC, given a procno and
- * an offset.
+ * the proclist_node field's offset within struct PGPROC.
  */
 static inline proclist_node *
 proclist_node_get(int procno, size_t node_offset)
@@ -53,13 +53,15 @@ proclist_node_get(int procno, size_t node_offset)
 }
 
 /*
- * Insert a node at the beginning of a list.
+ * Insert a process at the beginning of a list.
  */
 static inline void
 proclist_push_head_offset(proclist_head *list, int procno, size_t node_offset)
 {
    proclist_node *node = proclist_node_get(procno, node_offset);
 
+   Assert(node->next == 0 && node->prev == 0);
+
    if (list->head == INVALID_PGPROCNO)
    {
        Assert(list->tail == INVALID_PGPROCNO);
@@ -79,13 +81,15 @@ proclist_push_head_offset(proclist_head *list, int procno, size_t node_offset)
 }
 
 /*
- * Insert a node at the end of a list.
+ * Insert a process at the end of a list.
  */
 static inline void
 proclist_push_tail_offset(proclist_head *list, int procno, size_t node_offset)
 {
    proclist_node *node = proclist_node_get(procno, node_offset);
 
+   Assert(node->next == 0 && node->prev == 0);
+
    if (list->tail == INVALID_PGPROCNO)
    {
        Assert(list->head == INVALID_PGPROCNO);
@@ -105,30 +109,38 @@ proclist_push_tail_offset(proclist_head *list, int procno, size_t node_offset)
 }
 
 /*
- * Delete a node.  The node must be in the list.
+ * Delete a process from a list --- it must be in the list!
  */
 static inline void
 proclist_delete_offset(proclist_head *list, int procno, size_t node_offset)
 {
    proclist_node *node = proclist_node_get(procno, node_offset);
 
+   Assert(node->next != 0 || node->prev != 0);
+
    if (node->prev == INVALID_PGPROCNO)
+   {
+       Assert(list->head == procno);
        list->head = node->next;
+   }
    else
        proclist_node_get(node->prev, node_offset)->next = node->next;
 
    if (node->next == INVALID_PGPROCNO)
+   {
+       Assert(list->tail == procno);
        list->tail = node->prev;
+   }
    else
        proclist_node_get(node->next, node_offset)->prev = node->prev;
 
-   node->next = node->prev = INVALID_PGPROCNO;
+   node->next = node->prev = 0;
 }
 
 /*
- * Check if a node is currently in a list.  It must be known that the node is
- * not in any _other_ proclist that uses the same proclist_node, so that the
- * only possibilities are that it is in this list or none.
+ * Check if a process is currently in a list.  It must be known that the
+ * process is not in any _other_ proclist that uses the same proclist_node,
+ * so that the only possibilities are that it is in this list or none.
  */
 static inline bool
 proclist_contains_offset(proclist_head *list, int procno,
@@ -136,27 +148,26 @@ proclist_contains_offset(proclist_head *list, int procno,
 {
    proclist_node *node = proclist_node_get(procno, node_offset);
 
-   /*
-    * If this is not a member of a proclist, then the next and prev pointers
-    * should be 0. Circular lists are not allowed so this condition is not
-    * confusable with a real pgprocno 0.
-    */
+   /* If it's not in any list, it's definitely not in this one. */
    if (node->prev == 0 && node->next == 0)
        return false;
 
-   /* If there is a previous node, then this node must be in the list. */
-   if (node->prev != INVALID_PGPROCNO)
-       return true;
-
    /*
-    * There is no previous node, so the only way this node can be in the list
-    * is if it's the head node.
+    * It must, in fact, be in this list.  Ideally, in assert-enabled builds,
+    * we'd verify that.  But since this function is typically used while
+    * holding a spinlock, crawling the whole list is unacceptable.  However,
+    * we can verify matters in O(1) time when the node is a list head or
+    * tail, and that seems worth doing, since in practice that should often
+    * be enough to catch mistakes.
     */
-   return list->head == procno;
+   Assert(node->prev != INVALID_PGPROCNO || list->head == procno);
+   Assert(node->next != INVALID_PGPROCNO || list->tail == procno);
+
+   return true;
 }
 
 /*
- * Remove and return the first node from a list (there must be one).
+ * Remove and return the first process from a list (there must be one).
  */
 static inline PGPROC *
 proclist_pop_head_node_offset(proclist_head *list, size_t node_offset)
@@ -205,4 +216,4 @@ proclist_pop_head_node_offset(proclist_head *list, size_t node_offset)
             proclist_node_get((iter).cur,                                  \
                               offsetof(PGPROC, link_member))->next)
 
-#endif
+#endif                         /* PROCLIST_H */

diff --git a/src/include/storage/proclist_types.h b/src/include/storage/proclist_types.h
index 237fb7613fc722652e1557956c771f27a28a0ba8..f4dac10fb6403dcc95db7ee140709a61968518a9 100644 (file)
--- a/src/include/storage/proclist_types.h
+++ b/src/include/storage/proclist_types.h
@@ -16,7 +16,12 @@
 #define PROCLIST_TYPES_H
 
 /*
- * A node in a list of processes.
+ * A node in a doubly-linked list of processes.  The link fields contain
+ * the 0-based PGPROC indexes of the next and previous process, or
+ * INVALID_PGPROCNO in the next-link of the last node and the prev-link
+ * of the first node.  A node that is currently not in any list
+ * should have next == prev == 0; this is not a possible state for a node
+ * that is in a list, because we disallow circularity.
  */
 typedef struct proclist_node
 {
@@ -25,7 +30,8 @@ typedef struct proclist_node
 } proclist_node;
 
 /*
- * Head of a doubly-linked list of PGPROCs, identified by pgprocno.
+ * Header of a doubly-linked list of PGPROCs, identified by pgprocno.
+ * An empty list is represented by head == tail == INVALID_PGPROCNO.
  */
 typedef struct proclist_head
 {
@@ -42,4 +48,4 @@ typedef struct proclist_mutable_iter
    int         next;           /* pgprocno of the next PGPROC */
 } proclist_mutable_iter;
 
-#endif
+#endif                         /* PROCLIST_TYPES_H */