Acquire read lock on a buffer while writing it out, to prevent

concurrent modifications to the page by other backends.

diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 77b82b82c307fcb39be0987acbec1efe02ff182a..6711500e15b672963d0713765ae6c3480fe7662f 100644 (file)
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *       $Header: /cvsroot/pgsql/src/backend/storage/buffer/bufmgr.c,v 1.82 2000/06/08 22:37:20 momjian Exp $
+ *       $Header: /cvsroot/pgsql/src/backend/storage/buffer/bufmgr.c,v 1.83 2000/09/25 04:11:09 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -429,17 +429,10 @@ BufferAlloc(Relation reln,
        inProgress = FALSE;
        for (buf = (BufferDesc *) NULL; buf == (BufferDesc *) NULL;)
        {
-
-               /* GetFreeBuffer will abort if it can't find a free buffer */
                buf = GetFreeBuffer();
 
-               /*
-                * But it can return buf == NULL if we are in aborting transaction
-                * now and so elog(ERROR,...) in GetFreeBuffer will not abort
-                * again.
-                */
-               if (buf == NULL)
-                       return NULL;
+               /* GetFreeBuffer will abort if it can't find a free buffer */
+               Assert(buf);
 
                /*
                 * There should be exactly one pin on the buffer after it is
@@ -789,11 +782,21 @@ FlushBuffer(Buffer buffer, bool release)
        WaitIO(bufHdr, BufMgrLock); /* confirm end of IO */
        bufHdr->flags &= ~BM_JUST_DIRTIED;
        StartBufferIO(bufHdr, false);           /* output IO start */
+
        SpinRelease(BufMgrLock);
 
+       /*
+        * Grab a read lock on the buffer to ensure that no
+        * other backend changes its contents while we write it;
+        * see comments in BufferSync().
+        */
+       LockBuffer(BufferDescriptorGetBuffer(bufHdr), BUFFER_LOCK_SHARE);
+
        status = smgrflush(DEFAULT_SMGR, bufrel, bufHdr->tag.blockNum,
                                           (char *) MAKE_PTR(bufHdr->data));
 
+       LockBuffer(BufferDescriptorGetBuffer(bufHdr), BUFFER_LOCK_UNLOCK);
+
        /* drop relcache refcnt incremented by RelationIdCacheGetRelation */
        RelationDecrementReferenceCount(bufrel);
 
@@ -1017,19 +1020,6 @@ ClearBufferDirtiedByMe(Buffer buffer, BufferDesc *bufHdr)
  *             that have been dirtied by the current xact and flush them to disk.
  *             We do *not* flush dirty buffers that have been dirtied by other xacts.
  *             (This is a substantial change from pre-7.0 behavior.)
- *
- *     OLD COMMENTS (do these still apply?)
- *
- *             Also, we need to be sure that no other transaction is
- *             modifying the page as we flush it.      This is only a problem for objects
- *             that use a non-two-phase locking protocol, like btree indices.  For
- *             those objects, we would like to set a write lock for the duration of
- *             our IO.  Another possibility is to code updates to btree pages
- *             carefully, so that writing them out out of order cannot cause
- *             any unrecoverable errors.
- *
- *             I don't want to think hard about this right now, so I will try
- *             to come back to it later.
  */
 static void
 BufferSync()
@@ -1112,15 +1102,28 @@ BufferSync()
                                        bufHdr->flags &= ~BM_JUST_DIRTIED;
                                        StartBufferIO(bufHdr, false);           /* output IO start */
 
+                                       SpinRelease(BufMgrLock);
+
+                                       /*
+                                        * Grab a read lock on the buffer to ensure that no
+                                        * other backend changes its contents while we write it;
+                                        * otherwise we could write a non-self-consistent page
+                                        * image to disk, which'd be bad news if the other
+                                        * transaction aborts before writing its changes.
+                                        *
+                                        * Note that we still need the BM_JUST_DIRTIED mechanism
+                                        * in case someone dirties the buffer just before we
+                                        * grab this lock or just after we release it.
+                                        */
+                                       LockBuffer(BufferDescriptorGetBuffer(bufHdr),
+                                                          BUFFER_LOCK_SHARE);
+
                                        /*
                                         * If we didn't have the reldesc in our local cache,
                                         * write this page out using the 'blind write' storage
                                         * manager routine.  If we did find it, use the
                                         * standard interface.
                                         */
-#ifndef OPTIMIZE_SINGLE
-                                       SpinRelease(BufMgrLock);
-#endif  /* OPTIMIZE_SINGLE */
                                        if (reln == (Relation) NULL)
                                        {
                                                status = smgrblindwrt(DEFAULT_SMGR,
@@ -1137,9 +1140,14 @@ BufferSync()
                                                                                   bufHdr->tag.blockNum,
                                                                                (char *) MAKE_PTR(bufHdr->data));
                                        }
-#ifndef OPTIMIZE_SINGLE
+
+                                       /*
+                                        * Release the per-buffer readlock, reacquire BufMgrLock.
+                                        */
+                                       LockBuffer(BufferDescriptorGetBuffer(bufHdr),
+                                                          BUFFER_LOCK_UNLOCK);
+
                                        SpinAcquire(BufMgrLock);
-#endif  /* OPTIMIZE_SINGLE */
 
                                        UnpinBuffer(bufHdr);
                                        if (status == SM_FAIL)
@@ -1522,9 +1530,14 @@ BufferReplace(BufferDesc *bufHdr)
        /* To check if block content changed while flushing. - vadim 01/17/97 */
        bufHdr->flags &= ~BM_JUST_DIRTIED;
 
-#ifndef OPTIMIZE_SINGLE
        SpinRelease(BufMgrLock);
-#endif  /* OPTIMIZE_SINGLE */
+
+       /*
+        * Grab a read lock on the buffer to ensure that no
+        * other backend changes its contents while we write it;
+        * see comments in BufferSync().
+        */
+       LockBuffer(BufferDescriptorGetBuffer(bufHdr), BUFFER_LOCK_SHARE);
 
        if (reln != (Relation) NULL)
        {
@@ -1540,9 +1553,9 @@ BufferReplace(BufferDesc *bufHdr)
                                                          false);       /* no fsync */
        }
 
-#ifndef OPTIMIZE_SINGLE
+       LockBuffer(BufferDescriptorGetBuffer(bufHdr), BUFFER_LOCK_UNLOCK);
+
        SpinAcquire(BufMgrLock);
-#endif  /* OPTIMIZE_SINGLE */
 
        /* drop relcache refcnt incremented by RelationIdCacheGetRelation */
        if (reln != (Relation) NULL)