From: Antonin Houska <ah@cybertec.at>
To: pgsql-hackers@postgresql.org
Subject: Attempt to consolidate reading of XLOG page
Date: Thu, 11 Apr 2019 18:05:42 +0200
Message-ID: <14984.1554998742@spoje.net> (raw)
While working on the instance encryption I found it annoying to apply
decyption of XLOG page to three different functions. Attached is a patch that
tries to merge them all into one function, XLogRead(). The existing
implementations differ in the way new segment is opened. So I added a pointer
to callback function as a new argument. This callback handles the specific
ways to determine segment file name and to open the file.
I can split the patch into multiple diffs to make detailed review easier, but
first I'd like to hear if anything is seriously wrong about this
design. Thanks.
--
Antonin Houska
Web: https://www.cybertec-postgresql.comAttachments:
[text/x-diff] reuse_xlogread.patch (36.5K, ../14984.1554998742@spoje.net/2-reuse_xlogread.patch)
download | inline diff:diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.cindex f9a4960f8a..444b5bf910 100644--- a/src/backend/access/transam/twophase.c+++ b/src/backend/access/transam/twophase.c@@ -1369,7 +1369,7 @@ ParsePrepareRecord(uint8 info, char *xlrec, xl_xact_parsed_prepare *parsed)
bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage));
}
-+static XLogReadPos *readPos = NULL;
/*
* Reads 2PC data from xlog. During checkpoint this data will be moved to
@@ -1386,8 +1386,17 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len)
XLogReaderState *xlogreader;
char *errormsg;
- xlogreader = XLogReaderAllocate(wal_segment_size, &read_local_xlog_page,- NULL);++ /* First time through? */+ if (readPos == NULL)+ readPos = XLogReadInitPos();++ /*+ * read_local_xlog_page() eventually calls XLogRead(), so pass the initial+ * position.+ */+ xlogreader = XLogReaderAllocate(wal_segment_size, read_local_xlog_page,+ readPos);
if (!xlogreader)
ereport(ERROR,
(errcode(ERRCODE_OUT_OF_MEMORY),
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.cindex 9196aa3aae..7d0fdfba87 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"
@@ -26,6 +28,7 @@
#include "replication/origin.h"
#ifndef FRONTEND
+#include "pgstat.h"
#include "utils/memutils.h"
#endif
@@ -1005,6 +1008,191 @@ out:
#endif /* FRONTEND */
+/*+ * Initialize XLOG file position for callers of XLogRead().+ */+XLogReadPos *+XLogReadInitPos(void)+{+ XLogReadPos *pos = (XLogReadPos *) palloc(sizeof(XLogReadPos));++ pos->segFile = -1;+ pos->segNo = 0;+ pos->segOff = 0;+ pos->tli = 0;+ pos->dir = NULL;++ return pos;+}++#ifdef FRONTEND+/*+ * Currently only pg_waldump.c is supposed to set these variables.+ */+const char *progname;+int WalSegSz;++/*+ * This is a front-end counterpart of XLogFileNameP.+ */+static char *+XLogFileNameFE(TimeLineID tli, XLogSegNo segno)+{+ char *result = palloc(MAXFNAMELEN);++ XLogFileName(result, tli, segno, WalSegSz);+ return result;+}++/*+ * XXX pg_waldump.c needs this function. Is there a smart way to put it into+ * src/common?+ */+static void fatal_error(const char *fmt,...) pg_attribute_printf(1, 2);++static void+fatal_error(const char *fmt,...)+{+ va_list args;++ fflush(stdout);++ fprintf(stderr, _("%s: FATAL: "), progname);+ va_start(args, fmt);+ vfprintf(stderr, _(fmt), args);+ va_end(args);+ fputc('\n', stderr);++ exit(EXIT_FAILURE);+}+#endif /* FRONTEND */++/*+ * Read 'count' bytes from WAL into 'buf', starting at location 'startptr'. If+ * tli is passed, get the data from timeline *tli. 'pos' is the current+ * position in the XLOG file and openSegment is a callback that opens the next+ * segment for reading.+ *+ * XXX probably this should be improved to suck data directly from the+ * WAL buffers when possible.+ *+ * Will open, and keep open, one WAL segment stored in the global file+ * descriptor sendFile. This means if XLogRead is used once, there will+ * always be one descriptor left open until the process ends, but never+ * more than one.+ */+void+XLogRead(char *buf, XLogRecPtr startptr, Size count,+ TimeLineID *tli, XLogReadPos *pos, XLogOpenSegment openSegment)+{+ char *p;+ XLogRecPtr recptr;+ Size nbytes;++ p = buf;+ recptr = startptr;+ nbytes = count;++ while (nbytes > 0)+ {+ uint32 startoff;+ int segbytes;+ int readbytes;+ int segsize;++#ifndef FRONTEND+ segsize = wal_segment_size;+#else+ segsize = WalSegSz;+#endif++ startoff = XLogSegmentOffset(recptr, segsize);++ if (pos->segFile < 0 ||+ !XLByteInSeg(recptr, pos->segNo, segsize) ||+ (tli != NULL && *tli != pos->tli))+ {+ XLogSegNo nextSegNo;++ /* Switch to another logfile segment */+ if (pos->segFile >= 0)+ close(pos->segFile);++ XLByteToSeg(recptr, nextSegNo, segsize);++ /* Open the next segment in the caller's way. */+ openSegment(nextSegNo, tli, pos);+ }++ /* Need to seek in the file? */+ if (pos->segOff != startoff)+ {+ if (lseek(pos->segFile, (off_t) startoff, SEEK_SET) < 0)+#ifndef FRONTEND+ ereport(ERROR,+ (errcode_for_file_access(),+ errmsg("could not seek in log segment %s to offset %u: %m",+ XLogFileNameP(pos->tli, pos->segNo),+ startoff)));+#else+ fatal_error("could not seek in log segment %s to offset %u",+ XLogFileNameFE(pos->tli, pos->segNo), startoff);+#endif+ pos->segOff = startoff;+ }++ /* How many bytes are within this segment? */+ if (nbytes > (segsize - startoff))+ segbytes = segsize - startoff;+ else+ segbytes = nbytes;++#ifndef FRONTEND+ pgstat_report_wait_start(WAIT_EVENT_WAL_READ);+#endif++ readbytes = read(pos->segFile, p, segbytes);++#ifndef FRONTEND+ pgstat_report_wait_end();+#endif+ if (readbytes < 0)+ {+#ifndef FRONTEND+ ereport(ERROR,+ (errcode_for_file_access(),+ errmsg("could not read from log segment %s, offset %u, length %zu: %m",+ XLogFileNameP(pos->tli, pos->segNo), pos->segOff,+ (Size) segbytes)));+#else+ fatal_error("could not read from log segment %s, offset %u, length %zu",+ XLogFileNameFE(pos->tli, pos->segNo), pos->segOff,+ (Size) segbytes);+#endif+ }+ else if (readbytes == 0)+ {+#ifndef FRONTEND+ ereport(ERROR,+ (errcode(ERRCODE_DATA_CORRUPTED),+ errmsg("could not read from log segment %s, offset %u: read %d of %zu",+ XLogFileNameP(pos->tli, pos->segNo), pos->segOff,+ readbytes, (Size) segbytes)));+#else+ fatal_error("could not read from log segment %s, offset %u: read %d of %zu",+ XLogFileNameFE(pos->tli, pos->segNo), pos->segOff,+ readbytes, (Size) segbytes);+#endif+ }++ /* Update state for read */+ recptr += readbytes;++ pos->segOff += readbytes;+ nbytes -= readbytes;+ p += readbytes;+ }+}
/* ----------------------------------------
* 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.cindex 10a663bae6..4f29c89c06 100644--- a/src/backend/access/transam/xlogutils.c+++ b/src/backend/access/transam/xlogutils.c@@ -639,128 +639,6 @@ XLogTruncateRelation(RelFileNode rnode, ForkNumber forkNum,
forget_invalid_pages(rnode, forkNum, nblocks);
}
-/*- * Read 'count' bytes from WAL into 'buf', starting at location 'startptr'- * in timeline 'tli'.- *- * Will open, and keep open, one WAL segment stored in the static file- * descriptor 'sendFile'. This means if XLogRead is used once, there will- * always be one descriptor left open until the process ends, but never- * more than one.- *- * XXX This is very similar to pg_waldump's XLogDumpXLogRead and to XLogRead- * in walsender.c but for small differences (such as lack of elog() in- * frontend). Probably these should be merged at some point.- */-static void-XLogRead(char *buf, int segsize, TimeLineID tli, XLogRecPtr startptr,- Size count)-{- char *p;- XLogRecPtr recptr;- Size nbytes;-- /* state maintained across calls */- static int sendFile = -1;- static XLogSegNo sendSegNo = 0;- static TimeLineID sendTLI = 0;- static uint32 sendOff = 0;-- Assert(segsize == wal_segment_size);-- p = buf;- recptr = startptr;- nbytes = count;-- while (nbytes > 0)- {- uint32 startoff;- int segbytes;- int readbytes;-- startoff = XLogSegmentOffset(recptr, segsize);-- /* Do we need to switch to a different xlog segment? */- if (sendFile < 0 || !XLByteInSeg(recptr, sendSegNo, segsize) ||- sendTLI != tli)- {- char path[MAXPGPATH];-- if (sendFile >= 0)- close(sendFile);-- XLByteToSeg(recptr, sendSegNo, segsize);-- XLogFilePath(path, tli, sendSegNo, segsize);-- sendFile = BasicOpenFile(path, O_RDONLY | PG_BINARY);-- if (sendFile < 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)));- }- sendOff = 0;- sendTLI = tli;- }-- /* Need to seek in the file? */- if (sendOff != startoff)- {- if (lseek(sendFile, (off_t) startoff, SEEK_SET) < 0)- {- char path[MAXPGPATH];- int save_errno = errno;-- XLogFilePath(path, tli, sendSegNo, segsize);- errno = save_errno;- ereport(ERROR,- (errcode_for_file_access(),- errmsg("could not seek in log segment %s to offset %u: %m",- path, startoff)));- }- sendOff = startoff;- }-- /* How many bytes are within this segment? */- if (nbytes > (segsize - startoff))- segbytes = segsize - startoff;- else- segbytes = nbytes;-- pgstat_report_wait_start(WAIT_EVENT_WAL_READ);- readbytes = read(sendFile, p, segbytes);- pgstat_report_wait_end();- if (readbytes <= 0)- {- char path[MAXPGPATH];- int save_errno = errno;-- XLogFilePath(path, tli, sendSegNo, segsize);- errno = save_errno;- ereport(ERROR,- (errcode_for_file_access(),- errmsg("could not read from log segment %s, offset %u, length %lu: %m",- path, sendOff, (unsigned long) segbytes)));- }-- /* Update state for read */- recptr += readbytes;-- sendOff += readbytes;- nbytes -= readbytes;- p += readbytes;- }-}-
/*
* Determine which timeline to read an xlog page from and set the
* XLogReaderState's currTLI to that timeline ID.
@@ -896,6 +774,37 @@ XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage, uint32 wa
}
}
+/*+ * Callback for XLogRead() to open the next segment.+ */+static void+read_local_xlog_page_open_segment(XLogSegNo segNo, TimeLineID *tli,+ XLogReadPos *pos)+{+ char path[MAXPGPATH];++ XLogFilePath(path, *tli, segNo, wal_segment_size);+ pos->segFile = BasicOpenFile(path, O_RDONLY | PG_BINARY);++ if (pos->segFile < 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)));+ }++ pos->segNo = segNo;+ pos->segOff = 0;+ pos->tli = *tli;+}+
/*
* read_page callback for reading local xlog files
*
@@ -1017,14 +926,16 @@ read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
count = read_upto - targetPagePtr;
}
+ Assert(state->wal_segment_size == wal_segment_size);+
/*
* Even though we just determined how much of the page can be validly read
* 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->wal_segment_size, *pageTLI, targetPagePtr,- XLOG_BLCKSZ);-+ XLogRead(cur_page, targetPagePtr, XLOG_BLCKSZ, pageTLI,+ (XLogReadPos *) state->private_data,+ read_local_xlog_page_open_segment);
/* number of valid bytes in the buffer */
return count;
}
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.cindex 424fe86a1b..20c1ad4a35 100644--- a/src/backend/replication/logical/logical.c+++ b/src/backend/replication/logical/logical.c@@ -124,6 +124,7 @@ StartupDecodingContext(List *output_plugin_options,
bool need_full_snapshot,
bool fast_forward,
XLogPageReadCB read_page,
+ void *read_page_arg,
LogicalOutputPluginWriterPrepareWrite prepare_write,
LogicalOutputPluginWriterWrite do_write,
LogicalOutputPluginWriterUpdateProgress update_progress)
@@ -172,14 +173,13 @@ StartupDecodingContext(List *output_plugin_options,
ctx->slot = slot;
- ctx->reader = XLogReaderAllocate(wal_segment_size, read_page, ctx);+ ctx->reader = XLogReaderAllocate(wal_segment_size, read_page,+ read_page_arg);
if (!ctx->reader)
ereport(ERROR,
(errcode(ERRCODE_OUT_OF_MEMORY),
errmsg("out of memory")));
- ctx->reader->private_data = ctx;-
ctx->reorder = ReorderBufferAllocate();
ctx->snapshot_builder =
AllocateSnapshotBuilder(ctx->reorder, xmin_horizon, start_lsn,
@@ -234,6 +234,7 @@ CreateInitDecodingContext(char *plugin,
bool need_full_snapshot,
XLogRecPtr restart_lsn,
XLogPageReadCB read_page,
+ void *read_page_arg,
LogicalOutputPluginWriterPrepareWrite prepare_write,
LogicalOutputPluginWriterWrite do_write,
LogicalOutputPluginWriterUpdateProgress update_progress)
@@ -330,8 +331,8 @@ CreateInitDecodingContext(char *plugin,
ctx = StartupDecodingContext(NIL, restart_lsn, xmin_horizon,
need_full_snapshot, false,
- read_page, prepare_write, do_write,- update_progress);+ read_page, read_page_arg, prepare_write,+ do_write, update_progress);
/* call output plugin initialization callback */
old_context = MemoryContextSwitchTo(ctx->context);
@@ -376,6 +377,7 @@ CreateDecodingContext(XLogRecPtr start_lsn,
List *output_plugin_options,
bool fast_forward,
XLogPageReadCB read_page,
+ void *read_page_arg,
LogicalOutputPluginWriterPrepareWrite prepare_write,
LogicalOutputPluginWriterWrite do_write,
LogicalOutputPluginWriterUpdateProgress update_progress)
@@ -428,8 +430,8 @@ CreateDecodingContext(XLogRecPtr start_lsn,
ctx = StartupDecodingContext(output_plugin_options,
start_lsn, InvalidTransactionId, false,
- fast_forward, read_page, prepare_write,- do_write, update_progress);+ fast_forward, read_page, read_page_arg,+ prepare_write, do_write, update_progress);
/* call output plugin initialization callback */
old_context = MemoryContextSwitchTo(ctx->context);
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.cindex d974400d6e..b2f30d53f5 100644--- a/src/backend/replication/logical/logicalfuncs.c+++ b/src/backend/replication/logical/logicalfuncs.c@@ -248,13 +248,20 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
PG_TRY();
{
- /* restart at slot's confirmed_flush */+ /*+ * Restart at slot's confirmed_flush.+ *+ * logical_read_local_xlog_page() eventually calls XLogRead(), so set+ * the initial position.+ */
ctx = CreateDecodingContext(InvalidXLogRecPtr,
options,
false,
logical_read_local_xlog_page,
+ XLogReadInitPos(),
LogicalOutputPrepareWrite,
- LogicalOutputWrite, NULL);+ LogicalOutputWrite,+ NULL);
MemoryContextSwitchTo(oldcontext);
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.cindex 182fe5bc82..dbcaa9c1d8 100644--- a/src/backend/replication/slotfuncs.c+++ b/src/backend/replication/slotfuncs.c@@ -14,6 +14,7 @@
#include "access/htup_details.h"
#include "access/xlog_internal.h"
+#include "access/xlogreader.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "replication/decode.h"
@@ -144,8 +145,9 @@ create_logical_replication_slot(char *name, char *plugin,
ctx = CreateInitDecodingContext(plugin, NIL,
false, /* do not build snapshot */
restart_lsn,
- logical_read_local_xlog_page, NULL, NULL,- NULL);+ logical_read_local_xlog_page,+ XLogReadInitPos(),+ NULL, NULL, NULL);
/* build initial snapshot, might take a while */
DecodingContextFindStartpoint(ctx);
@@ -401,11 +403,15 @@ pg_logical_replication_slot_advance(XLogRecPtr moveto)
* Create our decoding context in fast_forward mode, passing start_lsn
* as InvalidXLogRecPtr, so that we start processing from my slot's
* confirmed_flush.
+ *+ * logical_read_local_xlog_page() eventually calls XLogRead(), so set+ * the initial position.
*/
ctx = CreateDecodingContext(InvalidXLogRecPtr,
NIL,
true, /* fast_forward */
logical_read_local_xlog_page,
+ XLogReadInitPos(),
NULL, NULL, NULL);
/*
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.cindex aae6adc15c..56f9ae88b1 100644--- a/src/backend/replication/walsender.c+++ b/src/backend/replication/walsender.c@@ -128,16 +128,7 @@ bool log_replication_commands = false;
*/
bool wake_wal_senders = false;
-/*- * These variables are used similarly to openLogFile/SegNo/Off,- * but for walsender to read the XLOG.- */-static int sendFile = -1;-static XLogSegNo sendSegNo = 0;-static uint32 sendOff = 0;--/* Timeline ID of the currently open file */-static TimeLineID curFileTimeLine = 0;+static XLogReadPos *sendPos= NULL;
/*
* These variables keep track of the state of the timeline we're currently
@@ -256,7 +247,8 @@ 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(char *buf, XLogRecPtr startptr, Size count);+static void WalSndOpenSegment(XLogSegNo segNo, TimeLineID *tli,+ XLogReadPos *pos);
/* Initialize walsender process before entering the main command loop */
@@ -285,6 +277,9 @@ InitWalSender(void)
/* Initialize empty timestamp buffer for lag tracking. */
lag_tracker = MemoryContextAllocZero(TopMemoryContext, sizeof(LagTracker));
++ /* Make sure we can remember the current read position in XLOG. */+ sendPos = XLogReadInitPos();
}
/*
@@ -301,10 +296,10 @@ WalSndErrorCleanup(void)
ConditionVariableCancelSleep();
pgstat_report_wait_end();
- if (sendFile >= 0)+ if (sendPos && sendPos->segFile >= 0)
{
- close(sendFile);- sendFile = -1;+ close(sendPos->segFile);+ sendPos->segFile = -1;
}
if (MyReplicationSlot != NULL)
@@ -787,7 +782,8 @@ 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(cur_page, targetPagePtr, XLOG_BLCKSZ);+ XLogRead(cur_page, targetPagePtr, XLOG_BLCKSZ, NULL, sendPos,+ WalSndOpenSegment);
return count;
}
@@ -933,9 +929,13 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
need_full_snapshot = true;
}
+ /*+ * logical_read_xlog_page() eventually calls XLogRead(), so pass the+ * initial position.+ */
ctx = CreateInitDecodingContext(cmd->plugin, NIL, need_full_snapshot,
InvalidXLogRecPtr,
- logical_read_xlog_page,+ logical_read_xlog_page, sendPos,
WalSndPrepareWrite, WalSndWriteData,
WalSndUpdateProgress);
@@ -1083,10 +1083,13 @@ StartLogicalReplication(StartReplicationCmd *cmd)
* position.
*
* Do this before sending CopyBoth, so that any errors are reported early.
+ *+ * logical_read_xlog_page() eventually calls XLogRead(), so pass the+ * initial position.
*/
logical_decoding_ctx =
CreateDecodingContext(cmd->startpoint, cmd->options, false,
- logical_read_xlog_page,+ logical_read_xlog_page, sendPos,
WalSndPrepareWrite, WalSndWriteData,
WalSndUpdateProgress);
@@ -2344,187 +2347,76 @@ WalSndKill(int code, Datum arg)
}
/*
- * Read 'count' bytes from WAL into 'buf', starting at location 'startptr'- *- * XXX probably this should be improved to suck data directly from the- * WAL buffers when possible.- *- * Will open, and keep open, one WAL segment stored in the global file- * descriptor sendFile. This means if XLogRead is used once, there will- * always be one descriptor left open until the process ends, but never- * more than one.+ * Callback for XLogRead() to open the next segment.
*/
-static void-XLogRead(char *buf, XLogRecPtr startptr, Size count)+void+WalSndOpenSegment(XLogSegNo segNo, TimeLineID *tli, XLogReadPos *pos)
{
- char *p;- XLogRecPtr recptr;- Size nbytes;- XLogSegNo segno;+ char path[MAXPGPATH];-retry:- p = buf;- recptr = startptr;- nbytes = count;+ /*+ * The timeline is determined below, caller should not do anything about+ * it.+ */+ Assert(tli == NULL);- while (nbytes > 0)+ /*-------+ * 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. -------+ */+ pos->tli = sendTimeLine;+ if (sendTimeLineIsHistoric)
{
- uint32 startoff;- int segbytes;- int readbytes;-- startoff = XLogSegmentOffset(recptr, wal_segment_size);-- if (sendFile < 0 || !XLByteInSeg(recptr, sendSegNo, wal_segment_size))- {- char path[MAXPGPATH];-- /* Switch to another logfile segment */- if (sendFile >= 0)- close(sendFile);+ XLogSegNo endSegNo;- XLByteToSeg(recptr, sendSegNo, wal_segment_size);-- /*-------- * 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.- *-------- */- curFileTimeLine = sendTimeLine;- if (sendTimeLineIsHistoric)- {- XLogSegNo endSegNo;-- XLByteToSeg(sendTimeLineValidUpto, endSegNo, wal_segment_size);- if (sendSegNo == endSegNo)- curFileTimeLine = sendTimeLineNextTLI;- }-- XLogFilePath(path, curFileTimeLine, sendSegNo, wal_segment_size);-- sendFile = BasicOpenFile(path, O_RDONLY | PG_BINARY);- if (sendFile < 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(curFileTimeLine, sendSegNo))));- else- ereport(ERROR,- (errcode_for_file_access(),- errmsg("could not open file \"%s\": %m",- path)));- }- sendOff = 0;- }-- /* Need to seek in the file? */- if (sendOff != startoff)- {- if (lseek(sendFile, (off_t) startoff, SEEK_SET) < 0)- ereport(ERROR,- (errcode_for_file_access(),- errmsg("could not seek in log segment %s to offset %u: %m",- XLogFileNameP(curFileTimeLine, sendSegNo),- startoff)));- sendOff = startoff;- }+ XLByteToSeg(sendTimeLineValidUpto, endSegNo, wal_segment_size);+ if (pos->segNo == endSegNo)+ pos->tli = sendTimeLineNextTLI;+ }- /* How many bytes are within this segment? */- if (nbytes > (wal_segment_size - startoff))- segbytes = wal_segment_size - startoff;- else- segbytes = nbytes;+ XLogFilePath(path, pos->tli, segNo, wal_segment_size);+ pos->segFile = BasicOpenFile(path, O_RDONLY | PG_BINARY);- pgstat_report_wait_start(WAIT_EVENT_WAL_READ);- readbytes = read(sendFile, p, segbytes);- pgstat_report_wait_end();- if (readbytes < 0)- {+ if (pos->segFile < 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("could not read from log segment %s, offset %u, length %zu: %m",- XLogFileNameP(curFileTimeLine, sendSegNo),- sendOff, (Size) segbytes)));- }- else if (readbytes == 0)- {+ errmsg("requested WAL segment %s has already been removed",+ XLogFileNameP(pos->tli, pos->segNo))));+ else
ereport(ERROR,
- (errcode(ERRCODE_DATA_CORRUPTED),- errmsg("could not read from log segment %s, offset %u: read %d of %zu",- XLogFileNameP(curFileTimeLine, sendSegNo),- sendOff, readbytes, (Size) segbytes)));- }-- /* Update state for read */- recptr += readbytes;-- sendOff += readbytes;- nbytes -= readbytes;- p += readbytes;+ (errcode_for_file_access(),+ errmsg("could not open file \"%s\": %m",+ path)));
}
- /*- * 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(startptr, segno, wal_segment_size);- CheckXLogRemoved(segno, ThisTimeLineID);-- /*- * 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 && sendFile >= 0)- {- close(sendFile);- sendFile = -1;-- goto retry;- }- }+ pos->segNo = segNo;+ pos->segOff = 0;
}
/*
@@ -2544,6 +2436,7 @@ XLogSendPhysical(void)
XLogRecPtr startptr;
XLogRecPtr endptr;
Size nbytes;
+ XLogSegNo segno;
/* If requested switch the WAL sender to the stopping state. */
if (got_STOPPING)
@@ -2686,9 +2579,9 @@ XLogSendPhysical(void)
if (sendTimeLineIsHistoric && sendTimeLineValidUpto <= sentPtr)
{
/* close the current file. */
- if (sendFile >= 0)- close(sendFile);- sendFile = -1;+ if (sendPos->segFile >= 0)+ close(sendPos->segFile);+ sendPos->segFile = -1;
/* Send CopyDone */
pq_putmessage_noblock('c', NULL, 0);
@@ -2759,7 +2652,48 @@ XLogSendPhysical(void)
* calls.
*/
enlargeStringInfo(&output_message, nbytes);
- XLogRead(&output_message.data[output_message.len], startptr, nbytes);++retry:+ XLogRead(&output_message.data[output_message.len], startptr, nbytes,+ NULL, /* WalSndOpenSegment will determine TLI */+ sendPos,+ WalSndOpenSegment);++ /*+ * 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(startptr, segno, wal_segment_size);+ CheckXLogRemoved(segno, ThisTimeLineID);++ /*+ * 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 && sendPos->segFile >= 0)+ {+ close(sendPos->segFile);+ sendPos->segFile = -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.cindex e106fb2ed1..7dd63dd735 100644--- a/src/bin/pg_waldump/pg_waldump.c+++ b/src/bin/pg_waldump/pg_waldump.c@@ -26,9 +26,9 @@
#include "rmgrdesc.h"
-static const char *progname;+const char *progname;-static int WalSegSz;+int WalSegSz;
typedef struct XLogDumpPrivate
{
@@ -37,6 +37,7 @@ typedef struct XLogDumpPrivate
XLogRecPtr startptr;
XLogRecPtr endptr;
bool endptr_reached;
+ XLogReadPos *pos;
} XLogDumpPrivate;
typedef struct XLogDumpConfig
@@ -296,126 +297,45 @@ identify_target_directory(XLogDumpPrivate *private, char *directory,
fatal_error("could not find any WAL file");
}
-/*- * Read count bytes from a segment file in the specified directory, for the- * given timeline, containing the specified record pointer; store the data in- * the passed buffer.- */
static void
-XLogDumpXLogRead(const char *directory, TimeLineID timeline_id,- XLogRecPtr startptr, char *buf, Size count)+XLogDumpOpenSegment(XLogSegNo segNo, TimeLineID *tli, XLogReadPos *pos)
{
- char *p;- XLogRecPtr recptr;- Size nbytes;-- static int sendFile = -1;- static XLogSegNo sendSegNo = 0;- static uint32 sendOff = 0;+ char fname[MAXPGPATH];+ int tries;- p = buf;- recptr = startptr;- nbytes = count;+ XLogFileName(fname, *tli, segNo, WalSegSz);- while (nbytes > 0)+ /*+ * 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++)
{
- uint32 startoff;- int segbytes;- int readbytes;-- startoff = XLogSegmentOffset(recptr, WalSegSz);-- if (sendFile < 0 || !XLByteInSeg(recptr, sendSegNo, WalSegSz))- {- char fname[MAXFNAMELEN];- int tries;-- /* Switch to another logfile segment */- if (sendFile >= 0)- close(sendFile);-- XLByteToSeg(recptr, sendSegNo, WalSegSz);-- XLogFileName(fname, timeline_id, sendSegNo, WalSegSz);-- /*- * 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++)- {- sendFile = open_file_in_directory(directory, fname);- if (sendFile >= 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 (sendFile < 0)- fatal_error("could not find file \"%s\": %s",- fname, strerror(errno));- sendOff = 0;- }-- /* Need to seek in the file? */- if (sendOff != startoff)- {- if (lseek(sendFile, (off_t) startoff, SEEK_SET) < 0)- {- int err = errno;- char fname[MAXPGPATH];-- XLogFileName(fname, timeline_id, sendSegNo, WalSegSz);-- fatal_error("could not seek in log file %s to offset %u: %s",- fname, startoff, strerror(err));- }- sendOff = startoff;- }-- /* How many bytes are within this segment? */- if (nbytes > (WalSegSz - startoff))- segbytes = WalSegSz - startoff;- else- segbytes = nbytes;-- readbytes = read(sendFile, p, segbytes);- if (readbytes <= 0)+ pos->segFile = open_file_in_directory(pos->dir, fname);+ if (pos->segFile >= 0)+ break;+ if (errno == ENOENT)
{
- int err = errno;- char fname[MAXPGPATH];
int save_errno = errno;
- XLogFileName(fname, timeline_id, sendSegNo, WalSegSz);- errno = save_errno;+ /* File not there yet, try again */+ pg_usleep(500 * 1000);- if (readbytes < 0)- fatal_error("could not read from log file %s, offset %u, length %d: %s",- fname, sendOff, segbytes, strerror(err));- else if (readbytes == 0)- fatal_error("could not read from log file %s, offset %u: read %d of %zu",- fname, sendOff, readbytes, (Size) segbytes);+ errno = save_errno;+ continue;
}
+ /* Any other error, fall through and fail */+ break;+ }- /* Update state for read */- recptr += readbytes;+ if (pos->segFile < 0)+ fatal_error("could not find file \"%s\": %s",+ fname, strerror(errno));- sendOff += readbytes;- nbytes -= readbytes;- p += readbytes;- }+ pos->segNo = segNo;+ pos->segOff = 0;
}
/*
@@ -441,8 +361,8 @@ XLogDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
}
}
- XLogDumpXLogRead(private->inpath, private->timeline, targetPagePtr,- readBuff, count);+ XLogRead(readBuff, targetPagePtr, count, &private->timeline,+ private->pos, XLogDumpOpenSegment);
return count;
}
@@ -852,6 +772,7 @@ main(int argc, char **argv)
private.startptr = InvalidXLogRecPtr;
private.endptr = InvalidXLogRecPtr;
private.endptr_reached = false;
+ private.pos = XLogReadInitPos();
config.bkp_details = false;
config.stop_after_records = -1;
@@ -1083,6 +1004,9 @@ main(int argc, char **argv)
else
identify_target_directory(&private, private.inpath, NULL);
+ /* The XLOG position can be used separate from "private". */+ private.pos->dir = private.inpath;+
/* we don't know what to print */
if (XLogRecPtrIsInvalid(private.startptr))
{
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.hindex f3bae0bf49..9bddbd3042 100644--- a/src/include/access/xlogreader.h+++ b/src/include/access/xlogreader.h@@ -219,6 +219,31 @@ extern void XLogReaderInvalReadState(XLogReaderState *state);
extern XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr);
#endif /* FRONTEND */
+/*+ * Position in XLOG file while reading it.+ */+typedef struct XLogReadPos+{+ int segFile; /* segment file descriptor */+ XLogSegNo segNo; /* segment number */+ uint32 segOff; /* offset in the segment */+ TimeLineID tli; /* timeline ID of the currently open file */++ char *dir; /* directory (only needed by frontends) */+} XLogReadPos;++/*+ * Callback to open the specified XLOG segment 'segNo' in timeline 'tli' for+ * reading and update the position accordingly.+ */+typedef void (*XLogOpenSegment) (XLogSegNo segNo, TimeLineID *tli,+ XLogReadPos *pos);++extern XLogReadPos *XLogReadInitPos(void);+extern void XLogRead(char *buf, XLogRecPtr startptr, Size count,+ TimeLineID *tli, XLogReadPos *pos,+ XLogOpenSegment openSegment);+
/* Functions for decoding an XLogRecord */
extern bool DecodeXLogRecord(XLogReaderState *state, XLogRecord *record,
diff --git a/src/include/replication/logical.h b/src/include/replication/logical.hindex 0a2a63a48c..59b29433eb 100644--- a/src/include/replication/logical.h+++ b/src/include/replication/logical.h@@ -99,6 +99,7 @@ extern LogicalDecodingContext *CreateInitDecodingContext(char *plugin,
bool need_full_snapshot,
XLogRecPtr restart_lsn,
XLogPageReadCB read_page,
+ void *read_page_arg,
LogicalOutputPluginWriterPrepareWrite prepare_write,
LogicalOutputPluginWriterWrite do_write,
LogicalOutputPluginWriterUpdateProgress update_progress);
@@ -107,6 +108,7 @@ extern LogicalDecodingContext *CreateDecodingContext(
List *output_plugin_options,
bool fast_forward,
XLogPageReadCB read_page,
+ void *read_page_arg,
LogicalOutputPluginWriterPrepareWrite prepare_write,
LogicalOutputPluginWriterWrite do_write,
LogicalOutputPluginWriterUpdateProgress update_progress);
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: pgsql-hackers@postgresql.org
Cc: ah@cybertec.at
Subject: Re: Attempt to consolidate reading of XLOG page
In-Reply-To: <14984.1554998742@spoje.net>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by DDX for PostgreSQL; see mirroring instructions
for how to clone and mirror all data and code used for this inbox