diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c index a8407be09fc..cc61f830982 100644 --- a/src/backend/access/transam/xlogreader.c +++ b/src/backend/access/transam/xlogreader.c @@ -56,9 +56,12 @@ static bool ValidXLogRecord(XLogReaderState *state, XLogRecord *record, static void ResetDecoder(XLogReaderState *state); static void WALOpenSegmentInit(WALOpenSegment *seg, WALSegmentContext *segcxt, int segsize, const char *waldir); -#if defined(USE_ZSTD) && !defined(FRONTEND) +#ifndef FRONTEND +static void xlogreader_memory_context_reset_cb(void *arg); +#ifdef USE_ZSTD static void XLogReaderFreeZstdContext(void *arg); #endif +#endif /* size of the buffer allocated for error message. */ #define MAX_ERRORMSG_LEN 1000 @@ -166,6 +169,12 @@ XLogReaderAllocate(int wal_segment_size, const char *waldir, void XLogReaderFree(XLogReaderState *state) { +#ifndef FRONTEND + if (state->reset_cb_registered) + MemoryContextUnregisterResetCallback(GetMemoryChunkContext(state), + &state->reset_cb); +#endif + if (state->seg.ws_file != -1) state->routine.segment_close(state); @@ -189,7 +198,43 @@ XLogReaderFree(XLogReaderState *state) pfree(state); } -#if defined(USE_ZSTD) && !defined(FRONTEND) +#ifndef FRONTEND + +/* + * Memory context reset callback for an XLogReader. + */ +static void +xlogreader_memory_context_reset_cb(void *arg) +{ + XLogReaderState *state = (XLogReaderState *) arg; + + /* Close the WAL segment file that was left open */ + if (state->seg.ws_file >= 0) + state->routine.segment_close(state); +} + +/* + * Register a memory reset callback, closing a segment, if necessary. + * + * This is useful when opening a WAL segment file with BasicOpenFile(), which + * doesn't have the automatic file descriptor cleanup that OpenTransientFile() + * provides, to guarantee that the segment is closed before XLogReaderFree() is + * reached (on ERRORs, for example). + */ +void +XLogReaderRegisterResetCallback(XLogReaderState *state) +{ + if (state->reset_cb_registered) + return; + + state->reset_cb.func = xlogreader_memory_context_reset_cb; + state->reset_cb.arg = state; + MemoryContextRegisterResetCallback(GetMemoryChunkContext(state), + &state->reset_cb); + state->reset_cb_registered = true; +} + +#ifdef USE_ZSTD /* Release the zstd decomoression context. */ static void XLogReaderFreeZstdContext(void *arg) @@ -198,6 +243,8 @@ XLogReaderFreeZstdContext(void *arg) } #endif +#endif /* FRONTEND */ + /* * Allocate readRecordBuf to fit a record of at least the given length. * diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c index 58b9dab6a90..ff71e14015d 100644 --- a/src/backend/access/transam/xlogutils.c +++ b/src/backend/access/transam/xlogutils.c @@ -836,7 +836,10 @@ wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo, XLogFilePath(path, tli, nextSegNo, state->segcxt.ws_segsize); state->seg.ws_file = BasicOpenFile(path, O_RDONLY | PG_BINARY); if (state->seg.ws_file >= 0) + { + XLogReaderRegisterResetCallback(state); return; + } if (errno == ENOENT) ereport(ERROR, diff --git a/src/backend/postmaster/walsummarizer.c b/src/backend/postmaster/walsummarizer.c index ff246b07a21..8333a4eaaa3 100644 --- a/src/backend/postmaster/walsummarizer.c +++ b/src/backend/postmaster/walsummarizer.c @@ -1608,6 +1608,7 @@ summarizer_wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo, state->seg.ws_file = BasicOpenFile(path, O_RDONLY | PG_BINARY); if (state->seg.ws_file >= 0) { + XLogReaderRegisterResetCallback(state); *tli_p = tli; return; } diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index e9331de3df5..10bc3e0cae4 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -3346,7 +3346,10 @@ WalSndSegmentOpen(XLogReaderState *state, XLogSegNo nextSegNo, XLogFilePath(path, *tli_p, nextSegNo, state->segcxt.ws_segsize); state->seg.ws_file = BasicOpenFile(path, O_RDONLY | PG_BINARY); if (state->seg.ws_file >= 0) + { + XLogReaderRegisterResetCallback(state); return; + } /* * If the file is not found, assume it's because the standby asked for a diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h index 64a9492f6ed..e48a0db9ce9 100644 --- a/src/include/access/xlogreader.h +++ b/src/include/access/xlogreader.h @@ -109,7 +109,9 @@ typedef struct XLogReaderRoutine /* * WAL segment close callback. ->seg.ws_file shall be set to a negative - * number. + * number. This shall not raise an error, as it may be called in a memory + * context reset callback if XLogReaderRegisterResetCallback() has been + * used. */ WALSegmentCloseCB segment_close; } XLogReaderRoutine; @@ -324,6 +326,12 @@ struct XLogReaderState * data. */ bool nonblocking; + +#ifndef FRONTEND + /* Reset callback for the memory context holding this reader. */ + MemoryContextCallback reset_cb; + bool reset_cb_registered; +#endif }; /* @@ -344,6 +352,11 @@ extern XLogReaderState *XLogReaderAllocate(int wal_segment_size, /* Free an XLogReader */ extern void XLogReaderFree(XLogReaderState *state); +#ifndef FRONTEND +/* Register a memory context reset callback */ +extern void XLogReaderRegisterResetCallback(XLogReaderState *state); +#endif + /* Optionally provide a circular decoding buffer to allow readahead. */ extern void XLogReaderSetDecodeBuffer(XLogReaderState *state, void *buffer,