database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-06-16 06:01:02 +03:00
Code Activity

This patch extracts page buffer pooling and the simple

Browse Source 
least-recently-used strategy from clog.c into slru.c.  It doesn't
change any visible behaviour and passes all regression tests plus a
TruncateCLOG test done manually.

Apart from refactoring I made a little change to SlruRecentlyUsed,
formerly ClogRecentlyUsed:  It now skips incrementing lru_counts, if
slotno is already the LRU slot, thus saving a few CPU cycles.  To make
this work, lru_counts are initialised to 1 in SimpleLruInit.

SimpleLru will be used by pg_subtrans (part of the nested transactions
project), so the main purpose of this patch is to avoid future code
duplication.

Manfred Koizar
This commit is contained in:
Bruce Momjian
2003-06-11 22:37:46 +00:00
parent 240dc5cddc
commit 0abe7431c6
6 changed files with 990 additions and 799 deletions
Download Patch File Download Diff File Expand all files Collapse all files

828
src/backend/access/transam/clog.c
View File

@ -13,7 +13,7 @@
* Portions Copyright (c) 1996-2002, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
* $Header: /cvsroot/pgsql/src/backend/access/transam/clog.c,v 1.15 2003/05/03 03:52:07 momjian Exp $
* $Header: /cvsroot/pgsql/src/backend/access/transam/clog.c,v 1.16 2003/06/11 22:37:45 momjian Exp $
*
*-------------------------------------------------------------------------
*/
@ -26,15 +26,14 @@
#include <unistd.h>
#include "access/clog.h"
#include "access/slru.h"
#include "storage/lwlock.h"
#include "miscadmin.h"
/*
* Defines for CLOG page and segment sizes. A page is the same BLCKSZ
* as is used everywhere else in Postgres. The CLOG segment size can be
* chosen somewhat arbitrarily; we make it 1 million transactions by default,
* or 256Kb.
* as is used everywhere else in Postgres.
*
* Note: because TransactionIds are 32 bits and wrap around at 0xFFFFFFFF,
* CLOG page numbering also wraps around at 0xFFFFFFFF/CLOG_XACTS_PER_PAGE,
@ -43,17 +42,12 @@
* segment and page numbers in TruncateCLOG (see CLOGPagePrecedes).
*/
#define CLOG_BLCKSZ BLCKSZ
/* We need two bits per xact, so four xacts fit in a byte */
#define CLOG_BITS_PER_XACT 2
#define CLOG_XACTS_PER_BYTE 4
#define CLOG_XACTS_PER_PAGE (CLOG_BLCKSZ * CLOG_XACTS_PER_BYTE)
#define CLOG_XACTS_PER_PAGE (BLCKSZ * CLOG_XACTS_PER_BYTE)
#define CLOG_XACT_BITMASK ((1 << CLOG_BITS_PER_XACT) - 1)
#define CLOG_XACTS_PER_SEGMENT 0x100000
#define CLOG_PAGES_PER_SEGMENT (CLOG_XACTS_PER_SEGMENT / CLOG_XACTS_PER_PAGE)
#define TransactionIdToPage(xid) ((xid) / (TransactionId) CLOG_XACTS_PER_PAGE)
#define TransactionIdToPgIndex(xid) ((xid) % (TransactionId) CLOG_XACTS_PER_PAGE)
#define TransactionIdToByte(xid) (TransactionIdToPgIndex(xid) / CLOG_XACTS_PER_BYTE)
@ -63,47 +57,6 @@
/*----------
* Shared-memory data structures for CLOG control
*
* We use a simple least-recently-used scheme to manage a pool of page
* buffers for the CLOG. Under ordinary circumstances we expect that write
* traffic will occur mostly to the latest CLOG page (and to the just-prior
* page, soon after a page transition). Read traffic will probably touch
* a larger span of pages, but in any case a fairly small number of page
* buffers should be sufficient. So, we just search the buffers using plain
* linear search; there's no need for a hashtable or anything fancy.
* The management algorithm is straight LRU except that we will never swap
* out the latest page (since we know it's going to be hit again eventually).
*
* We use an overall LWLock to protect the shared data structures, plus
* per-buffer LWLocks that synchronize I/O for each buffer. A process
* that is reading in or writing out a page buffer does not hold the control
* lock, only the per-buffer lock for the buffer it is working on.
*
* To change the page number or state of a buffer, one must normally hold
* the control lock. (The sole exception to this rule is that a writer
* process changes the state from DIRTY to WRITE_IN_PROGRESS while holding
* only the per-buffer lock.) If the buffer's state is neither EMPTY nor
* CLEAN, then there may be processes doing (or waiting to do) I/O on the
* buffer, so the page number may not be changed, and the only allowed state
* transition is to change WRITE_IN_PROGRESS to DIRTY after dirtying the page.
* To do any other state transition involving a buffer with potential I/O
* processes, one must hold both the per-buffer lock and the control lock.
* (Note the control lock must be acquired second; do not wait on a buffer
* lock while holding the control lock.) A process wishing to read a page
* marks the buffer state as READ_IN_PROGRESS, then drops the control lock,
* acquires the per-buffer lock, and rechecks the state before proceeding.
* This recheck takes care of the possibility that someone else already did
* the read, while the early marking prevents someone else from trying to
* read the same page into a different buffer.
*
* Note we are assuming that read and write of the state value is atomic,
* since I/O processes may examine and change the state while not holding
* the control lock.
*
* As with the regular buffer manager, it is possible for another process
* to re-dirty a page that is currently being written out. This is handled
* by setting the page's state from WRITE_IN_PROGRESS to DIRTY. The writing
* process must notice this and not mark the page CLEAN when it's done.
*
* XLOG interactions: this module generates an XLOG record whenever a new
* CLOG page is initialized to zeroes. Other writes of CLOG come from
* recording of transaction commit or abort in xact.c, which generates its
@ -117,90 +70,12 @@
*----------
*/
typedef enum
{
CLOG_PAGE_EMPTY, /* CLOG buffer is not in use */
CLOG_PAGE_READ_IN_PROGRESS, /* CLOG page is being read in */
CLOG_PAGE_CLEAN, /* CLOG page is valid and not dirty */
CLOG_PAGE_DIRTY, /* CLOG page is valid but needs write */
CLOG_PAGE_WRITE_IN_PROGRESS /* CLOG page is being written out */
} ClogPageStatus;
/*
* Shared-memory state for CLOG.
*/
typedef struct ClogCtlData
{
/*
* Info for each buffer slot. Page number is undefined when status is
* EMPTY. lru_count is essentially the number of operations since
* last use of this page; the page with highest lru_count is the best
* candidate to replace.
*/
char *page_buffer[NUM_CLOG_BUFFERS];
ClogPageStatus page_status[NUM_CLOG_BUFFERS];
int page_number[NUM_CLOG_BUFFERS];
unsigned int page_lru_count[NUM_CLOG_BUFFERS];
/*
* latest_page_number is the page number of the current end of the
* CLOG; this is not critical data, since we use it only to avoid
* swapping out the latest page.
*/
int latest_page_number;
} ClogCtlData;
static ClogCtlData *ClogCtl = NULL;
/*
* ClogBufferLocks is set during CLOGShmemInit and does not change thereafter.
* The value is automatically inherited by backends via fork, and
* doesn't need to be in shared memory.
*/
static LWLockId *ClogBufferLocks; /* Per-buffer I/O locks */
/*
* ClogDir is set during CLOGShmemInit and does not change thereafter.
* The value is automatically inherited by backends via fork, and
* doesn't need to be in shared memory.
*/
static char ClogDir[MAXPGPATH];
#define ClogFileName(path, seg) \
snprintf(path, MAXPGPATH, "%s/%04X", ClogDir, seg)
/*
* Macro to mark a buffer slot "most recently used".
*/
#define ClogRecentlyUsed(slotno) \
do { \
int iilru; \
for (iilru = 0; iilru < NUM_CLOG_BUFFERS; iilru++) \
ClogCtl->page_lru_count[iilru]++; \
ClogCtl->page_lru_count[slotno] = 0; \
} while (0)
/* Saved info for CLOGReportIOError */
typedef enum
{
CLOG_OPEN_FAILED,
CLOG_CREATE_FAILED,
CLOG_SEEK_FAILED,
CLOG_READ_FAILED,
CLOG_WRITE_FAILED
} ClogErrorCause;
static ClogErrorCause clog_errcause;
static int clog_errno;
static SlruCtlData ClogCtlData;
static SlruCtl ClogCtl = &ClogCtlData;
static int ZeroCLOGPage(int pageno, bool writeXlog);
static int ReadCLOGPage(int pageno, TransactionId xid);
static void WriteCLOGPage(int slotno);
static bool CLOGPhysicalReadPage(int pageno, int slotno);
static bool CLOGPhysicalWritePage(int pageno, int slotno);
static void CLOGReportIOError(int pageno, TransactionId xid);
static int SelectLRUCLOGPage(int pageno);
static bool ScanCLOGDirectory(int cutoffPage, bool doDeletions);
static bool CLOGPagePrecedes(int page1, int page2);
static void WriteZeroPageXlogRec(int pageno);
@ -217,16 +92,15 @@ TransactionIdSetStatus(TransactionId xid, XidStatus status)
int pageno = TransactionIdToPage(xid);
int byteno = TransactionIdToByte(xid);
int bshift = TransactionIdToBIndex(xid) * CLOG_BITS_PER_XACT;
int slotno;
char *byteptr;
Assert(status == TRANSACTION_STATUS_COMMITTED ||
status == TRANSACTION_STATUS_ABORTED);
LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
LWLockAcquire(ClogCtl->locks->ControlLock, LW_EXCLUSIVE);
slotno = ReadCLOGPage(pageno, xid);
byteptr = ClogCtl->page_buffer[slotno] + byteno;
byteptr = SimpleLruReadPage(ClogCtl, pageno, xid, true);
byteptr += byteno;
/* Current state should be 0 or target state */
Assert(((*byteptr >> bshift) & CLOG_XACT_BITMASK) == 0 ||
@ -234,9 +108,9 @@ TransactionIdSetStatus(TransactionId xid, XidStatus status)
*byteptr |= (status << bshift);
ClogCtl->page_status[slotno] = CLOG_PAGE_DIRTY;
/* ...->page_status[slotno] = CLOG_PAGE_DIRTY; already done */
LWLockRelease(CLogControlLock);
LWLockRelease(ClogCtl->locks->ControlLock);
}
/*
@ -251,18 +125,17 @@ TransactionIdGetStatus(TransactionId xid)
int pageno = TransactionIdToPage(xid);
int byteno = TransactionIdToByte(xid);
int bshift = TransactionIdToBIndex(xid) * CLOG_BITS_PER_XACT;
int slotno;
char *byteptr;
XidStatus status;
LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
LWLockAcquire(ClogCtl->locks->ControlLock, LW_EXCLUSIVE);
slotno = ReadCLOGPage(pageno, xid);
byteptr = ClogCtl->page_buffer[slotno] + byteno;
byteptr = SimpleLruReadPage(ClogCtl, pageno, xid, false);
byteptr += byteno;
status = (*byteptr >> bshift) & CLOG_XACT_BITMASK;
LWLockRelease(CLogControlLock);
LWLockRelease(ClogCtl->locks->ControlLock);
return status;
}
@ -271,70 +144,18 @@ TransactionIdGetStatus(TransactionId xid)
/*
* Initialization of shared memory for CLOG
*/
int
CLOGShmemSize(void)
{
return MAXALIGN(sizeof(ClogCtlData) + CLOG_BLCKSZ * NUM_CLOG_BUFFERS)
#ifdef EXEC_BACKEND
+ MAXALIGN(NUM_CLOG_BUFFERS * sizeof(LWLockId))
#endif
;
return SimpleLruShmemSize();
}
void
CLOGShmemInit(void)
{
bool found;
int slotno;
/* Handle ClogCtl */
/* this must agree with space requested by CLOGShmemSize() */
ClogCtl = (ClogCtlData *) ShmemInitStruct("CLOG Ctl",
MAXALIGN(sizeof(ClogCtlData) +
CLOG_BLCKSZ * NUM_CLOG_BUFFERS), &found);
if (!IsUnderPostmaster)
/* Initialize ClogCtl shared memory area */
{
char *bufptr;
Assert(!found);
memset(ClogCtl, 0, sizeof(ClogCtlData));
bufptr = (char *)ClogCtl + sizeof(ClogCtlData);
for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
{
ClogCtl->page_buffer[slotno] = bufptr;
ClogCtl->page_status[slotno] = CLOG_PAGE_EMPTY;
bufptr += CLOG_BLCKSZ;
}
/* ClogCtl->latest_page_number will be set later */
}
else
Assert(found);
/* Handle ClogBufferLocks */
#ifdef EXEC_BACKEND
ClogBufferLocks = (LWLockId *) ShmemInitStruct("CLOG Buffer Locks",
NUM_CLOG_BUFFERS * sizeof(LWLockId), &found);
Assert((!found && !IsUnderPostmaster) || (found && IsUnderPostmaster));
#else
ClogBufferLocks = malloc(NUM_CLOG_BUFFERS * sizeof(LWLockId));
Assert(ClogBufferLocks);
#endif
if (!IsUnderPostmaster)
for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
ClogBufferLocks[slotno] = LWLockAssign();
/* Init CLOG directory path */
snprintf(ClogDir, MAXPGPATH, "%s/pg_clog", DataDir);
SimpleLruInit(ClogCtl, "CLOG Ctl", "pg_clog");
ClogCtl->PagePrecedes = CLOGPagePrecedes;
}
/*
@ -348,16 +169,16 @@ BootStrapCLOG(void)
{
int slotno;
LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
LWLockAcquire(ClogCtl->locks->ControlLock, LW_EXCLUSIVE);
/* Create and zero the first page of the commit log */
slotno = ZeroCLOGPage(0, false);
/* Make sure it's written out */
WriteCLOGPage(slotno);
Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
SimpleLruWritePage(ClogCtl, slotno);
/* Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN); */
LWLockRelease(CLogControlLock);
LWLockRelease(ClogCtl->locks->ControlLock);
}
/*
@ -372,24 +193,7 @@ BootStrapCLOG(void)
static int
ZeroCLOGPage(int pageno, bool writeXlog)
{
int slotno;
/* Find a suitable buffer slot for the page */
slotno = SelectLRUCLOGPage(pageno);
Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY ||
ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN ||
ClogCtl->page_number[slotno] == pageno);
/* Mark the slot as containing this page */
ClogCtl->page_number[slotno] = pageno;
ClogCtl->page_status[slotno] = CLOG_PAGE_DIRTY;
ClogRecentlyUsed(slotno);
/* Set the buffer to zeroes */
MemSet(ClogCtl->page_buffer[slotno], 0, CLOG_BLCKSZ);
/* Assume this page is now the latest active page */
ClogCtl->latest_page_number = pageno;
int slotno = SimpleLruZeroPage(ClogCtl, pageno);
if (writeXlog)
WriteZeroPageXlogRec(pageno);
@ -397,430 +201,6 @@ ZeroCLOGPage(int pageno, bool writeXlog)
return slotno;
}
/*
* Find a CLOG page in a shared buffer, reading it in if necessary.
* The page number must correspond to an already-initialized page.
*
* The passed-in xid is used only for error reporting, and may be
* InvalidTransactionId if no specific xid is associated with the action.
*
* Return value is the shared-buffer slot number now holding the page.
* The buffer's LRU access info is updated.
*
* Control lock must be held at entry, and will be held at exit.
*/
static int
ReadCLOGPage(int pageno, TransactionId xid)
{
/* Outer loop handles restart if we lose the buffer to someone else */
for (;;)
{
int slotno;
bool ok;
/* See if page already is in memory; if not, pick victim slot */
slotno = SelectLRUCLOGPage(pageno);
/* Did we find the page in memory? */
if (ClogCtl->page_number[slotno] == pageno &&
ClogCtl->page_status[slotno] != CLOG_PAGE_EMPTY)
{
/* If page is still being read in, we cannot use it yet */
if (ClogCtl->page_status[slotno] != CLOG_PAGE_READ_IN_PROGRESS)
{
/* otherwise, it's ready to use */
ClogRecentlyUsed(slotno);
return slotno;
}
}
else
{
/* We found no match; assert we selected a freeable slot */
Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY ||
ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
}
/* Mark the slot read-busy (no-op if it already was) */
ClogCtl->page_number[slotno] = pageno;
ClogCtl->page_status[slotno] = CLOG_PAGE_READ_IN_PROGRESS;
/*
* Temporarily mark page as recently-used to discourage
* SelectLRUCLOGPage from selecting it again for someone else.
*/
ClogCtl->page_lru_count[slotno] = 0;
/* Release shared lock, grab per-buffer lock instead */
LWLockRelease(CLogControlLock);
LWLockAcquire(ClogBufferLocks[slotno], LW_EXCLUSIVE);
/*
* Check to see if someone else already did the read, or took the
* buffer away from us. If so, restart from the top.
*/
if (ClogCtl->page_number[slotno] != pageno ||
ClogCtl->page_status[slotno] != CLOG_PAGE_READ_IN_PROGRESS)
{
LWLockRelease(ClogBufferLocks[slotno]);
LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
continue;
}
/* Okay, do the read */
ok = CLOGPhysicalReadPage(pageno, slotno);
/* Re-acquire shared control lock and update page state */
LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
Assert(ClogCtl->page_number[slotno] == pageno &&
ClogCtl->page_status[slotno] == CLOG_PAGE_READ_IN_PROGRESS);
ClogCtl->page_status[slotno] = ok ? CLOG_PAGE_CLEAN : CLOG_PAGE_EMPTY;
LWLockRelease(ClogBufferLocks[slotno]);
/* Now it's okay to elog if we failed */
if (!ok)
CLOGReportIOError(pageno, xid);
ClogRecentlyUsed(slotno);
return slotno;
}
}
/*
* Write a CLOG page from a shared buffer, if necessary.
* Does nothing if the specified slot is not dirty.
*
* NOTE: only one write attempt is made here. Hence, it is possible that
* the page is still dirty at exit (if someone else re-dirtied it during
* the write). However, we *do* attempt a fresh write even if the page
* is already being written; this is for checkpoints.
*
* Control lock must be held at entry, and will be held at exit.
*/
static void
WriteCLOGPage(int slotno)
{
int pageno;
bool ok;
/* Do nothing if page does not need writing */
if (ClogCtl->page_status[slotno] != CLOG_PAGE_DIRTY &&
ClogCtl->page_status[slotno] != CLOG_PAGE_WRITE_IN_PROGRESS)
return;
pageno = ClogCtl->page_number[slotno];
/* Release shared lock, grab per-buffer lock instead */
LWLockRelease(CLogControlLock);
LWLockAcquire(ClogBufferLocks[slotno], LW_EXCLUSIVE);
/*
* Check to see if someone else already did the write, or took the
* buffer away from us. If so, do nothing. NOTE: we really should
* never see WRITE_IN_PROGRESS here, since that state should only
* occur while the writer is holding the buffer lock. But accept it
* so that we have a recovery path if a writer aborts.
*/
if (ClogCtl->page_number[slotno] != pageno ||
(ClogCtl->page_status[slotno] != CLOG_PAGE_DIRTY &&
ClogCtl->page_status[slotno] != CLOG_PAGE_WRITE_IN_PROGRESS))
{
LWLockRelease(ClogBufferLocks[slotno]);
LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
return;
}
/*
* Mark the slot write-busy. After this point, a transaction status
* update on this page will mark it dirty again. NB: we are assuming
* that read/write of the page status field is atomic, since we change
* the state while not holding control lock. However, we cannot set
* this state any sooner, or we'd possibly fool a previous writer into
* thinking he's successfully dumped the page when he hasn't.
* (Scenario: other writer starts, page is redirtied, we come along
* and set WRITE_IN_PROGRESS again, other writer completes and sets
* CLEAN because redirty info has been lost, then we think it's clean
* too.)
*/
ClogCtl->page_status[slotno] = CLOG_PAGE_WRITE_IN_PROGRESS;
/* Okay, do the write */
ok = CLOGPhysicalWritePage(pageno, slotno);
/* Re-acquire shared control lock and update page state */
LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
Assert(ClogCtl->page_number[slotno] == pageno &&
(ClogCtl->page_status[slotno] == CLOG_PAGE_WRITE_IN_PROGRESS ||
ClogCtl->page_status[slotno] == CLOG_PAGE_DIRTY));
/* Cannot set CLEAN if someone re-dirtied page since write started */
if (ClogCtl->page_status[slotno] == CLOG_PAGE_WRITE_IN_PROGRESS)
ClogCtl->page_status[slotno] = ok ? CLOG_PAGE_CLEAN : CLOG_PAGE_DIRTY;
LWLockRelease(ClogBufferLocks[slotno]);
/* Now it's okay to elog if we failed */
if (!ok)
CLOGReportIOError(pageno, InvalidTransactionId);
}
/*
* Physical read of a (previously existing) page into a buffer slot
*
* On failure, we cannot just elog(ERROR) since caller has put state in
* shared memory that must be undone. So, we return FALSE and save enough
* info in static variables to let CLOGReportIOError make the report.
*
* For now, assume it's not worth keeping a file pointer open across
* read/write operations. We could cache one virtual file pointer ...
*/
static bool
CLOGPhysicalReadPage(int pageno, int slotno)
{
int segno = pageno / CLOG_PAGES_PER_SEGMENT;
int rpageno = pageno % CLOG_PAGES_PER_SEGMENT;
int offset = rpageno * CLOG_BLCKSZ;
char path[MAXPGPATH];
int fd;
ClogFileName(path, segno);
/*
* In a crash-and-restart situation, it's possible for us to receive
* commands to set the commit status of transactions whose bits are in
* already-truncated segments of the commit log (see notes in
* CLOGPhysicalWritePage). Hence, if we are InRecovery, allow the
* case where the file doesn't exist, and return zeroes instead.
*/
fd = BasicOpenFile(path, O_RDWR | PG_BINARY, S_IRUSR | S_IWUSR);
if (fd < 0)
{
if (errno != ENOENT || !InRecovery)
{
clog_errcause = CLOG_OPEN_FAILED;
clog_errno = errno;
return false;
}
elog(LOG, "clog file %s doesn't exist, reading as zeroes", path);
MemSet(ClogCtl->page_buffer[slotno], 0, CLOG_BLCKSZ);
return true;
}
if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
{
clog_errcause = CLOG_SEEK_FAILED;
clog_errno = errno;
return false;
}
errno = 0;
if (read(fd, ClogCtl->page_buffer[slotno], CLOG_BLCKSZ) != CLOG_BLCKSZ)
{
clog_errcause = CLOG_READ_FAILED;
clog_errno = errno;
return false;
}
close(fd);
return true;
}
/*
* Physical write of a page from a buffer slot
*
* On failure, we cannot just elog(ERROR) since caller has put state in
* shared memory that must be undone. So, we return FALSE and save enough
* info in static variables to let CLOGReportIOError make the report.
*
* For now, assume it's not worth keeping a file pointer open across
* read/write operations. We could cache one virtual file pointer ...
*/
static bool
CLOGPhysicalWritePage(int pageno, int slotno)
{
int segno = pageno / CLOG_PAGES_PER_SEGMENT;
int rpageno = pageno % CLOG_PAGES_PER_SEGMENT;
int offset = rpageno * CLOG_BLCKSZ;
char path[MAXPGPATH];
int fd;
ClogFileName(path, segno);
/*
* If the file doesn't already exist, we should create it. It is
* possible for this to need to happen when writing a page that's not
* first in its segment; we assume the OS can cope with that. (Note:
* it might seem that it'd be okay to create files only when
* ZeroCLOGPage is called for the first page of a segment. However,
* if after a crash and restart the REDO logic elects to replay the
* log from a checkpoint before the latest one, then it's possible
* that we will get commands to set transaction status of transactions
* that have already been truncated from the commit log. Easiest way
* to deal with that is to accept references to nonexistent files here
* and in CLOGPhysicalReadPage.)
*/
fd = BasicOpenFile(path, O_RDWR | PG_BINARY, S_IRUSR | S_IWUSR);
if (fd < 0)
{
if (errno != ENOENT)
{
clog_errcause = CLOG_OPEN_FAILED;
clog_errno = errno;
return false;
}
fd = BasicOpenFile(path, O_RDWR | O_CREAT | O_EXCL | PG_BINARY,
S_IRUSR | S_IWUSR);
if (fd < 0)
{
clog_errcause = CLOG_CREATE_FAILED;
clog_errno = errno;
return false;
}
}
if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
{
clog_errcause = CLOG_SEEK_FAILED;
clog_errno = errno;
return false;
}
errno = 0;
if (write(fd, ClogCtl->page_buffer[slotno], CLOG_BLCKSZ) != CLOG_BLCKSZ)
{
/* if write didn't set errno, assume problem is no disk space */
if (errno == 0)
errno = ENOSPC;
clog_errcause = CLOG_WRITE_FAILED;
clog_errno = errno;
return false;
}
close(fd);
return true;
}
/*
* Issue the error message after failure of CLOGPhysicalReadPage or
* CLOGPhysicalWritePage. Call this after cleaning up shared-memory state.
*/
static void
CLOGReportIOError(int pageno, TransactionId xid)
{
int segno = pageno / CLOG_PAGES_PER_SEGMENT;
int rpageno = pageno % CLOG_PAGES_PER_SEGMENT;
int offset = rpageno * CLOG_BLCKSZ;
char path[MAXPGPATH];
/* XXX TODO: provide xid as context in error messages */
ClogFileName(path, segno);
errno = clog_errno;
switch (clog_errcause)
{
case CLOG_OPEN_FAILED:
elog(ERROR, "open of %s failed: %m", path);
break;
case CLOG_CREATE_FAILED:
elog(ERROR, "creation of file %s failed: %m", path);
break;
case CLOG_SEEK_FAILED:
elog(ERROR, "lseek of clog file %u, offset %u failed: %m",
segno, offset);
break;
case CLOG_READ_FAILED:
elog(ERROR, "read of clog file %u, offset %u failed: %m",
segno, offset);
break;
case CLOG_WRITE_FAILED:
elog(ERROR, "write of clog file %u, offset %u failed: %m",
segno, offset);
break;
default:
/* can't get here, we trust */
elog(ERROR, "unknown CLOG I/O error");
break;
}
}
/*
* Select the slot to re-use when we need a free slot.
*
* The target page number is passed because we need to consider the
* possibility that some other process reads in the target page while
* we are doing I/O to free a slot. Hence, check or recheck to see if
* any slot already holds the target page, and return that slot if so.
* Thus, the returned slot is *either* a slot already holding the pageno
* (could be any state except EMPTY), *or* a freeable slot (state EMPTY
* or CLEAN).
*
* Control lock must be held at entry, and will be held at exit.
*/
static int
SelectLRUCLOGPage(int pageno)
{
/* Outer loop handles restart after I/O */
for (;;)
{
int slotno;
int bestslot = 0;
unsigned int bestcount = 0;
/* See if page already has a buffer assigned */
for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
{
if (ClogCtl->page_number[slotno] == pageno &&
ClogCtl->page_status[slotno] != CLOG_PAGE_EMPTY)
return slotno;
}
/*
* If we find any EMPTY slot, just select that one. Else locate
* the least-recently-used slot that isn't the latest CLOG page.
*/
for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
{
if (ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY)
return slotno;
if (ClogCtl->page_lru_count[slotno] > bestcount &&
ClogCtl->page_number[slotno] != ClogCtl->latest_page_number)
{
bestslot = slotno;
bestcount = ClogCtl->page_lru_count[slotno];
}
}
/*
* If the selected page is clean, we're set.
*/
if (ClogCtl->page_status[bestslot] == CLOG_PAGE_CLEAN)
return bestslot;
/*
* We need to do I/O. Normal case is that we have to write it
* out, but it's possible in the worst case to have selected a
* read-busy page. In that case we use ReadCLOGPage to wait for
* the read to complete.
*/
if (ClogCtl->page_status[bestslot] == CLOG_PAGE_READ_IN_PROGRESS)
(void) ReadCLOGPage(ClogCtl->page_number[bestslot],
InvalidTransactionId);
else
WriteCLOGPage(bestslot);
/*
* Now loop back and try again. This is the easiest way of
* dealing with corner cases such as the victim page being
* re-dirtied while we wrote it.
*/
}
}
/*
* This must be called ONCE during postmaster or standalone-backend startup,
* after StartupXLOG has initialized ShmemVariableCache->nextXid.
@ -831,7 +211,7 @@ StartupCLOG(void)
/*
* Initialize our idea of the latest page number.
*/
ClogCtl->latest_page_number = TransactionIdToPage(ShmemVariableCache->nextXid);
SimpleLruSetLatestPage(ClogCtl, TransactionIdToPage(ShmemVariableCache->nextXid));
}
/*
@ -840,18 +220,7 @@ StartupCLOG(void)
void
ShutdownCLOG(void)
{
int slotno;
LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
{
WriteCLOGPage(slotno);
Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY ||
ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
}
LWLockRelease(CLogControlLock);
SimpleLruFlush(ClogCtl, false);
}
/*
@ -860,21 +229,7 @@ ShutdownCLOG(void)
void
CheckPointCLOG(void)
{
int slotno;
LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
{
WriteCLOGPage(slotno);
/*
* We cannot assert that the slot is clean now, since another
* process might have re-dirtied it already. That's okay.
*/
}
LWLockRelease(CLogControlLock);
SimpleLruFlush(ClogCtl, true);
}
@ -901,12 +256,12 @@ ExtendCLOG(TransactionId newestXact)
pageno = TransactionIdToPage(newestXact);
LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
LWLockAcquire(ClogCtl->locks->ControlLock, LW_EXCLUSIVE);
/* Zero the page and make an XLOG entry about it */
ZeroCLOGPage(pageno, true);
LWLockRelease(CLogControlLock);
LWLockRelease(ClogCtl->locks->ControlLock);
}
@ -928,126 +283,15 @@ void
TruncateCLOG(TransactionId oldestXact)
{
int cutoffPage;
int slotno;
/*
* The cutoff point is the start of the segment containing oldestXact.
* We pass the *page* containing oldestXact to SimpleLruTruncate.
*/
oldestXact -= oldestXact % CLOG_XACTS_PER_SEGMENT;
cutoffPage = TransactionIdToPage(oldestXact);
if (!ScanCLOGDirectory(cutoffPage, false))
return; /* nothing to remove */
/* Perform a forced CHECKPOINT */
CreateCheckPoint(false, true);
/*
* Scan CLOG shared memory and remove any pages preceding the cutoff
* page, to ensure we won't rewrite them later. (Any dirty pages
* should have been flushed already during the checkpoint, we're just
* being extra careful here.)
*/
LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
restart:;
/*
* While we are holding the lock, make an important safety check: the
* planned cutoff point must be <= the current CLOG endpoint page.
* Otherwise we have already wrapped around, and proceeding with the
* truncation would risk removing the current CLOG segment.
*/
if (CLOGPagePrecedes(ClogCtl->latest_page_number, cutoffPage))
{
LWLockRelease(CLogControlLock);
elog(LOG, "unable to truncate commit log: apparent wraparound");
return;
}
for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
{
if (ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY)
continue;
if (!CLOGPagePrecedes(ClogCtl->page_number[slotno], cutoffPage))
continue;
/*
* If page is CLEAN, just change state to EMPTY (expected case).
*/
if (ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN)
{
ClogCtl->page_status[slotno] = CLOG_PAGE_EMPTY;
continue;
}
/*
* Hmm, we have (or may have) I/O operations acting on the page,
* so we've got to wait for them to finish and then start again.
* This is the same logic as in SelectLRUCLOGPage.
*/
if (ClogCtl->page_status[slotno] == CLOG_PAGE_READ_IN_PROGRESS)
(void) ReadCLOGPage(ClogCtl->page_number[slotno],
InvalidTransactionId);
else
WriteCLOGPage(slotno);
goto restart;
}
LWLockRelease(CLogControlLock);
/* Now we can remove the old CLOG segment(s) */
(void) ScanCLOGDirectory(cutoffPage, true);
SimpleLruTruncate(ClogCtl, cutoffPage);
}
/*
* TruncateCLOG subroutine: scan CLOG directory for removable segments.
* Actually remove them iff doDeletions is true. Return TRUE iff any
* removable segments were found. Note: no locking is needed.
*/
static bool
ScanCLOGDirectory(int cutoffPage, bool doDeletions)
{
bool found = false;
DIR *cldir;
struct dirent *clde;
int segno;
int segpage;
char path[MAXPGPATH];
cldir = opendir(ClogDir);
if (cldir == NULL)
elog(ERROR, "could not open transaction-commit log directory (%s): %m",
ClogDir);
errno = 0;
while ((clde = readdir(cldir)) != NULL)
{
if (strlen(clde->d_name) == 4 &&
strspn(clde->d_name, "0123456789ABCDEF") == 4)
{
segno = (int) strtol(clde->d_name, NULL, 16);
segpage = segno * CLOG_PAGES_PER_SEGMENT;
if (CLOGPagePrecedes(segpage, cutoffPage))
{
found = true;
if (doDeletions)
{
elog(LOG, "removing commit log file %s", clde->d_name);
snprintf(path, MAXPGPATH, "%s/%s", ClogDir, clde->d_name);
unlink(path);
}
}
}
errno = 0;
}
if (errno)
elog(ERROR, "could not read transaction-commit log directory (%s): %m",
ClogDir);
closedir(cldir);
return found;
}
/*
* Decide which of two CLOG page numbers is "older" for truncation purposes.
@ -1107,13 +351,13 @@ clog_redo(XLogRecPtr lsn, XLogRecord *record)
memcpy(&pageno, XLogRecGetData(record), sizeof(int));
LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
LWLockAcquire(ClogCtl->locks->ControlLock, LW_EXCLUSIVE);
slotno = ZeroCLOGPage(pageno, false);
WriteCLOGPage(slotno);
Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
SimpleLruWritePage(ClogCtl, slotno);
/* Assert(ClogCtl->page_status[slotno] == SLRU_PAGE_CLEAN); */
LWLockRelease(CLogControlLock);
LWLockRelease(ClogCtl->locks->ControlLock);
}
}