public inbox for [email protected]
help / color / mirror / Atom feed[PATCH v17 1/4] Move callback-call from ReadPageInternal to XLogReadRecord.
2+ messages / 2 participants
[nested] [flat]
* [PATCH v17 1/4] Move callback-call from ReadPageInternal to XLogReadRecord.
@ 2019-09-05 11:21 Kyotaro Horiguchi <[email protected]>
0 siblings, 0 replies; 2+ messages in thread
From: Kyotaro Horiguchi @ 2019-09-05 11:21 UTC (permalink / raw)
The current WAL record reader reads page data using a call back
function. Although it is not so problematic alone, it would be a
problem if we are going to do add tasks like encryption which is
performed on page data before WAL reader reads them. To avoid that the
record reader facility has to have a new code path corresponds to
every new callback, this patch separates page reader from WAL record
reading facility by modifying the current WAL record reader to a state
machine.
As the first step of that change, this patch moves the page reader
function out of ReadPageInternal, then the remaining tasks of the
function are taken over by the new function XLogNeedData. As the
result XLogPageRead directly calls the page reader callback function
according to the feedback from XLogNeedData.
---
src/backend/access/transam/xlog.c | 16 +-
src/backend/access/transam/xlogreader.c | 281 ++++++++++++++----------
src/backend/access/transam/xlogutils.c | 12 +-
src/backend/replication/walsender.c | 10 +-
src/bin/pg_rewind/parsexlog.c | 21 +-
src/bin/pg_waldump/pg_waldump.c | 8 +-
src/include/access/xlogreader.h | 25 ++-
src/include/access/xlogutils.h | 2 +-
8 files changed, 222 insertions(+), 153 deletions(-)
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 377afb8732..606328a65a 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -917,7 +917,7 @@ static bool InstallXLogFileSegment(XLogSegNo *segno, char *tmppath,
static int XLogFileRead(XLogSegNo segno, int emode, TimeLineID tli,
XLogSource source, bool notfoundOk);
static int XLogFileReadAnyTLI(XLogSegNo segno, int emode, XLogSource source);
-static int XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
+static bool XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
int reqLen, XLogRecPtr targetRecPtr, char *readBuf);
static bool WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
bool fetching_ckpt, XLogRecPtr tliRecPtr);
@@ -4350,7 +4350,6 @@ ReadRecord(XLogReaderState *xlogreader, int emode,
XLogRecord *record;
XLogPageReadPrivate *private = (XLogPageReadPrivate *) xlogreader->private_data;
- /* Pass through parameters to XLogPageRead */
private->fetching_ckpt = fetching_ckpt;
private->emode = emode;
private->randAccess = (xlogreader->ReadRecPtr == InvalidXLogRecPtr);
@@ -11987,7 +11986,7 @@ CancelBackup(void)
* XLogPageRead() to try fetching the record from another source, or to
* sleep and retry.
*/
-static int
+static bool
XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, int reqLen,
XLogRecPtr targetRecPtr, char *readBuf)
{
@@ -12046,7 +12045,8 @@ retry:
readLen = 0;
readSource = XLOG_FROM_ANY;
- return -1;
+ xlogreader->readLen = -1;
+ return false;
}
}
@@ -12141,7 +12141,8 @@ retry:
goto next_record_is_invalid;
}
- return readLen;
+ xlogreader->readLen = readLen;
+ return true;
next_record_is_invalid:
lastSourceFailed = true;
@@ -12155,8 +12156,9 @@ next_record_is_invalid:
/* In standby-mode, keep trying */
if (StandbyMode)
goto retry;
- else
- return -1;
+
+ xlogreader->readLen = -1;
+ return false;
}
/*
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index 42738eb940..7d04995b7b 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -36,8 +36,8 @@
static void report_invalid_record(XLogReaderState *state, const char *fmt,...)
pg_attribute_printf(2, 3);
static bool allocate_recordbuf(XLogReaderState *state, uint32 reclength);
-static int ReadPageInternal(XLogReaderState *state, XLogRecPtr pageptr,
- int reqLen);
+static bool XLogNeedData(XLogReaderState *state, XLogRecPtr pageptr,
+ int reqLen, bool header_inclusive);
static void XLogReaderInvalReadState(XLogReaderState *state);
static bool ValidXLogRecordHeader(XLogReaderState *state, XLogRecPtr RecPtr,
XLogRecPtr PrevRecPtr, XLogRecord *record, bool randAccess);
@@ -276,7 +276,6 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
uint32 targetRecOff;
uint32 pageHeaderSize;
bool gotheader;
- int readOff;
/*
* randAccess indicates whether to verify the previous-record pointer of
@@ -326,14 +325,20 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
* byte to cover the whole record header, or at least the part of it that
* fits on the same page.
*/
- readOff = ReadPageInternal(state, targetPagePtr,
- Min(targetRecOff + SizeOfXLogRecord, XLOG_BLCKSZ));
- if (readOff < 0)
+ while (XLogNeedData(state, targetPagePtr,
+ Min(targetRecOff + SizeOfXLogRecord, XLOG_BLCKSZ),
+ targetRecOff != 0))
+ {
+ if (!state->routine.page_read(state, state->readPagePtr, state->readLen,
+ RecPtr, state->readBuf))
+ break;
+ }
+
+ if (!state->page_verified)
goto err;
/*
- * ReadPageInternal always returns at least the page header, so we can
- * examine it now.
+ * We have at least the page header, so we can examine it now.
*/
pageHeaderSize = XLogPageHeaderSize((XLogPageHeader) state->readBuf);
if (targetRecOff == 0)
@@ -359,8 +364,8 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
goto err;
}
- /* ReadPageInternal has verified the page header */
- Assert(pageHeaderSize <= readOff);
+ /* XLogNeedData has verified the page header */
+ Assert(pageHeaderSize <= state->readLen);
/*
* Read the record length.
@@ -432,18 +437,27 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
do
{
+ int rest_len = total_len - gotlen;
+
/* Calculate pointer to beginning of next page */
targetPagePtr += XLOG_BLCKSZ;
/* Wait for the next page to become available */
- readOff = ReadPageInternal(state, targetPagePtr,
- Min(total_len - gotlen + SizeOfXLogShortPHD,
- XLOG_BLCKSZ));
+ while (XLogNeedData(state, targetPagePtr,
+ Min(rest_len, XLOG_BLCKSZ),
+ false))
+ {
+ if (!state->routine.page_read(state, state->readPagePtr,
+ state->readLen,
+ state->ReadRecPtr,
+ state->readBuf))
+ break;
+ }
- if (readOff < 0)
+ if (!state->page_verified)
goto err;
- Assert(SizeOfXLogShortPHD <= readOff);
+ Assert(SizeOfXLogShortPHD <= state->readLen);
/* Check that the continuation on next page looks valid */
pageHeader = (XLogPageHeader) state->readBuf;
@@ -473,21 +487,14 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
/* Append the continuation from this page to the buffer */
pageHeaderSize = XLogPageHeaderSize(pageHeader);
- if (readOff < pageHeaderSize)
- readOff = ReadPageInternal(state, targetPagePtr,
- pageHeaderSize);
-
- Assert(pageHeaderSize <= readOff);
+ Assert(pageHeaderSize <= state->readLen);
contdata = (char *) state->readBuf + pageHeaderSize;
len = XLOG_BLCKSZ - pageHeaderSize;
if (pageHeader->xlp_rem_len < len)
len = pageHeader->xlp_rem_len;
- if (readOff < pageHeaderSize + len)
- readOff = ReadPageInternal(state, targetPagePtr,
- pageHeaderSize + len);
-
+ Assert (pageHeaderSize + len <= state->readLen);
memcpy(buffer, (char *) contdata, len);
buffer += len;
gotlen += len;
@@ -517,9 +524,16 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
else
{
/* Wait for the record data to become available */
- readOff = ReadPageInternal(state, targetPagePtr,
- Min(targetRecOff + total_len, XLOG_BLCKSZ));
- if (readOff < 0)
+ while (XLogNeedData(state, targetPagePtr,
+ Min(targetRecOff + total_len, XLOG_BLCKSZ), true))
+ {
+ if (!state->routine.page_read(state, state->readPagePtr,
+ state->readLen,
+ state->ReadRecPtr, state->readBuf))
+ break;
+ }
+
+ if (!state->page_verified)
goto err;
/* Record does not cross a page boundary */
@@ -562,109 +576,138 @@ err:
}
/*
- * Read a single xlog page including at least [pageptr, reqLen] of valid data
- * via the page_read() callback.
+ * Checks that an xlog page loaded in state->readBuf is including at least
+ * [pageptr, reqLen] and the page is valid. header_inclusive indicates that
+ * reqLen is calculated including page header length.
*
- * Returns -1 if the required page cannot be read for some reason; errormsg_buf
- * is set in that case (unless the error occurs in the page_read callback).
+ * Returns false if the buffer already contains the requested data, or found
+ * error. state->page_verified is set to true for the former and false for the
+ * latter.
*
- * We fetch the page from a reader-local cache if we know we have the required
- * data and if there hasn't been any error since caching the data.
+ * Otherwise returns true and requests data loaded onto state->readBuf by
+ * state->readPagePtr and state->readLen. The caller shall call this function
+ * again after filling the buffer at least with that portion of data and set
+ * state->readLen to the length of actually loaded data.
+ *
+ * If header_inclusive is false, corrects reqLen internally by adding the
+ * actual page header length and may request caller for new data.
*/
-static int
-ReadPageInternal(XLogReaderState *state, XLogRecPtr pageptr, int reqLen)
+static bool
+XLogNeedData(XLogReaderState *state, XLogRecPtr pageptr, int reqLen,
+ bool header_inclusive)
{
- int readLen;
uint32 targetPageOff;
XLogSegNo targetSegNo;
- XLogPageHeader hdr;
+ uint32 addLen = 0;
- Assert((pageptr % XLOG_BLCKSZ) == 0);
+ /* Some data is loaded, but page header is not verified yet. */
+ if (!state->page_verified &&
+ !XLogRecPtrIsInvalid(state->readPagePtr) && state->readLen >= 0)
+ {
+ uint32 pageHeaderSize;
+ /* just loaded new data so needs to verify page header */
+
+ /* The caller must have loaded at least page header */
+ Assert (state->readLen >= SizeOfXLogShortPHD);
+
+ /*
+ * We have enough data to check the header length. Recheck the loaded
+ * length against the actual header length.
+ */
+ pageHeaderSize = XLogPageHeaderSize((XLogPageHeader) state->readBuf);
+
+ /* Request more data if we don't have the full header. */
+ if (state->readLen < pageHeaderSize)
+ {
+ state->readLen = pageHeaderSize;
+ return true;
+ }
+
+ /* Now that we know we have the full header, validate it. */
+ if (!XLogReaderValidatePageHeader(state, state->readPagePtr,
+ (char *) state->readBuf))
+ {
+ /* That's bad. Force reading the page again. */
+ XLogReaderInvalReadState(state);
+
+ return false;
+ }
+
+ state->page_verified = true;
+
+ XLByteToSeg(state->readPagePtr, state->seg.ws_segno,
+ state->segcxt.ws_segsize);
+ }
+
+ /*
+ * The loaded page may not be the one caller is supposing to read when we
+ * are verifying the first page of new segment. In that case, skip further
+ * verification and immediately load the target page.
+ */
+ if (state->page_verified && pageptr == state->readPagePtr)
+ {
+ /*
+ * calculate additional length for page header keeping the total
+ * length within the block size.
+ */
+ if (!header_inclusive)
+ {
+ uint32 pageHeaderSize =
+ XLogPageHeaderSize((XLogPageHeader) state->readBuf);
+
+ addLen = pageHeaderSize;
+ if (reqLen + pageHeaderSize <= XLOG_BLCKSZ)
+ addLen = pageHeaderSize;
+ else
+ addLen = XLOG_BLCKSZ - reqLen;
+
+ Assert(addLen >= 0);
+ }
+
+ /* Return if we already have it. */
+ if (reqLen + addLen <= state->readLen)
+ return false;
+ }
+
+ /* Data is not in our buffer, request the caller for it. */
XLByteToSeg(pageptr, targetSegNo, state->segcxt.ws_segsize);
targetPageOff = XLogSegmentOffset(pageptr, state->segcxt.ws_segsize);
+ Assert((pageptr % XLOG_BLCKSZ) == 0);
- /* check whether we have all the requested data already */
- if (targetSegNo == state->seg.ws_segno &&
- targetPageOff == state->segoff && reqLen <= state->readLen)
- return state->readLen;
+ /*
+ * Every time we request to load new data of a page to the caller, even if
+ * we looked at a part of it before, we need to do verification on the next
+ * invocation as the caller might now be rereading data from a different
+ * source.
+ */
+ state->page_verified = false;
/*
- * Data is not in our buffer.
- *
- * Every time we actually read the segment, even if we looked at parts of
- * it before, we need to do verification as the page_read callback might
- * now be rereading data from a different source.
- *
* Whenever switching to a new WAL segment, we read the first page of the
* file and validate its header, even if that's not where the target
* record is. This is so that we can check the additional identification
* info that is present in the first page's "long" header.
+ * Don't do this if the caller requested the first page in the segment.
*/
if (targetSegNo != state->seg.ws_segno && targetPageOff != 0)
{
- XLogRecPtr targetSegmentPtr = pageptr - targetPageOff;
-
- readLen = state->routine.page_read(state, targetSegmentPtr, XLOG_BLCKSZ,
- state->currRecPtr,
- state->readBuf);
- if (readLen < 0)
- goto err;
-
- /* we can be sure to have enough WAL available, we scrolled back */
- Assert(readLen == XLOG_BLCKSZ);
-
- if (!XLogReaderValidatePageHeader(state, targetSegmentPtr,
- state->readBuf))
- goto err;
+ /*
+ * Then we'll see that the targetSegNo now matches the ws_segno, and
+ * will not come back here, but will request the actual target page.
+ */
+ state->readPagePtr = pageptr - targetPageOff;
+ state->readLen = XLOG_BLCKSZ;
+ return true;
}
/*
- * First, read the requested data length, but at least a short page header
- * so that we can validate it.
+ * Request the caller to load the page. We need at least a short page
+ * header so that we can validate it.
*/
- readLen = state->routine.page_read(state, pageptr, Max(reqLen, SizeOfXLogShortPHD),
- state->currRecPtr,
- state->readBuf);
- if (readLen < 0)
- goto err;
-
- Assert(readLen <= XLOG_BLCKSZ);
-
- /* Do we have enough data to check the header length? */
- if (readLen <= SizeOfXLogShortPHD)
- goto err;
-
- Assert(readLen >= reqLen);
-
- hdr = (XLogPageHeader) state->readBuf;
-
- /* still not enough */
- if (readLen < XLogPageHeaderSize(hdr))
- {
- readLen = state->routine.page_read(state, pageptr, XLogPageHeaderSize(hdr),
- state->currRecPtr,
- state->readBuf);
- if (readLen < 0)
- goto err;
- }
-
- /*
- * Now that we know we have the full header, validate it.
- */
- if (!XLogReaderValidatePageHeader(state, pageptr, (char *) hdr))
- goto err;
-
- /* update read state information */
- state->seg.ws_segno = targetSegNo;
- state->segoff = targetPageOff;
- state->readLen = readLen;
-
- return readLen;
-
-err:
- XLogReaderInvalReadState(state);
- return -1;
+ state->readPagePtr = pageptr;
+ state->readLen = Max(reqLen + addLen, SizeOfXLogShortPHD);
+ return true;
}
/*
@@ -673,9 +716,7 @@ err:
static void
XLogReaderInvalReadState(XLogReaderState *state)
{
- state->seg.ws_segno = 0;
- state->segoff = 0;
- state->readLen = 0;
+ state->readPagePtr = InvalidXLogRecPtr;
}
/*
@@ -953,7 +994,6 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
XLogRecPtr targetPagePtr;
int targetRecOff;
uint32 pageHeaderSize;
- int readLen;
/*
* Compute targetRecOff. It should typically be equal or greater than
@@ -961,7 +1001,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
* that, except when caller has explicitly specified the offset that
* falls somewhere there or when we are skipping multi-page
* continuation record. It doesn't matter though because
- * ReadPageInternal() is prepared to handle that and will read at
+ * XLogNeedData() is prepared to handle that and will read at
* least short page-header worth of data
*/
targetRecOff = tmpRecPtr % XLOG_BLCKSZ;
@@ -969,19 +1009,24 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
/* scroll back to page boundary */
targetPagePtr = tmpRecPtr - targetRecOff;
- /* Read the page containing the record */
- readLen = ReadPageInternal(state, targetPagePtr, targetRecOff);
- if (readLen < 0)
+ while(XLogNeedData(state, targetPagePtr, targetRecOff,
+ targetRecOff != 0))
+ {
+ if (!state->routine.page_read(state, state->readPagePtr,
+ state->readLen,
+ state->ReadRecPtr, state->readBuf))
+ break;
+ }
+
+ if (!state->page_verified)
goto err;
header = (XLogPageHeader) state->readBuf;
pageHeaderSize = XLogPageHeaderSize(header);
- /* make sure we have enough data for the page header */
- readLen = ReadPageInternal(state, targetPagePtr, pageHeaderSize);
- if (readLen < 0)
- goto err;
+ /* we should have read the page header */
+ Assert (state->readLen >= pageHeaderSize);
/* skip over potential continuation data */
if (header->xlp_info & XLP_FIRST_IS_CONTRECORD)
diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c
index d17d660f46..46eda33f25 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -686,8 +686,8 @@ XLogTruncateRelation(RelFileNode rnode, ForkNumber forkNum,
void
XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage, uint32 wantLength)
{
- const XLogRecPtr lastReadPage = (state->seg.ws_segno *
- state->segcxt.ws_segsize + state->segoff);
+ const XLogRecPtr lastReadPage = state->seg.ws_segno *
+ state->segcxt.ws_segsize + state->readLen;
Assert(wantPage != InvalidXLogRecPtr && wantPage % XLOG_BLCKSZ == 0);
Assert(wantLength <= XLOG_BLCKSZ);
@@ -824,7 +824,7 @@ wal_segment_close(XLogReaderState *state)
* exists for normal backends, so we have to do a check/sleep/repeat style of
* loop for now.
*/
-int
+bool
read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
int reqLen, XLogRecPtr targetRecPtr, char *cur_page)
{
@@ -926,7 +926,8 @@ read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
else if (targetPagePtr + reqLen > read_upto)
{
/* not enough data there */
- return -1;
+ state->readLen = -1;
+ return false;
}
else
{
@@ -944,7 +945,8 @@ read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
WALReadRaiseError(&errinfo);
/* number of valid bytes in the buffer */
- return count;
+ state->readLen = count;
+ return true;
}
/*
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 23baa4498a..c36b91ae5e 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -806,7 +806,7 @@ StartReplication(StartReplicationCmd *cmd)
* which has to do a plain sleep/busy loop, because the walsender's latch gets
* set every time WAL is flushed.
*/
-static int
+static bool
logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
XLogRecPtr targetRecPtr, char *cur_page)
{
@@ -826,7 +826,10 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req
/* fail if not (implies we are going to shut down) */
if (flushptr < targetPagePtr + reqLen)
- return -1;
+ {
+ state->readLen = -1;
+ return false;
+ }
if (targetPagePtr + XLOG_BLCKSZ <= flushptr)
count = XLOG_BLCKSZ; /* more than one block available */
@@ -854,7 +857,8 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req
XLByteToSeg(targetPagePtr, segno, state->segcxt.ws_segsize);
CheckXLogRemoved(segno, state->seg.ws_tli);
- return count;
+ state->readLen = count;
+ return true;
}
/*
diff --git a/src/bin/pg_rewind/parsexlog.c b/src/bin/pg_rewind/parsexlog.c
index 59ebac7d6a..cf119848b0 100644
--- a/src/bin/pg_rewind/parsexlog.c
+++ b/src/bin/pg_rewind/parsexlog.c
@@ -47,7 +47,7 @@ typedef struct XLogPageReadPrivate
int tliIndex;
} XLogPageReadPrivate;
-static int SimpleXLogPageRead(XLogReaderState *xlogreader,
+static bool SimpleXLogPageRead(XLogReaderState *xlogreader,
XLogRecPtr targetPagePtr,
int reqLen, XLogRecPtr targetRecPtr, char *readBuf);
@@ -246,7 +246,7 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex,
}
/* XLogReader callback function, to read a WAL page */
-static int
+static bool
SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
int reqLen, XLogRecPtr targetRecPtr, char *readBuf)
{
@@ -306,7 +306,8 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
if (private->restoreCommand == NULL)
{
pg_log_error("could not open file \"%s\": %m", xlogfpath);
- return -1;
+ xlogreader->readLen = -1;
+ return false;
}
/*
@@ -319,7 +320,10 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
private->restoreCommand);
if (xlogreadfd < 0)
- return -1;
+ {
+ xlogreader->readLen = -1;
+ return false;
+ }
else
pg_log_debug("using file \"%s\" restored from archive",
xlogfpath);
@@ -335,7 +339,8 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
if (lseek(xlogreadfd, (off_t) targetPageOff, SEEK_SET) < 0)
{
pg_log_error("could not seek in file \"%s\": %m", xlogfpath);
- return -1;
+ xlogreader->readLen = -1;
+ return false;
}
@@ -348,13 +353,15 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
pg_log_error("could not read file \"%s\": read %d of %zu",
xlogfpath, r, (Size) XLOG_BLCKSZ);
- return -1;
+ xlogreader->readLen = -1;
+ return false;
}
Assert(targetSegNo == xlogreadsegno);
xlogreader->seg.ws_tli = targetHistory[private->tliIndex].tli;
- return XLOG_BLCKSZ;
+ xlogreader->readLen = XLOG_BLCKSZ;
+ return true;
}
/*
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 610f65e471..0f0e0723a3 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -331,7 +331,7 @@ WALDumpCloseSegment(XLogReaderState *state)
}
/* pg_waldump's XLogReaderRoutine->page_read callback */
-static int
+static bool
WALDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
XLogRecPtr targetPtr, char *readBuff)
{
@@ -348,7 +348,8 @@ WALDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
else
{
private->endptr_reached = true;
- return -1;
+ state->readLen = -1;
+ return false;
}
}
@@ -373,7 +374,8 @@ WALDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
(Size) errinfo.wre_req);
}
- return count;
+ state->readLen = count;
+ return true;
}
/*
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
index 21d200d3df..23d9e70a59 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -57,8 +57,8 @@ typedef struct WALSegmentContext
typedef struct XLogReaderState XLogReaderState;
-/* Function type definitions for various xlogreader interactions */
-typedef int (*XLogPageReadCB) (XLogReaderState *xlogreader,
+/* Function type definition for the read_page callback */
+typedef bool (*XLogPageReadCB) (XLogReaderState *xlogreader,
XLogRecPtr targetPagePtr,
int reqLen,
XLogRecPtr targetRecPtr,
@@ -175,6 +175,20 @@ struct XLogReaderState
XLogRecPtr ReadRecPtr; /* start of last record read */
XLogRecPtr EndRecPtr; /* end+1 of last record read */
+ /* ----------------------------------------
+ * Communication with page reader
+ * readBuf is XLOG_BLCKSZ bytes, valid up to at least readLen bytes.
+ * ----------------------------------------
+ */
+ /* variables to communicate with page reader */
+ XLogRecPtr readPagePtr; /* page pointer to read */
+ int32 readLen; /* bytes requested to reader, or actual bytes
+ * read by reader, which must be larger than
+ * the request, or -1 on error */
+ TimeLineID readPageTLI; /* TLI for data currently in readBuf */
+ char *readBuf; /* buffer to store data */
+ bool page_verified; /* is the page on the buffer verified? */
+
/* ----------------------------------------
* Decoded representation of current record
@@ -203,13 +217,6 @@ struct XLogReaderState
* ----------------------------------------
*/
- /*
- * Buffer for currently read page (XLOG_BLCKSZ bytes, valid up to at least
- * readLen bytes)
- */
- char *readBuf;
- uint32 readLen;
-
/* last read XLOG position for data currently in readBuf */
WALSegmentContext segcxt;
WALOpenSegment seg;
diff --git a/src/include/access/xlogutils.h b/src/include/access/xlogutils.h
index 9ac602b674..364a21c4ea 100644
--- a/src/include/access/xlogutils.h
+++ b/src/include/access/xlogutils.h
@@ -47,7 +47,7 @@ extern Buffer XLogReadBufferExtended(RelFileNode rnode, ForkNumber forknum,
extern Relation CreateFakeRelcacheEntry(RelFileNode rnode);
extern void FreeFakeRelcacheEntry(Relation fakerel);
-extern int read_local_xlog_page(XLogReaderState *state,
+extern bool read_local_xlog_page(XLogReaderState *state,
XLogRecPtr targetPagePtr, int reqLen,
XLogRecPtr targetRecPtr, char *cur_page);
extern void wal_segment_open(XLogReaderState *state,
--
2.27.0
----Next_Part(Thu_Mar__4_11_28_53_2021_014)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
filename="v17-0002-Move-page-reader-out-of-XLogReadRecord.patch"
^ permalink raw reply [nested|flat] 2+ messages in thread
* pl/pgperl Patch for adding $_FN detail just like triggers have for $_TD
@ 2024-08-28 21:53 Mark Murawski <[email protected]>
0 siblings, 0 replies; 2+ messages in thread
From: Mark Murawski @ 2024-08-28 21:53 UTC (permalink / raw)
To: [email protected]
Hi Hackers!
This would be version v1 of this feature
Basically, the subject says it all: pl/pgperl Patch for being able to
tell which function you're in.
This is a hashref so it will be possible to populate new and exciting
other details in the future as the need arises
This also greatly improves logging capabilities for things like catching
warnings, Because as it stands right now, there's no information that
can assist with locating the source of a warning like this:
# tail -f /var/log/postgresql.log
******* GOT A WARNING - Use of uninitialized value $prefix in
concatenation (.) or string at (eval 531) line 48.
Now, with $_FN you can do this:
CREATE OR REPLACE FUNCTION throw_warning() RETURNS text LANGUAGE plperlu
AS $function$
use warnings;
use strict;
use Data::Dumper;
$SIG{__WARN__} = sub {
elog(NOTICE, Dumper($_FN));
print STDERR "In Function: $_FN->{name}: $_[0]\n";
};
my $a;
print "$a"; # uninit!
return undef;
$function$
;
This patch is against 12 which is still our production branch. This
could easily be also patched against newer releases as well.
I've been using this code in production now for about 3 years, it's
greatly helped track down issues. And there shouldn't be anything
platform-specific here, it's all regular perl API
I'm not sure about adding testing. This is my first postgres patch, so
any guidance on adding regression testing would be appreciated.
The rationale for this has come from the need to know the source
function name, and we've typically resorted to things like this in the past:
CREATE OR REPLACE FUNCTION throw_warning() RETURNS text LANGUAGE plperlu
AS $function$
my $function_name = 'throw_warning';
$SIG{__WARN__} = sub { print STDERR "In Function: $function_name:
$_[0]\n"; }
$function$
;
We've literally had to copy/paste this all over and it's something that
postgres should just 'give you' since it knows the name already, just
like when triggers pass you $_TD with all the pertinent information
A wishlist item would be for postgres plperl to automatically prepend
the function name and schema when throwing perl warnings so you don't
have to do your own __WARN__ handler, but this is the next best thing.
Attachments:
[text/x-patch] plperl-add-FN-v1.patch (6.6K, ../../[email protected]/2-plperl-add-FN-v1.patch)
download | inline diff:
diff --git a/src/pl/plperl/plperl.c b/src/pl/plperl/plperl.c
index f400050f38d..24b8f98cf1e 100644
--- a/src/pl/plperl/plperl.c
+++ b/src/pl/plperl/plperl.c
@@ -2116,274 +2116,298 @@ plperlu_validator(PG_FUNCTION_ARGS)
* supplied in s, and returns a reference to it
*/
static void
plperl_create_sub(plperl_proc_desc *prodesc, const char *s, Oid fn_oid)
{
dTHX;
dSP;
char subname[NAMEDATALEN + 40];
HV *pragma_hv = newHV();
SV *subref = NULL;
int count;
sprintf(subname, "%s__%u", prodesc->proname, fn_oid);
if (plperl_use_strict)
hv_store_string(pragma_hv, "strict", (SV *) newAV());
ENTER;
SAVETMPS;
PUSHMARK(SP);
EXTEND(SP, 4);
PUSHs(sv_2mortal(cstr2sv(subname)));
PUSHs(sv_2mortal(newRV_noinc((SV *) pragma_hv)));
/*
* Use 'false' for $prolog in mkfunc, which is kept for compatibility in
* case a module such as PostgreSQL::PLPerl::NYTprof replaces the function
* compiler.
*/
PUSHs(&PL_sv_no);
PUSHs(sv_2mortal(cstr2sv(s)));
PUTBACK;
/*
* G_KEEPERR seems to be needed here, else we don't recognize compile
* errors properly. Perhaps it's because there's another level of eval
* inside mksafefunc?
*/
count = call_pv("PostgreSQL::InServer::mkfunc",
G_SCALAR | G_EVAL | G_KEEPERR);
SPAGAIN;
if (count == 1)
{
SV *sub_rv = (SV *) POPs;
if (sub_rv && SvROK(sub_rv) && SvTYPE(SvRV(sub_rv)) == SVt_PVCV)
{
subref = newRV_inc(SvRV(sub_rv));
}
}
PUTBACK;
FREETMPS;
LEAVE;
if (SvTRUE(ERRSV))
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("%s", strip_trailing_ws(sv2cstr(ERRSV)))));
if (!subref)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("didn't get a CODE reference from compiling function \"%s\"",
prodesc->proname)));
prodesc->reference = subref;
return;
}
/**********************************************************************
* plperl_init_shared_libs() -
**********************************************************************/
static void
plperl_init_shared_libs(pTHX)
{
char *file = __FILE__;
newXS("DynaLoader::boot_DynaLoader", boot_DynaLoader, file);
newXS("PostgreSQL::InServer::Util::bootstrap",
boot_PostgreSQL__InServer__Util, file);
/* newXS for...::SPI::bootstrap is in select_perl_context() */
}
static SV *
plperl_call_perl_func(plperl_proc_desc *desc, FunctionCallInfo fcinfo)
{
dTHX;
dSP;
SV *retval;
int i;
int count;
Oid *argtypes = NULL;
int nargs = 0;
+ HV *hv; // hash
+ SV *FNsv; // scalar reference to the hash
+ SV *svFN; // local reference to the hash
+
ENTER;
SAVETMPS;
+ /* Give functions some metadata about what's going on in $_FN (Similar to $_TD for triggers) */
+
+ FNsv = get_sv("main::_FN", GV_ADD);
+ if (!FNsv)
+ ereport(ERROR,
+ (errcode(ERRCODE_EXTERNAL_ROUTINE_EXCEPTION),
+ errmsg("couldn't fetch $_FN")));
+
+ hv = newHV(); // create new hash
+ hv_store_string(hv, "name", cstr2sv(desc->proname));
+
+ save_item(FNsv); /* local $_FN */
+
+ sv_upgrade(FNsv, SVt_RV);
+ SvRV_set(FNsv, (SV*)hv);
+ SvROK_on(FNsv);
+
+ svFN = newRV_noinc((SV *) hv); // reference to the new hash
+ sv_setsv(FNsv, svFN);
+
PUSHMARK(SP);
EXTEND(sp, desc->nargs);
/* Get signature for true functions; inline blocks have no args. */
if (fcinfo->flinfo->fn_oid)
get_func_signature(fcinfo->flinfo->fn_oid, &argtypes, &nargs);
Assert(nargs == desc->nargs);
for (i = 0; i < desc->nargs; i++)
{
if (fcinfo->args[i].isnull)
PUSHs(&PL_sv_undef);
else if (desc->arg_is_rowtype[i])
{
SV *sv = plperl_hash_from_datum(fcinfo->args[i].value);
PUSHs(sv_2mortal(sv));
}
else
{
SV *sv;
Oid funcid;
if (OidIsValid(desc->arg_arraytype[i]))
sv = plperl_ref_from_pg_array(fcinfo->args[i].value, desc->arg_arraytype[i]);
else if ((funcid = get_transform_fromsql(argtypes[i], current_call_data->prodesc->lang_oid, current_call_data->prodesc->trftypes)))
sv = (SV *) DatumGetPointer(OidFunctionCall1(funcid, fcinfo->args[i].value));
else
{
char *tmp;
tmp = OutputFunctionCall(&(desc->arg_out_func[i]),
fcinfo->args[i].value);
sv = cstr2sv(tmp);
pfree(tmp);
}
PUSHs(sv_2mortal(sv));
}
}
PUTBACK;
/* Do NOT use G_KEEPERR here */
count = call_sv(desc->reference, G_SCALAR | G_EVAL);
SPAGAIN;
if (count != 1)
{
PUTBACK;
FREETMPS;
LEAVE;
ereport(ERROR,
(errcode(ERRCODE_EXTERNAL_ROUTINE_EXCEPTION),
errmsg("didn't get a return item from function")));
}
if (SvTRUE(ERRSV))
{
(void) POPs;
PUTBACK;
FREETMPS;
LEAVE;
/* XXX need to find a way to determine a better errcode here */
ereport(ERROR,
(errcode(ERRCODE_EXTERNAL_ROUTINE_EXCEPTION),
errmsg("%s", strip_trailing_ws(sv2cstr(ERRSV)))));
}
retval = newSVsv(POPs);
PUTBACK;
FREETMPS;
LEAVE;
return retval;
}
static SV *
plperl_call_perl_trigger_func(plperl_proc_desc *desc, FunctionCallInfo fcinfo,
SV *td)
{
dTHX;
dSP;
SV *retval,
*TDsv;
int i,
count;
Trigger *tg_trigger = ((TriggerData *) fcinfo->context)->tg_trigger;
ENTER;
SAVETMPS;
TDsv = get_sv("main::_TD", 0);
if (!TDsv)
ereport(ERROR,
(errcode(ERRCODE_EXTERNAL_ROUTINE_EXCEPTION),
errmsg("couldn't fetch $_TD")));
save_item(TDsv); /* local $_TD */
sv_setsv(TDsv, td);
PUSHMARK(sp);
EXTEND(sp, tg_trigger->tgnargs);
for (i = 0; i < tg_trigger->tgnargs; i++)
PUSHs(sv_2mortal(cstr2sv(tg_trigger->tgargs[i])));
PUTBACK;
/* Do NOT use G_KEEPERR here */
count = call_sv(desc->reference, G_SCALAR | G_EVAL);
SPAGAIN;
if (count != 1)
{
PUTBACK;
FREETMPS;
LEAVE;
ereport(ERROR,
(errcode(ERRCODE_EXTERNAL_ROUTINE_EXCEPTION),
errmsg("didn't get a return item from trigger function")));
}
if (SvTRUE(ERRSV))
{
(void) POPs;
PUTBACK;
FREETMPS;
LEAVE;
/* XXX need to find a way to determine a better errcode here */
ereport(ERROR,
(errcode(ERRCODE_EXTERNAL_ROUTINE_EXCEPTION),
errmsg("%s", strip_trailing_ws(sv2cstr(ERRSV)))));
}
retval = newSVsv(POPs);
PUTBACK;
FREETMPS;
LEAVE;
return retval;
}
static void
plperl_call_perl_event_trigger_func(plperl_proc_desc *desc,
FunctionCallInfo fcinfo,
SV *td)
{
dTHX;
dSP;
SV *retval,
*TDsv;
int count;
ENTER;
SAVETMPS;
TDsv = get_sv("main::_TD", 0);
if (!TDsv)
ereport(ERROR,
(errcode(ERRCODE_EXTERNAL_ROUTINE_EXCEPTION),
errmsg("couldn't fetch $_TD")));
save_item(TDsv); /* local $_TD */
sv_setsv(TDsv, td);
PUSHMARK(sp);
^ permalink raw reply [nested|flat] 2+ messages in thread
end of thread, other threads:[~2024-08-28 21:53 UTC | newest]
Thread overview: 2+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2019-09-05 11:21 [PATCH v17 1/4] Move callback-call from ReadPageInternal to XLogReadRecord. Kyotaro Horiguchi <[email protected]>
2024-08-28 21:53 pl/pgperl Patch for adding $_FN detail just like triggers have for $_TD Mark Murawski <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox