1. Checkpoint.undo may be after checkpoint itself:

   - no more elog(STOP) in StartupXLOG();
   - both checkpoint' undo & redo are used to define
     oldest on-line log file.
2. Ability to pre-allocate a few log files at checkpoint time
   (wal_files option). Off by default.

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 136ce1d..c0b6104 100644 (file)
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -6,7 +6,7 @@
  * Portions Copyright (c) 1996-2000, PostgreSQL, Inc
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $Header: /cvsroot/pgsql/src/backend/access/transam/xlog.c,v 1.47 2000/12/30 06:52:34 vadim Exp $
+ * $Header: /cvsroot/pgsql/src/backend/access/transam/xlog.c,v 1.48 2001/01/09 06:24:32 vadim Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -40,6 +40,7 @@
 #include "miscadmin.h"
 
 int                    XLOGbuffers = 8;
+int                    XLOGfiles = 0;  /* how many files to pre-allocate */
 XLogRecPtr     MyLastRecPtr = {0, 0};
 uint32         CritSectionCount = 0;
 bool           InRecovery = false;
@@ -182,6 +183,18 @@ typedef struct BkpBlock
 #define XLogLastSeg            (0xffffffff / XLogSegSize)
 #define XLogFileSize   (XLogLastSeg * XLogSegSize)
 
+#define NextLogSeg(_logId, _logSeg)            \
+{\
+       if (_logSeg >= XLogLastSeg)\
+       {\
+               _logId++;\
+               _logSeg = 0;\
+       }\
+       else\
+               _logSeg++;\
+}
+
+
 #define XLogFileName(path, log, seg)   \
                        snprintf(path, MAXPGPATH, "%s%c%08X%08X",       \
                                         XLogDir, SEP_CHAR, log, seg)
@@ -856,8 +869,8 @@ XLogWrite(char *buffer)
                        UpdateControlFile();
                        SpinRelease(ControlFileLockId);
                        if (!usexistent)        /* there was no file */
-                               elog(LOG, "XLogWrite: had to create new log file - "
-                                       "you probably should do checkpoints more often");
+                               elog(LOG, "XLogWrite: new log file created - "
+                                       "try to increase WAL_FILES");
                }
 
                if (logFile < 0)
@@ -1186,7 +1199,7 @@ ReadRecord(XLogRecPtr *RecPtr, char *buffer)
                tmpRecPtr.xrecoff += SizeOfXLogPHD;
        }
        else if (!XRecOffIsValid(RecPtr->xrecoff))
-               elog(STOP, "ReadRecord: invalid record offset in (%u, %u)",
+               elog(STOP, "ReadRecord: invalid record offset at (%u, %u)",
                         RecPtr->xlogid, RecPtr->xrecoff);
 
        if (readFile >= 0 && (RecPtr->xlogid != readId ||
@@ -1232,15 +1245,21 @@ ReadRecord(XLogRecPtr *RecPtr, char *buffer)
        record = (XLogRecord *) ((char *) readBuf + RecPtr->xrecoff % BLCKSZ);
 
 got_record:;
+       if (record->xl_len == 0)
+       {
+               elog(emode, "ReadRecord: record with zero len at (%u, %u)",
+                       RecPtr->xlogid, RecPtr->xrecoff);
+               goto next_record_is_invalid;
+       }
        if (record->xl_len > _INTL_MAXLOGRECSZ)
        {
-               elog(emode, "ReadRecord: too long record len %u in (%u, %u)",
+               elog(emode, "ReadRecord: too long record len %u at (%u, %u)",
                        record->xl_len, RecPtr->xlogid, RecPtr->xrecoff);
                goto next_record_is_invalid;
        }
        if (record->xl_rmid > RM_MAX_ID)
        {
-               elog(emode, "ReadRecord: invalid resource managed id %u in (%u, %u)",
+               elog(emode, "ReadRecord: invalid resource managed id %u at (%u, %u)",
                         record->xl_rmid, RecPtr->xlogid, RecPtr->xrecoff);
                goto next_record_is_invalid;
        }
@@ -1841,8 +1860,6 @@ StartupXLOG()
                elog(STOP, "Invalid redo in checkPoint record");
        if (checkPoint.undo.xrecoff == 0)
                checkPoint.undo = RecPtr;
-       if (XLByteLT(RecPtr, checkPoint.undo))
-               elog(STOP, "Invalid undo in checkPoint record");
 
        if (XLByteLT(checkPoint.undo, RecPtr) || 
                XLByteLT(checkPoint.redo, RecPtr))
@@ -1969,6 +1986,23 @@ StartupXLOG()
                CreateCheckPoint(true);
                XLogCloseRelationCache();
        }
+
+       if (XLOGfiles > 0)              /* pre-allocate log files */
+       {
+               uint32  _logId = logId,
+                               _logSeg = logSeg;
+               int             lf, i;
+               bool    usexistent;
+
+               for (i = 1; i <= XLOGfiles; i++)
+               {
+                       NextLogSeg(_logId, _logSeg);
+                       usexistent = false;
+                       lf = XLogFileInit(_logId, _logSeg, &usexistent);
+                       close(lf);
+               }
+       }
+
        InRecovery = false;
 
        ControlFile->state = DB_IN_PRODUCTION;
@@ -2117,48 +2151,70 @@ CreateCheckPoint(bool shutdown)
 
        SpinAcquire(ControlFileLockId);
        if (shutdown)
+       {
+               /* probably should delete extra log files */
                ControlFile->state = DB_SHUTDOWNED;
-       else    /* create new log file */
+       }
+       else    /* create new log file(s) */
        {
-               if (recptr.xrecoff % XLogSegSize >= 
-                       (uint32) (0.75 * XLogSegSize))
+               int             lf;
+               bool    usexistent = true;
+
+               _logId = recptr.xlogid;
+               _logSeg = (recptr.xrecoff - 1) / XLogSegSize;
+               if (XLOGfiles > 0)
                {
-                       int             lf;
-                       bool    usexistent = true;
+                       struct timeval  delay;
+                       int                             i;
 
-                       _logId = recptr.xlogid;
-                       _logSeg = recptr.xrecoff / XLogSegSize;
-                       if (_logSeg >= XLogLastSeg)
+                       for (i = 1; i <= XLOGfiles; i++)
                        {
-                               _logId++;
-                               _logSeg = 0;
+                               usexistent = true;
+                               NextLogSeg(_logId, _logSeg);
+                               lf = XLogFileInit(_logId, _logSeg, &usexistent);
+                               close(lf);
+                               /*
+                                * Give up ControlFileLockId for 1/50 sec to let other
+                                * backends switch to new log file in XLogWrite()
+                                */
+                               SpinRelease(ControlFileLockId);
+                               delay.tv_sec = 0;
+                               delay.tv_usec = 20000;
+                               (void) select(0, NULL, NULL, NULL, &delay);
+                               SpinAcquire(ControlFileLockId);
                        }
-                       else
-                               _logSeg++;
+               }
+               else if ((recptr.xrecoff - 1) % XLogSegSize >= 
+                       (uint32) (0.75 * XLogSegSize))
+               {
+                       NextLogSeg(_logId, _logSeg);
                        lf = XLogFileInit(_logId, _logSeg, &usexistent);
                        close(lf);
                }
        }
 
        ControlFile->checkPoint = MyLastRecPtr;
-
-       _logId = ControlFile->logId;
-       _logSeg = ControlFile->logSeg - 1;
        strcpy(archdir, ControlFile->archdir);
-
        ControlFile->time = time(NULL);
        UpdateControlFile();
        SpinRelease(ControlFileLockId);
 
        /*
         * Delete offline log files. Get oldest online
-        * log file from undo rec if it's valid.
+        * log file from redo or undo record, whatever
+        * is older.
         */
-       if (checkPoint.undo.xrecoff != 0)
+       if (checkPoint.undo.xrecoff != 0 && 
+               XLByteLT(checkPoint.undo, checkPoint.redo))
        {
                _logId = checkPoint.undo.xlogid;
                _logSeg = checkPoint.undo.xrecoff / XLogSegSize;
        }
+       else
+       {
+               _logId = checkPoint.redo.xlogid;
+               _logSeg = checkPoint.redo.xrecoff / XLogSegSize;
+       }
        if (_logId || _logSeg)
        {
                if (_logSeg)

diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 6b9b504..9c23f73 100644 (file)
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -4,7 +4,7 @@
  * Support for grand unified configuration scheme, including SET
  * command, configuration file, and command line options.
  *
- * $Header: /cvsroot/pgsql/src/backend/utils/misc/guc.c,v 1.26 2000/12/03 14:36:46 petere Exp $
+ * $Header: /cvsroot/pgsql/src/backend/utils/misc/guc.c,v 1.27 2001/01/09 06:24:33 vadim Exp $
  *
  * Copyright 2000 by PostgreSQL Global Development Group
  * Written by Peter Eisentraut <peter_e@gmx.net>.
@@ -38,6 +38,7 @@ extern bool Log_connections;
 
 extern int CheckPointTimeout;
 extern int XLOGbuffers;
+extern int XLOGfiles;
 extern int XLOG_DEBUG;
 extern int CommitDelay;
 
@@ -274,6 +275,9 @@ ConfigureNamesInt[] =
        {"wal_buffers",                 PGC_POSTMASTER,                 &XLOGbuffers,
         8, 4, INT_MAX},
 
+       {"wal_files",                   PGC_POSTMASTER,                 &XLOGfiles,
+        0, 0, 64},
+
        {"wal_debug",                   PGC_SUSET,                              &XLOG_DEBUG,
         0, 0, 16},