Seems I was too optimistic in supposing that sinval's maxMsgNum could be

read and written without a lock.  The value itself is atomic, sure, but on
processors with weak memory ordering it's possible for a reader to see the
value change before it sees the associated message written into the buffer
array.  Fix by introducing a spinlock that's used just to read and write
maxMsgNum.  (We could do this with less overhead if we recognized a concept
of "memory access barrier"; is it worth introducing such a thing?  At the
moment probably not --- I can't measure any clear slowdown from adding the
spinlock, so this solution is probably fine.)  Per buildfarm results.

diff --git a/src/backend/storage/ipc/sinvaladt.c b/src/backend/storage/ipc/sinvaladt.c
index 0befc4a93419a526d857219c7a80dc60a38e3436..e414e4a507105aef95fdde4e830b915b60490aa9 100644 (file)
--- a/src/backend/storage/ipc/sinvaladt.c
+++ b/src/backend/storage/ipc/sinvaladt.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *   $PostgreSQL: pgsql/src/backend/storage/ipc/sinvaladt.c,v 1.71 2008/06/19 21:32:56 tgl Exp $
+ *   $PostgreSQL: pgsql/src/backend/storage/ipc/sinvaladt.c,v 1.72 2008/06/20 00:24:53 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -23,6 +23,7 @@
 #include "storage/proc.h"
 #include "storage/shmem.h"
 #include "storage/sinvaladt.h"
+#include "storage/spin.h"
 
 
 /*
@@ -84,10 +85,20 @@
  * can operate in parallel with one or more readers, because the writer
  * has no need to touch anyone's ProcState, except in the infrequent cases
  * when SICleanupQueue is needed.  The only point of overlap is that
- * the writer might change maxMsgNum while readers are looking at it.
- * This should be okay: we are assuming that fetching or storing an int
- * is atomic, an assumption also made elsewhere in Postgres.  However
- * readers mustn't assume that maxMsgNum isn't changing under them.
+ * the writer wants to change maxMsgNum while readers need to read it.
+ * We deal with that by having a spinlock that readers must take for just
+ * long enough to read maxMsgNum, while writers take it for just long enough
+ * to write maxMsgNum.  (The exact rule is that you need the spinlock to
+ * read maxMsgNum if you are not holding SInvalWriteLock, and you need the
+ * spinlock to write maxMsgNum unless you are holding both locks.)
+ *
+ * Note: since maxMsgNum is an int and hence presumably atomically readable/
+ * writable, the spinlock might seem unnecessary.  The reason it is needed
+ * is to provide a memory barrier: we need to be sure that messages written
+ * to the array are actually there before maxMsgNum is increased, and that
+ * readers will see that data after fetching maxMsgNum.  Multiprocessors
+ * that have weak memory-ordering guarantees can fail without the memory
+ * barrier instructions that are included in the spinlock sequences.
  */
 
 
@@ -153,6 +164,8 @@ typedef struct SISeg
    int         lastBackend;    /* index of last active procState entry, +1 */
    int         maxBackends;    /* size of procState array */
 
+   slock_t     msgnumLock;     /* spinlock protecting maxMsgNum */
+
    /*
     * Circular buffer holding shared-inval messages
     */
@@ -209,12 +222,13 @@ CreateSharedInvalidationState(void)
    if (found)
        return;
 
-   /* Clear message counters, save size of procState array */
+   /* Clear message counters, save size of procState array, init spinlock */
    shmInvalBuffer->minMsgNum = 0;
    shmInvalBuffer->maxMsgNum = 0;
    shmInvalBuffer->nextThreshold = CLEANUP_MIN;
    shmInvalBuffer->lastBackend = 0;
    shmInvalBuffer->maxBackends = MaxBackends;
+   SpinLockInit(&shmInvalBuffer->msgnumLock);
 
    /* The buffer[] array is initially all unused, so we need not fill it */
 
@@ -365,6 +379,7 @@ SIInsertDataEntries(const SharedInvalidationMessage *data, int n)
    {
        int     nthistime = Min(n, WRITE_QUANTUM);
        int     numMsgs;
+       int     max;
 
        n -= nthistime;
 
@@ -388,10 +403,21 @@ SIInsertDataEntries(const SharedInvalidationMessage *data, int n)
        /*
         * Insert new message(s) into proper slot of circular buffer
         */
+       max = segP->maxMsgNum;
        while (nthistime-- > 0)
        {
-           segP->buffer[segP->maxMsgNum % MAXNUMMESSAGES] = *data++;
-           segP->maxMsgNum++;
+           segP->buffer[max % MAXNUMMESSAGES] = *data++;
+           max++;
+       }
+
+       /* Update current value of maxMsgNum using spinlock */
+       {
+           /* use volatile pointer to prevent code rearrangement */
+           volatile SISeg *vsegP = segP;
+
+           SpinLockAcquire(&vsegP->msgnumLock);
+           vsegP->maxMsgNum = max;
+           SpinLockRelease(&vsegP->msgnumLock);
        }
 
        LWLockRelease(SInvalWriteLock);
@@ -431,6 +457,7 @@ SIGetDataEntries(SharedInvalidationMessage *data, int datasize)
 {
    SISeg      *segP;
    ProcState  *stateP;
+   int         max;
    int         n;
    
    LWLockAcquire(SInvalReadLock, LW_SHARED);
@@ -438,6 +465,16 @@ SIGetDataEntries(SharedInvalidationMessage *data, int datasize)
    segP = shmInvalBuffer;
    stateP = &segP->procState[MyBackendId - 1];
 
+   /* Fetch current value of maxMsgNum using spinlock */
+   {
+       /* use volatile pointer to prevent code rearrangement */
+       volatile SISeg *vsegP = segP;
+
+       SpinLockAcquire(&vsegP->msgnumLock);
+       max = vsegP->maxMsgNum;
+       SpinLockRelease(&vsegP->msgnumLock);
+   }
+
    if (stateP->resetState)
    {
        /*
@@ -445,7 +482,7 @@ SIGetDataEntries(SharedInvalidationMessage *data, int datasize)
         * since the reset, as well; and that means we should clear the
         * signaled flag, too.
         */
-       stateP->nextMsgNum = segP->maxMsgNum;
+       stateP->nextMsgNum = max;
        stateP->resetState = false;
        stateP->signaled = false;
        LWLockRelease(SInvalReadLock);
@@ -459,13 +496,9 @@ SIGetDataEntries(SharedInvalidationMessage *data, int datasize)
     * There may be other backends that haven't read the message(s), so we
     * cannot delete them here.  SICleanupQueue() will eventually remove them
     * from the queue.
-    *
-    * Note: depending on the compiler, we might read maxMsgNum only once
-    * here, or each time through the loop.  It doesn't matter (as long as
-    * each fetch is atomic).
     */
    n = 0;
-   while (n < datasize && stateP->nextMsgNum < segP->maxMsgNum)
+   while (n < datasize && stateP->nextMsgNum < max)
    {
        data[n++] = segP->buffer[stateP->nextMsgNum % MAXNUMMESSAGES];
        stateP->nextMsgNum++;
@@ -474,7 +507,7 @@ SIGetDataEntries(SharedInvalidationMessage *data, int datasize)
    /*
     * Reset our "signaled" flag whenever we have caught up completely.
     */
-   if (stateP->nextMsgNum >= segP->maxMsgNum)
+   if (stateP->nextMsgNum >= max)
        stateP->signaled = false;
 
    LWLockRelease(SInvalReadLock);