public inbox for [email protected]
help / color / mirror / Atom feedFrom: Antonin Houska <[email protected]>
Subject: [PATCH 1/2] Use only xlogreader.c:XLogRead()
Date: Fri, 11 Nov 2019 12:07:22 +0200
The implementations in xlogutils.c and walsender.c are just renamed now, to be
removed by the following diff.
---
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index 6d8581919c..bbdb131525 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -17,6 +17,8 @@
*/
#include "postgres.h"
+#include <unistd.h>
+
#include "access/transam.h"
#include "access/xlogrecord.h"
#include "access/xlog_internal.h"
@@ -27,6 +29,7 @@
#ifndef FRONTEND
#include "miscadmin.h"
+#include "pgstat.h"
#include "utils/memutils.h"
#endif
@@ -626,7 +629,15 @@ ReadPageInternal(XLogReaderState *state, XLogRecPtr pageptr, int reqLen)
if (!XLogReaderValidatePageHeader(state, pageptr, (char *) hdr))
goto err;
- /* update read state information */
+ /*
+ * Update read state information.
+ *
+ * If XLogRead() is was called by ->read_page, it should have updated the
+ * ->seg fields accordingly (since we never request more than a single
+ * page, neither ws_segno nor ws_off should have advanced beyond
+ * targetSegNo and targetPageOff respectively). However it's not mandatory
+ * for ->read_page to call XLogRead().
+ */
state->seg.ws_segno = targetSegNo;
state->seg.ws_off = targetPageOff;
state->readLen = readLen;
@@ -1016,6 +1027,101 @@ out:
#endif /* FRONTEND */
+/*
+ * Read 'count' bytes from WAL fetched from timeline 'tli' into 'buf',
+ * starting at location 'startptr'. 'seg' is the last segment used,
+ * 'openSegment' is a callback to open the next segment and 'segcxt' is
+ * additional segment info that does not fit into 'seg'.
+ *
+ * 'errinfo' should point to XLogReadError structure which will receive error
+ * details in case the read fails.
+ *
+ * Returns true if succeeded, false if failed.
+ *
+ * XXX probably this should be improved to suck data directly from the
+ * WAL buffers when possible.
+ */
+bool
+XLogRead(char *buf, XLogRecPtr startptr, Size count, TimeLineID tli,
+ WALOpenSegment *seg, WALSegmentContext *segcxt,
+ WALSegmentOpen openSegment, XLogReadError *errinfo)
+{
+ char *p;
+ XLogRecPtr recptr;
+ Size nbytes;
+
+ p = buf;
+ recptr = startptr;
+ nbytes = count;
+
+ while (nbytes > 0)
+ {
+ int segbytes;
+ int readbytes;
+
+ seg->ws_off = XLogSegmentOffset(recptr, segcxt->ws_segsize);
+
+ if (seg->ws_file < 0 ||
+ !XLByteInSeg(recptr, seg->ws_segno, segcxt->ws_segsize) ||
+ tli != seg->ws_tli)
+ {
+ XLogSegNo nextSegNo;
+
+ /* Switch to another logfile segment */
+ if (seg->ws_file >= 0)
+ close(seg->ws_file);
+
+ XLByteToSeg(recptr, nextSegNo, segcxt->ws_segsize);
+
+ /* Open the next segment in the caller's way. */
+ openSegment(nextSegNo, segcxt, &tli, &seg->ws_file);
+
+ /* Update the current segment info. */
+ seg->ws_tli = tli;
+ seg->ws_segno = nextSegNo;
+ }
+
+ /* How many bytes are within this segment? */
+ if (nbytes > (segcxt->ws_segsize - seg->ws_off))
+ segbytes = segcxt->ws_segsize - seg->ws_off;
+ else
+ segbytes = nbytes;
+
+#ifndef FRONTEND
+ pgstat_report_wait_start(WAIT_EVENT_WAL_READ);
+#endif
+
+ /*
+ * Failure to read the data does not necessarily imply non-zero errno.
+ * Set it to zero so that caller can distinguish the failure that does
+ * not affect errno.
+ */
+ errno = 0;
+
+ readbytes = pg_pread(seg->ws_file, p, segbytes, seg->ws_off);
+
+#ifndef FRONTEND
+ pgstat_report_wait_end();
+#endif
+
+ if (readbytes <= 0)
+ {
+ errinfo->read_errno = errno;
+ errinfo->readbytes = readbytes;
+ errinfo->reqbytes = segbytes;
+ errinfo->seg = seg;
+ return false;
+ }
+
+ /* Update state for read */
+ recptr += readbytes;
+ nbytes -= readbytes;
+ p += readbytes;
+ }
+
+ return true;
+}
+
/* ----------------------------------------
* Functions for decoding the data and block references in a record.
* ----------------------------------------
diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c
index 5f1e5ba75d..31c0101d4b 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -653,8 +653,8 @@ XLogTruncateRelation(RelFileNode rnode, ForkNumber forkNum,
* frontend). Probably these should be merged at some point.
*/
static void
-XLogRead(char *buf, int segsize, TimeLineID tli, XLogRecPtr startptr,
- Size count)
+XLogReadOld(char *buf, int segsize, TimeLineID tli, XLogRecPtr startptr,
+ Size count)
{
char *p;
XLogRecPtr recptr;
@@ -896,6 +896,37 @@ XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage, uint32 wa
}
}
+/*
+ * Callback for XLogRead() to open the next segment.
+ */
+static void
+wal_segment_open(XLogSegNo nextSegNo, WALSegmentContext *segcxt,
+ TimeLineID *tli_p, int *file_p)
+{
+ TimeLineID tli = *tli_p;
+ char path[MAXPGPATH];
+ int file;
+
+ XLogFilePath(path, tli, nextSegNo, segcxt->ws_segsize);
+ file = BasicOpenFile(path, O_RDONLY | PG_BINARY);
+
+ if (file < 0)
+ {
+ if (errno == ENOENT)
+ ereport(ERROR,
+ (errcode_for_file_access(),
+ errmsg("requested WAL segment %s has already been removed",
+ path)));
+ else
+ ereport(ERROR,
+ (errcode_for_file_access(),
+ errmsg("could not open file \"%s\": %m",
+ path)));
+ }
+
+ *file_p = file;
+}
+
/*
* read_page callback for reading local xlog files
*
@@ -913,7 +944,9 @@ read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
{
XLogRecPtr read_upto,
loc;
+ TimeLineID tli;
int count;
+ XLogReadError errinfo;
loc = targetPagePtr + reqLen;
@@ -932,7 +965,7 @@ read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
read_upto = GetFlushRecPtr();
else
read_upto = GetXLogReplayRecPtr(&ThisTimeLineID);
- state->seg.ws_tli = ThisTimeLineID;
+ tli = ThisTimeLineID;
/*
* Check which timeline to get the record from.
@@ -982,14 +1015,14 @@ read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
read_upto = state->currTLIValidUntil;
/*
- * Setting ws_tli to our wanted record's TLI is slightly wrong;
- * the page might begin on an older timeline if it contains a
- * timeline switch, since its xlog segment will have been copied
- * from the prior timeline. This is pretty harmless though, as
- * nothing cares so long as the timeline doesn't go backwards. We
- * should read the page header instead; FIXME someday.
+ * Setting tli to our wanted record's TLI is slightly wrong; the
+ * page might begin on an older timeline if it contains a timeline
+ * switch, since its xlog segment will have been copied from the
+ * prior timeline. This is pretty harmless though, as nothing
+ * cares so long as the timeline doesn't go backwards. We should
+ * read the page header instead; FIXME someday.
*/
- state->seg.ws_tli = state->currTLI;
+ tli = state->currTLI;
/* No need to wait on a historical timeline */
break;
@@ -1020,9 +1053,41 @@ read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
* as 'count', read the whole page anyway. It's guaranteed to be
* zero-padded up to the page boundary if it's incomplete.
*/
- XLogRead(cur_page, state->segcxt.ws_segsize, state->seg.ws_tli, targetPagePtr,
- XLOG_BLCKSZ);
+ if (!XLogRead(cur_page, targetPagePtr, XLOG_BLCKSZ, tli, &state->seg,
+ &state->segcxt, wal_segment_open, &errinfo))
+ XLogReadRaiseError(&errinfo);
/* number of valid bytes in the buffer */
return count;
}
+
+#ifndef FRONTEND
+/*
+ * Backend-specific convenience code to handle read errors encountered by
+ * XLogRead().
+ */
+void
+XLogReadRaiseError(XLogReadError *errinfo)
+{
+ WALOpenSegment *seg = errinfo->seg;
+
+ if (errinfo->readbytes < 0)
+ {
+ errno = errinfo->read_errno;
+ ereport(ERROR,
+ (errcode_for_file_access(),
+ errmsg("could not read from log segment %s, offset %u, length %zu: %m",
+ XLogFileNameP(seg->ws_tli, seg->ws_segno),
+ seg->ws_off, (Size) errinfo->reqbytes)));
+ }
+ else
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("could not read from log segment %s, offset %u: length %zu",
+ XLogFileNameP(seg->ws_tli, seg->ws_segno),
+ seg->ws_off,
+ (Size) errinfo->reqbytes)));
+ }
+}
+#endif
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 7f5671504f..dd39fa9d17 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -248,9 +248,13 @@ static void LagTrackerWrite(XLogRecPtr lsn, TimestampTz local_flush_time);
static TimeOffset LagTrackerRead(int head, XLogRecPtr lsn, TimestampTz now);
static bool TransactionIdInRecentPast(TransactionId xid, uint32 epoch);
-static void XLogRead(WALSegmentContext *segcxt, char *buf, XLogRecPtr startptr, Size count);
+static void WalSndSegmentOpen(XLogSegNo nextSegNo, WALSegmentContext *segcxt,
+ TimeLineID *tli_p, int *file_p);
+static void XLogReadOld(WALSegmentContext *segcxt, char *buf,
+ XLogRecPtr startptr, Size count);
+
/* Initialize walsender process before entering the main command loop */
void
InitWalSender(void)
@@ -766,6 +770,8 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req
{
XLogRecPtr flushptr;
int count;
+ XLogReadError errinfo;
+ XLogSegNo segno;
XLogReadDetermineTimeline(state, targetPagePtr, reqLen);
sendTimeLineIsHistoric = (state->currTLI != ThisTimeLineID);
@@ -786,7 +792,27 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req
count = flushptr - targetPagePtr; /* part of the page available */
/* now actually read the data, we know it's there */
- XLogRead(sendCxt, cur_page, targetPagePtr, XLOG_BLCKSZ);
+ if (!XLogRead(cur_page,
+ targetPagePtr,
+ XLOG_BLCKSZ,
+ sendSeg->ws_tli, /* Pass the current TLI because only
+ * WalSndSegmentOpen controls whether new
+ * TLI is needed. */
+ sendSeg,
+ sendCxt,
+ WalSndSegmentOpen,
+ &errinfo))
+ XLogReadRaiseError(&errinfo);
+
+ /*
+ * After reading into the buffer, check that what we read was valid. We do
+ * this after reading, because even though the segment was present when we
+ * opened it, it might get recycled or removed while we read it. The
+ * read() succeeds in that case, but the data we tried to read might
+ * already have been overwritten with new WAL records.
+ */
+ XLByteToSeg(targetPagePtr, segno, sendCxt->ws_segsize);
+ CheckXLogRemoved(segno, sendSeg->ws_tli);
return count;
}
@@ -2362,7 +2388,7 @@ WalSndKill(int code, Datum arg)
* more than one.
*/
static void
-XLogRead(WALSegmentContext *segcxt, char *buf, XLogRecPtr startptr, Size count)
+XLogReadOld(WALSegmentContext *segcxt, char *buf, XLogRecPtr startptr, Size count)
{
char *p;
XLogRecPtr recptr;
@@ -2535,6 +2561,71 @@ retry:
}
}
+/*
+ * Callback for XLogRead() to open the next segment.
+ */
+void
+WalSndSegmentOpen(XLogSegNo nextSegNo, WALSegmentContext *segcxt,
+ TimeLineID *tli_p, int *file_p)
+{
+ char path[MAXPGPATH];
+
+ /*-------
+ * When reading from a historic timeline, and there is a timeline switch
+ * within this segment, read from the WAL segment belonging to the new
+ * timeline.
+ *
+ * For example, imagine that this server is currently on timeline 5, and
+ * we're streaming timeline 4. The switch from timeline 4 to 5 happened at
+ * 0/13002088. In pg_wal, we have these files:
+ *
+ * ...
+ * 000000040000000000000012
+ * 000000040000000000000013
+ * 000000050000000000000013
+ * 000000050000000000000014
+ * ...
+ *
+ * In this situation, when requested to send the WAL from segment 0x13, on
+ * timeline 4, we read the WAL from file 000000050000000000000013. Archive
+ * recovery prefers files from newer timelines, so if the segment was
+ * restored from the archive on this server, the file belonging to the old
+ * timeline, 000000040000000000000013, might not exist. Their contents are
+ * equal up to the switchpoint, because at a timeline switch, the used
+ * portion of the old segment is copied to the new file. -------
+ */
+ *tli_p = sendTimeLine;
+ if (sendTimeLineIsHistoric)
+ {
+ XLogSegNo endSegNo;
+
+ XLByteToSeg(sendTimeLineValidUpto, endSegNo, segcxt->ws_segsize);
+ if (sendSeg->ws_segno == endSegNo)
+ *tli_p = sendTimeLineNextTLI;
+ }
+
+ XLogFilePath(path, *tli_p, nextSegNo, segcxt->ws_segsize);
+ *file_p = BasicOpenFile(path, O_RDONLY | PG_BINARY);
+
+ if (*file_p < 0)
+ {
+ /*
+ * If the file is not found, assume it's because the standby asked for
+ * a too old WAL segment that has already been removed or recycled.
+ */
+ if (errno == ENOENT)
+ ereport(ERROR,
+ (errcode_for_file_access(),
+ errmsg("requested WAL segment %s has already been removed",
+ XLogFileNameP(*tli_p, nextSegNo))));
+ else
+ ereport(ERROR,
+ (errcode_for_file_access(),
+ errmsg("could not open file \"%s\": %m",
+ path)));
+ }
+}
+
/*
* Send out the WAL in its normal physical/stored form.
*
@@ -2552,6 +2643,8 @@ XLogSendPhysical(void)
XLogRecPtr startptr;
XLogRecPtr endptr;
Size nbytes;
+ XLogSegNo segno;
+ XLogReadError errinfo;
/* If requested switch the WAL sender to the stopping state. */
if (got_STOPPING)
@@ -2767,7 +2860,49 @@ XLogSendPhysical(void)
* calls.
*/
enlargeStringInfo(&output_message, nbytes);
- XLogRead(sendCxt, &output_message.data[output_message.len], startptr, nbytes);
+
+retry:
+ if (!XLogRead(&output_message.data[output_message.len],
+ startptr,
+ nbytes,
+ sendSeg->ws_tli, /* Pass the current TLI because only
+ * WalSndSegmentOpen controls whether new
+ * TLI is needed. */
+ sendSeg,
+ sendCxt,
+ WalSndSegmentOpen,
+ &errinfo))
+ XLogReadRaiseError(&errinfo);
+
+ /* See logical_read_xlog_page(). */
+ XLByteToSeg(startptr, segno, sendCxt->ws_segsize);
+ CheckXLogRemoved(segno, sendSeg->ws_tli);
+
+ /*
+ * During recovery, the currently-open WAL file might be replaced with the
+ * file of the same name retrieved from archive. So we always need to
+ * check what we read was valid after reading into the buffer. If it's
+ * invalid, we try to open and read the file again.
+ */
+ if (am_cascading_walsender)
+ {
+ WalSnd *walsnd = MyWalSnd;
+ bool reload;
+
+ SpinLockAcquire(&walsnd->mutex);
+ reload = walsnd->needreload;
+ walsnd->needreload = false;
+ SpinLockRelease(&walsnd->mutex);
+
+ if (reload && sendSeg->ws_file >= 0)
+ {
+ close(sendSeg->ws_file);
+ sendSeg->ws_file = -1;
+
+ goto retry;
+ }
+ }
+
output_message.len += nbytes;
output_message.data[output_message.len] = '\0';
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 1524e5eb1e..75142811b1 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -280,6 +280,46 @@ identify_target_directory(char *directory, char *fname)
return NULL; /* not reached */
}
+static void
+WALDumpOpenSegment(XLogSegNo nextSegNo, WALSegmentContext *segcxt,
+ TimeLineID *tli_p, int *file_p)
+{
+ TimeLineID tli = *tli_p;
+ char fname[MAXPGPATH];
+ int tries;
+
+ XLogFileName(fname, tli, nextSegNo, segcxt->ws_segsize);
+
+ /*
+ * In follow mode there is a short period of time after the server has
+ * written the end of the previous file before the new file is available.
+ * So we loop for 5 seconds looking for the file to appear before giving
+ * up.
+ */
+ for (tries = 0; tries < 10; tries++)
+ {
+ *file_p = open_file_in_directory(segcxt->ws_dir, fname);
+ if (*file_p >= 0)
+ break;
+ if (errno == ENOENT)
+ {
+ int save_errno = errno;
+
+ /* File not there yet, try again */
+ pg_usleep(500 * 1000);
+
+ errno = save_errno;
+ continue;
+ }
+ /* Any other error, fall through and fail */
+ break;
+ }
+
+ if (*file_p < 0)
+ fatal_error("could not find file \"%s\": %s",
+ fname, strerror(errno));
+}
+
/*
* Read count bytes from a segment file in the specified directory, for the
* given timeline, containing the specified record pointer; store the data in
@@ -411,6 +451,7 @@ XLogDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
{
XLogDumpPrivate *private = state->private_data;
int count = XLOG_BLCKSZ;
+ XLogReadError errinfo;
if (private->endptr != InvalidXLogRecPtr)
{
@@ -425,8 +466,23 @@ XLogDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
}
}
- XLogDumpXLogRead(state->segcxt.ws_dir, private->timeline, targetPagePtr,
- readBuff, count);
+ if (!XLogRead(readBuff, targetPagePtr, count, private->timeline,
+ &state->seg, &state->segcxt, WALDumpOpenSegment, &errinfo))
+ {
+ WALOpenSegment *seg = errinfo.seg;
+ char fname[MAXPGPATH];
+
+ XLogFileName(fname, seg->ws_tli, seg->ws_segno,
+ state->segcxt.ws_segsize);
+
+ if (errno != 0)
+ fatal_error("could not read from log file %s, offset %u, length %zu: %s",
+ fname, seg->ws_off, (Size) errinfo.reqbytes,
+ strerror(errinfo.read_errno));
+ else
+ fatal_error("could not read from log file %s, offset %u: length: %zu",
+ fname, seg->ws_off, (Size) errinfo.reqbytes);
+ }
return count;
}
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
index 1bbee386e8..f79f2c6980 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -218,6 +218,26 @@ extern XLogReaderState *XLogReaderAllocate(int wal_segment_size,
extern void XLogReaderFree(XLogReaderState *state);
/* Initialize supporting structures */
+/*
+ * Callback to open the specified WAL segment for reading.
+ *
+ * "nextSegNo" is the number of the segment to be opened.
+ *
+ * "segcxt" is additional information about the segment.
+ *
+ * "tli_p" is an input/output argument. XLogRead() uses it to pass the
+ * timeline in which the new segment should be found, but the callback can use
+ * it to return the TLI that it actually opened.
+ *
+ * "file_p" points to an address the segment file descriptor should be stored
+ * at.
+ *
+ * BasicOpenFile() is the preferred way to open the segment file in backend
+ * code, whereas open(2) should be used in frontend.
+ */
+typedef void (*WALSegmentOpen) (XLogSegNo nextSegNo, WALSegmentContext *segcxt,
+ TimeLineID *tli_p, int *file_p);
+
extern void WALOpenSegmentInit(WALOpenSegment *seg, WALSegmentContext *segcxt,
int segsize, const char *waldir);
@@ -232,6 +252,25 @@ extern bool XLogReaderValidatePageHeader(XLogReaderState *state,
#ifdef FRONTEND
extern XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr);
#endif /* FRONTEND */
+/*
+ * Error information that both backend and frontend caller can process.
+ *
+ * XXX Should the name be WALReadError? If so, we probably need to rename
+ * XLogRead() and XLogReadProcessError() too.
+ */
+typedef struct XLogReadError
+{
+ int read_errno; /* errno set by the last read(). */
+ int readbytes; /* Bytes read by the last read(). */
+ int reqbytes; /* Bytes requested to be read. */
+ WALOpenSegment *seg; /* Segment we tried to read from. */
+} XLogReadError;
+
+extern bool XLogRead(char *buf, XLogRecPtr startptr, Size count,
+ TimeLineID tli, WALOpenSegment *seg,
+ WALSegmentContext *segcxt, WALSegmentOpen openSegment,
+ XLogReadError *errinfo);
+
/* Functions for decoding an XLogRecord */
extern bool DecodeXLogRecord(XLogReaderState *state, XLogRecord *record,
diff --git a/src/include/access/xlogutils.h b/src/include/access/xlogutils.h
index 2df98e45b2..5c0cc910d7 100644
--- a/src/include/access/xlogutils.h
+++ b/src/include/access/xlogutils.h
@@ -54,4 +54,8 @@ extern int read_local_xlog_page(XLogReaderState *state,
extern void XLogReadDetermineTimeline(XLogReaderState *state,
XLogRecPtr wantPage, uint32 wantLength);
+
+#ifndef FRONTEND
+void XLogReadRaiseError(XLogReadError *errinfo);
+#endif
#endif
--=-=-=
Content-Type: text/x-diff
Content-Disposition: attachment;
filename=v09-0006-Remove-the-old-implemenations-of-XLogRead.patch
view thread (14+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected]
Subject: Re: [PATCH 1/2] Use only xlogreader.c:XLogRead()
In-Reply-To: <no-message-id-1883347@localhost>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox