slru.c

PostgreSQL7.4.6 for Linux

 *
 * 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
SlruPhysicalReadPage(SlruCtl ctl, int pageno, int slotno)
{
	SlruShared	shared = (SlruShared) ctl->shared;
	int			segno = pageno / SLRU_PAGES_PER_SEGMENT;
	int			rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
	int			offset = rpageno * BLCKSZ;
	char		path[MAXPGPATH];
	int			fd;

	SlruFileName(ctl, 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
	 * SlruPhysicalWritePage).	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)
		{
			slru_errcause = SLRU_OPEN_FAILED;
			slru_errno = errno;
			return false;
		}

		ereport(LOG,
				(errmsg("file \"%s\" doesn't exist, reading as zeroes",
						path)));
		MemSet(shared->page_buffer[slotno], 0, BLCKSZ);
		return true;
	}

	if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
	{
		slru_errcause = SLRU_SEEK_FAILED;
		slru_errno = errno;
		return false;
	}

	errno = 0;
	if (read(fd, shared->page_buffer[slotno], BLCKSZ) != BLCKSZ)
	{
		slru_errcause = SLRU_READ_FAILED;
		slru_errno = errno;
		return false;
	}

	close(fd);
	return true;
}

/*
 * Physical write of a page from a buffer slot
 *
 * On failure, we cannot just ereport(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 SlruReportIOError 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
SlruPhysicalWritePage(SlruCtl ctl, int pageno, int slotno)
{
	SlruShared	shared = (SlruShared) ctl->shared;
	int			segno = pageno / SLRU_PAGES_PER_SEGMENT;
	int			rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
	int			offset = rpageno * BLCKSZ;
	char		path[MAXPGPATH];
	int			fd;

	SlruFileName(ctl, 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
	 * SimpleLruZeroPage 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 SlruPhysicalReadPage.)
	 */
	fd = BasicOpenFile(path, O_RDWR | PG_BINARY, S_IRUSR | S_IWUSR);
	if (fd < 0)
	{
		if (errno != ENOENT)
		{
			slru_errcause = SLRU_OPEN_FAILED;
			slru_errno = errno;
			return false;
		}

		fd = BasicOpenFile(path, O_RDWR | O_CREAT | O_EXCL | PG_BINARY,
						   S_IRUSR | S_IWUSR);
		if (fd < 0)
		{
			slru_errcause = SLRU_CREATE_FAILED;
			slru_errno = errno;
			return false;
		}
	}

	if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
	{
		slru_errcause = SLRU_SEEK_FAILED;
		slru_errno = errno;
		return false;
	}

	errno = 0;
	if (write(fd, shared->page_buffer[slotno], BLCKSZ) != BLCKSZ)
	{
		/* if write didn't set errno, assume problem is no disk space */
		if (errno == 0)
			errno = ENOSPC;
		slru_errcause = SLRU_WRITE_FAILED;
		slru_errno = errno;
		return false;
	}

	close(fd);
	return true;
}

/*
 * Issue the error message after failure of SlruPhysicalReadPage or
 * SlruPhysicalWritePage.  Call this after cleaning up shared-memory state.
 */
static void
SlruReportIOError(SlruCtl ctl, int pageno, TransactionId xid)
{
	int			segno = pageno / SLRU_PAGES_PER_SEGMENT;
	int			rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
	int			offset = rpageno * BLCKSZ;
	char		path[MAXPGPATH];

	SlruFileName(ctl, path, segno);
	errno = slru_errno;
	switch (slru_errcause)
	{
		case SLRU_OPEN_FAILED:
			ereport(ERROR,
					(errcode_for_file_access(),
				errmsg("could not access status of transaction %u", xid),
					 errdetail("could not open file \"%s\": %m",
							   path)));
			break;
		case SLRU_CREATE_FAILED:
			ereport(ERROR,
					(errcode_for_file_access(),
				errmsg("could not access status of transaction %u", xid),
					 errdetail("could not create file \"%s\": %m",
							   path)));
			break;
		case SLRU_SEEK_FAILED:
			ereport(ERROR,
					(errcode_for_file_access(),
				errmsg("could not access status of transaction %u", xid),
				  errdetail("could not seek in file \"%s\" to offset %u: %m",
							path, offset)));
			break;
		case SLRU_READ_FAILED:
			ereport(ERROR,
					(errcode_for_file_access(),
				errmsg("could not access status of transaction %u", xid),
				   errdetail("could not read from file \"%s\" at offset %u: %m",
							 path, offset)));
			break;
		case SLRU_WRITE_FAILED:
			ereport(ERROR,
					(errcode_for_file_access(),
				errmsg("could not access status of transaction %u", xid),
				  errdetail("could not write to file \"%s\" at offset %u: %m",
							path, offset)));
			break;
		default:
			/* can't get here, we trust */
			elog(ERROR, "unrecognized SimpleLru error cause: %d",
				 (int) slru_errcause);
			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
SlruSelectLRUPage(SlruCtl ctl, int pageno)
{
	SlruShared	shared = (SlruShared) ctl->shared;

	/* 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 (shared->page_number[slotno] == pageno &&
				shared->page_status[slotno] != SLRU_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 page.
		 */
		for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
		{
			if (shared->page_status[slotno] == SLRU_PAGE_EMPTY)
				return slotno;
			if (shared->page_lru_count[slotno] > bestcount &&
				shared->page_number[slotno] != shared->latest_page_number)
			{
				bestslot = slotno;
				bestcount = shared->page_lru_count[slotno];
			}
		}

		/*
		 * If the selected page is clean, we're set.
		 */
		if (shared->page_status[bestslot] == SLRU_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 SimpleLruReadPage to wait
		 * for the read to complete.
		 */
		if (shared->page_status[bestslot] == SLRU_PAGE_READ_IN_PROGRESS)
			(void) SimpleLruReadPage(ctl, shared->page_number[bestslot],
									 InvalidTransactionId, false);
		else
			SimpleLruWritePage(ctl, 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
 */
void
SimpleLruSetLatestPage(SlruCtl ctl, int pageno)
{
	SlruShared	shared = (SlruShared) ctl->shared;

	shared->latest_page_number = pageno;
}

/*
 * This is called during checkpoint and postmaster/standalone-backend shutdown
 */
void
SimpleLruFlush(SlruCtl ctl, bool checkpoint)
{
#ifdef USE_ASSERT_CHECKING		/* only used in Assert() */
	SlruShared	shared = (SlruShared) ctl->shared;
#endif
	int			slotno;

	LWLockAcquire(ctl->locks->ControlLock, LW_EXCLUSIVE);

	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
	{
		SimpleLruWritePage(ctl, slotno);

		/*
		 * When called during a checkpoint, we cannot assert that the slot
		 * is clean now, since another process might have re-dirtied it
		 * already.  That's okay.
		 */
		Assert(checkpoint ||
			   shared->page_status[slotno] == SLRU_PAGE_EMPTY ||
			   shared->page_status[slotno] == SLRU_PAGE_CLEAN);
	}

	LWLockRelease(ctl->locks->ControlLock);
}

/*
 * Remove all segments before the one holding the passed page number
 *
 * When this is called, we know that the database logically contains no
 * reference to transaction IDs older than oldestXact.	However, we must
 * not remove any segment until we have performed a checkpoint, to ensure
 * that no such references remain on disk either; else a crash just after
 * the truncation might leave us with a problem.  Since CLOG segments hold
 * a large number of transactions, the opportunity to actually remove a
 * segment is fairly rare, and so it seems best not to do the checkpoint
 * unless we have confirmed that there is a removable segment.	Therefore
 * we issue the checkpoint command here, not in higher-level code as might
 * seem cleaner.
 */
void
SimpleLruTruncate(SlruCtl ctl, int cutoffPage)
{
	int			slotno;
	SlruShared	shared = (SlruShared) ctl->shared;

	/*
	 * The cutoff point is the start of the segment containing cutoffPage.
	 */
	cutoffPage -= cutoffPage % SLRU_PAGES_PER_SEGMENT;

	if (!SlruScanDirectory(ctl, cutoffPage, false))
		return;					/* nothing to remove */

	/* Perform a forced CHECKPOINT */
	CreateCheckPoint(false, true);

	/*
	 * Scan 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(ctl->locks->ControlLock, LW_EXCLUSIVE);

restart:;

	/*
	 * While we are holding the lock, make an important safety check: the
	 * planned cutoff point must be <= the current endpoint page.
	 * Otherwise we have already wrapped around, and proceeding with the
	 * truncation would risk removing the current segment.
	 */
	if (ctl->PagePrecedes(shared->latest_page_number, cutoffPage))
	{
		LWLockRelease(ctl->locks->ControlLock);
		ereport(LOG,
				(errmsg("could not truncate directory \"%s\": apparent wraparound",
						ctl->Dir)));
		return;
	}

	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
	{
		if (shared->page_status[slotno] == SLRU_PAGE_EMPTY)
			continue;
		if (!ctl->PagePrecedes(shared->page_number[slotno], cutoffPage))
			continue;

		/*
		 * If page is CLEAN, just change state to EMPTY (expected case).
		 */
		if (shared->page_status[slotno] == SLRU_PAGE_CLEAN)
		{
			shared->page_status[slotno] = SLRU_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 SlruSelectLRUPage.
		 */
		if (shared->page_status[slotno] == SLRU_PAGE_READ_IN_PROGRESS)
			(void) SimpleLruReadPage(ctl, shared->page_number[slotno],
									 InvalidTransactionId, false);
		else
			SimpleLruWritePage(ctl, slotno);
		goto restart;
	}

	LWLockRelease(ctl->locks->ControlLock);

	/* Now we can remove the old segment(s) */
	(void) SlruScanDirectory(ctl, cutoffPage, true);
}

/*
 * SlruTruncate subroutine: scan 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
SlruScanDirectory(SlruCtl ctl, int cutoffPage, bool doDeletions)
{
	bool		found = false;
	DIR		   *cldir;
	struct dirent *clde;
	int			segno;
	int			segpage;
	char		path[MAXPGPATH];

	cldir = AllocateDir(ctl->Dir);
	if (cldir == NULL)
		ereport(ERROR,
				(errcode_for_file_access(),
			   errmsg("could not open directory \"%s\": %m", ctl->Dir)));

	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 * SLRU_PAGES_PER_SEGMENT;
			if (ctl->PagePrecedes(segpage, cutoffPage))
			{
				found = true;
				if (doDeletions)
				{
					ereport(LOG,
							(errmsg("removing file \"%s/%s\"",
									ctl->Dir, clde->d_name)));
					snprintf(path, MAXPGPATH, "%s/%s", ctl->Dir, clde->d_name);
					unlink(path);
				}
			}
		}
		errno = 0;
	}
	if (errno)
		ereport(ERROR,
				(errcode_for_file_access(),
			   errmsg("could not read directory \"%s\": %m", ctl->Dir)));
	FreeDir(cldir);

	return found;
}