8000 Allow xlogreader to use different xlog blocksize. · Giperboloid/postgres@2799b39 · GitHub
[go: up one dir, main page]
Skip to content

Commit 2799b39

Browse files
Kyotaro Horiguchimacdice
Kyotaro Horiguchi
authored and
macdice
committed
Allow xlogreader to use different xlog blocksize.
XLog reader facility and its callers currently relies on the same constant XLOG_BLCKSZ. This patch moves the responsibility to determine the block size to the callers of the facility.
1 parent 1c9e163 commit 2799b39

File tree

8 files changed

+50
-39
lines changed
Filter options

8 files changed

+50
-39
lines changed

‎src/backend/access/transam/twophase.c

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1330,7 +1330,8 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len)
13301330
char *errormsg;
13311331
TimeLineID save_currtli = ThisTimeLineID;
13321332

1333-
xlogreader = XLogReaderAllocate(wal_segment_size, NULL, wal_segment_close);
1333+
xlogreader = XLogReaderAllocate(wal_segment_size, XLOG_BLCKSZ,
1334+
NULL, wal_segment_close);
13341335

13351336
if (!xlogreader)
13361337
ereport(ERROR,

‎src/backend/access/transam/xlog.c

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1225,7 +1225,10 @@ XLogInsertRecord(XLogRecData *rdata,
12251225
appendBinaryStringInfo(&recordBuf, rdata->data, rdata->len);
12261226

12271227
if (!debug_reader)
1228-
debug_reader = XLogReaderAllocate(wal_segment_size, NULL, NULL);
1228+
{
1229+
debug_reader =
1230+
XLogReaderAllocate(wal_segment_size, XLOG_BLCKSZ, NULL, NULL);
1231+
}
12291232

12301233
if (!debug_reader)
12311234
{
@@ -6611,8 +6614,8 @@ StartupXLOG(void)
66116614
OwnLatch(&XLogCtl->recoveryWakeupLatch);
66126615

66136616
/* Set up XLOG reader facility */
6614-
xlogreader =
6615-
XLogReaderAllocate(wal_segment_size, NULL, wal_segment_close);
6617+
xlogreader = XLogReaderAllocate(wal_segment_size, XLOG_BLCKSZ,
6618+
NULL, wal_segment_close);
66166619

66176620
if (!xlogreader)
66186621
ereport(ERROR,

‎src/backend/access/transam/xlogreader.c

Lines changed: 30 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -72,7 +72,7 @@ report_invalid_record(XLogReaderState *state, const char *fmt,...)
7272
* Returns NULL if the xlogreader couldn't be allocated.
7373
*/
7474
XLogReaderState *
75-
XLogReaderAllocate(int wal_segment_size, const char *waldir,
75+
XLogReaderAllocate(int wal_segment_size, int page_size, const char *waldir,
7676
WALSegmentCleanupCB cleanup_cb)
7777
{
7878
XLogReaderState *state;
@@ -87,6 +87,7 @@ XLogReaderAllocate(int wal_segment_size, const char *waldir,
8787
state->cleanup_cb = cleanup_cb;
8888

8989
state->max_block_id = -1;
90+
state->pageSize = page_size;
9091

9192
/*
9293
* Permanently allocate readBuf. We do it this way, rather than just
@@ -95,7 +96,7 @@ XLogReaderAllocate(int wal_segment_size, const char *waldir,
9596
* isn't guaranteed to have any particular alignment, whereas
9697
* palloc_extended() will provide MAXALIGN'd storage.
9798
*/
98-
state->readBuf = (char *) palloc_extended(XLOG_BLCKSZ,
99+
state->readBuf = (char *) palloc_extended(state->pageSize,
99100
MCXT_ALLOC_NO_OOM);
100101
if (!state->readBuf)
101102
{
@@ -163,7 +164,7 @@ XLogReaderFree(XLogReaderState *state)
163164
* readRecordBufSize is set to the new buffer size.
164165
*
165166
* To avoid useless small increases, round its size to a multiple of
166-
* XLOG_BLCKSZ, and make sure it's at least 5*Max(BLCKSZ, XLOG_BLCKSZ) to start
167+
* page size, and make sure it's at least 5*Max(BLCKSZ, <page size>) to start
167168
* with. (That is enough for all "normal" records, but very large commit or
168169
* abort records might need more space.)
169170
*/
@@ -172,8 +173,8 @@ allocate_recordbuf(XLogReaderState *state, uint32 reclength)
172173
{
173174
uint32 newSize = reclength;
174175

175-
newSize += XLOG_BLCKSZ - (newSize % XLOG_BLCKSZ);
176-
newSize = Max(newSize, 5 * Max(BLCKSZ, XLOG_BLCKSZ));
176+
newSize += state->pageSize - (newSize % state->pageSize);
177+
newSize = Max(newSize, 5 * Max(BLCKSZ, state->pageSize));
177178

178179
#ifndef FRONTEND
179180

@@ -238,6 +239,8 @@ void
238239
XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
239240
{
240241
Assert(!XLogRecPtrIsInvalid(RecPtr));
242+
Assert(state->readBuf != NULL &&
243+
state->readBuf == (void *) MAXALIGN(state->readBuf));
241244

242245
ResetDecoder(state);
243246

@@ -366,15 +369,15 @@ XLogReadRecord(XLogReaderState *state, XLogRecord **record, char **errormsg)
366369
XLogPageHeader pageHeader;
367370

368371
targetPagePtr =
369-
state->ReadRecPtr - (state->ReadRecPtr % XLOG_BLCKSZ);
370-
targetRecOff = state->ReadRecPtr % XLOG_BLCKSZ;
372+
state->ReadRecPtr - (state->ReadRecPtr % state->pageSize);
373+
targetRecOff = state->ReadRecPtr % state->pageSize;
371374

372375
/*
373376
* Check if we have enough data. For the first record in the page,
374377
* the requesting length doesn't contain page header.
375378
*/
376379
if (XLogNeedData(state, targetPagePtr,
377-
Min(targetRecOff + SizeOfXLogRecord, XLOG_BLCKSZ),
380+
Min(targetRecOff + SizeOfXLogRecord, state->pageSize),
378381
targetRecOff != 0))
379382
return XLREAD_NEED_DATA;
380383

@@ -421,7 +424,7 @@ XLogReadRecord(XLogReaderState *state, XLogRecord **record, char **errormsg)
421424
* verified that we got the whole header.
422425
*/
423426
prec = (XLogRecord *) (state->readBuf +
424-
state->ReadRecPtr % XLOG_BLCKSZ);
427+
state->ReadRecPtr % state->pageSize);
425428
total_len = prec->xl_tot_len;
426429

427430
/*
@@ -432,7 +435,7 @@ XLogReadRecord(XLogReaderState *state, XLogRecord **record, char **errormsg)
432435
* ensure that we enter the XLREAD_CONTINUATION state below;
433436
* otherwise we might fail to apply ValidXLogRecordHeader at all.
434437
*/
435-
if (targetRecOff <= XLOG_BLCKSZ - SizeOfXLogRecord)
438+
if (targetRecOff <= state->pageSize - SizeOfXLogRecord)
436439
{
437440
if (!ValidXLogRecordHeader(state, state->ReadRecPtr,
438441
state->PrevRecPtr, prec))
@@ -477,10 +480,10 @@ XLogReadRecord(XLogReaderState *state, XLogRecord **record, char **errormsg)
477480
* available
478481
*/
479482
targetPagePtr =
480-
state->ReadRecPtr - (state->ReadRecPtr % XLOG_BLCKSZ);
481-
targetRecOff = state->ReadRecPtr % XLOG_BLCKSZ;
483+
state->ReadRecPtr - (state->ReadRecPtr % state->pageSize);
484+
targetRecOff = state->ReadRecPtr % state->pageSize;
482485

483-
request_len = Min(targetRecOff + total_len, XLOG_BLCKSZ);
486+
request_len = Min(targetRecOff + total_len, state->pageSize);
484487
record_len = request_len - targetRecOff;
485488

486489
/* ReadRecPtr contains page header */
@@ -549,7 +552,7 @@ XLogReadRecord(XLogReaderState *state, XLogRecord **record, char **errormsg)
549552

550553
/* Calculate pointer to beginning of next page */
551554
state->recordContRecPtr = state->ReadRecPtr + record_len;
552-
Assert(state->recordContRecPtr % XLOG_BLCKSZ == 0);
555+
Assert(state->recordContRecPtr % state->pageSize == 0);
553556

554557
state->readRecordState = XLREAD_CONTINUATION;
555558
}
@@ -576,7 +579,7 @@ XLogReadRecord(XLogReaderState *state, XLogRecord **record, char **errormsg)
576579
/* this request contains page header */
577580
Assert (targetPagePtr != 0);
578581
if (XLogNeedData(state, targetPagePtr,
579-
Min(state->recordRemainLen, XLOG_BLCKSZ),
582+
Min(state->recordRemainLen, state->pageSize),
580583
false))
581584
return XLREAD_NEED_DATA;
582585

@@ -628,7 +631,7 @@ XLogReadRecord(XLogReaderState *state, XLogRecord **record, char **errormsg)
628631
Assert(state->readLen >= pageHeaderSize);
629632

630633
contdata = (char *) state->readBuf + pageHeaderSize;
631-
record_len = XLOG_BLCKSZ - pageHeaderSize;
634+
record_len = state->pageSize - pageHeaderSize;
632635
if (pageHeader->xlp_rem_len < record_len)
633636
record_len = pageHeader->xlp_rem_len;
634637

@@ -655,7 +658,7 @@ XLogReadRecord(XLogReaderState *state, XLogRecord **record, char **errormsg)
655658
}
656659

657660
/* Calculate pointer to beginning of next page, and continue */
658-
state->recordContRecPtr += XLOG_BLCKSZ;
661+
state->recordContRecPtr += state->pageSize;
659662
}
660663

661664
/* targetPagePtr is pointing the last-read page here */
@@ -789,10 +792,10 @@ XLogNeedData(XLogReaderState *state, XLogRecPtr pageptr, int reqLen,
789792
XLogPageHeaderSize((XLogPageHeader) state->readBuf);
790793

791794
addLen = pageHeaderSize;
792-
if (reqLen + pageHeaderSize <= XLOG_BLCKSZ)
795+
if (reqLen + pageHeaderSize <= state->pageSize)
793796
addLen = pageHeaderSize;
794797
else
795-
addLen = XLOG_BLCKSZ - reqLen;
798+
addLen = state->pageSize - reqLen;
796799

797800
Assert(addLen >= 0);
798801
}
@@ -805,7 +808,7 @@ XLogNeedData(XLogReaderState *state, XLogRecPtr pageptr, int reqLen,
805808
/* Data is not in our buffer, request the caller for it. */
806809
XLByteToSeg(pageptr, targetSegNo, state->segcxt.ws_segsize);
807810
targetPageOff = XLogSegmentOffset(pageptr, state->segcxt.ws_segsize);
808-
Assert((pageptr % XLOG_BLCKSZ) == 0);
811+
Assert((pageptr % state->pageSize) == 0);
809812

810813
/*
811814
* Every time we request to load new data of a page to the caller, even if
@@ -829,7 +832,7 @@ XLogNeedData(XLogReaderState *state, XLogRecPtr pageptr, int reqLen,
829832
* will not come back here, but will request the actual target page.
830833
*/
831834
state->readPagePtr = pageptr - targetPageOff;
832-
state->readLen = XLOG_BLCKSZ;
835+
state->readLen = state->pageSize;
833836
return true;
834837
}
835838

@@ -962,7 +965,7 @@ XLogReaderValidatePageHeader(XLogReaderState *state, XLogRecPtr recptr,
962965
int32 offset;
963966
XLogPageHeader hdr = (XLogPageHeader) phdr;
964967

965-
Assert((recptr % XLOG_BLCKSZ) == 0);
968+
Assert((recptr % state->pageSize) == 0);
966969

967970
XLByteToSeg(recptr, segno, state->segcxt.ws_segsize);
968971
offset = XLogSegmentOffset(recptr, state->segcxt.ws_segsize);
@@ -1016,10 +1019,10 @@ XLogReaderValidatePageHeader(XLogReaderState *state, XLogRecPtr recptr,
10161019
"WAL file is from different database system: incorrect segment size in page header");
10171020
return false;
10181021
}
1019-
else if (longhdr->xlp_xlog_blcksz != XLOG_BLCKSZ)
1022+
else if (longhdr->xlp_xlog_blcksz != state->pageSize)
10201023
{
10211024
report_invalid_record(state,
1022-
"WAL file is from different database system: incorrect XLOG_BLCKSZ in page header");
1025+
"WAL file is from different database system: incorrect block size in page header");
10231026
return false;
10241027
}
10251028
}
@@ -1140,7 +1143,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr,
11401143
* XLogNeedData() is prepared to handle that and will read at
11411144
* least short page-header worth of data
11421145
*/
1143-
targetRecOff = tmpRecPtr % XLOG_BLCKSZ;
1146+
targetRecOff = tmpRecPtr % state->pageSize;
11441147

11451148
/* scroll back to page boundary */
11461149
targetPagePtr = tmpRecPtr - targetRecOff;
@@ -1174,8 +1177,8 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr,
11741177
*
11751178
* Note that record headers are MAXALIGN'ed
11761179
*/
1177-
if (MAXALIGN(header->xlp_rem_len) >= (XLOG_BLCKSZ - pageHeaderSize))
1178-
tmpRecPtr = targetPagePtr + XLOG_BLCKSZ;
1180+
if (MAXALIGN(header->xlp_rem_len) >= (state->pageSize - pageHeaderSize))
1181+
tmpRecPtr = targetPagePtr + state->pageSize;
11791182
else
11801183
{
11811184
/*

‎src/backend/replication/logical/logical.c

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -199,11 +199,14 @@ StartupDecodingContext(List *output_plugin_options,
199199

200200
ctx->slot = slot;
201201

202-
ctx->reader = XLogReaderAllocate(wal_segment_size, NULL, cleanup_cb);
202+
ctx->reader =
203+
XLogReaderAllocate(wal_segment_size, XLOG_BLCKSZ, NULL, cleanup_cb);
204+
203205
if (!ctx->reader)
204206
ereport(ERROR,
205207
(errcode(ERRCODE_OUT_OF_MEMORY),
206208
errmsg("out of memory")));
209+
ctx->reader->readBuf = palloc(XLOG_BLCKSZ);
207210
ctx->page_read = page_read;
208211

209212
ctx->reorder = ReorderBufferAllocate();

‎src/backend/replication/walsender.c

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -579,8 +579,8 @@ StartReplication(StartReplicationCmd *cmd)
579579
errmsg("IDENTIFY_SYSTEM has not been run before START_REPLICATION")));
580580

581581
/* create xlogreader for physical replication */
582-
xlogreader =
583-
XLogReaderAllocate(wal_segment_size, NULL, wal_segment_close);
582+
xlogreader = XLogReaderAllocate(wal_segment_size, XLOG_BLCKSZ,
583+
NULL, wal_segment_close);
584584

585585
if (!xlogreader)
586586
ereport(ERROR,

‎src/bin/pg_rewind/parsexlog.c

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,7 @@ extractPageMap(const char *datadir, XLogRecPtr startpoint, int tliIndex,
6161
XLogReaderState *xlogreader;
6262
char *errormsg;
6363

64-
xlogreader = XLogReaderAllocate(WalSegSz, datadir, NULL);
64+
xlogreader = XLogReaderAllocate(WalSegSz, XLOG_BLCKSZ, datadir, NULL);
6565

6666
if (xlogreader == NULL)
6767
pg_fatal("out of memory");
@@ -121,7 +121,7 @@ readOneRecord(const char *datadir, XLogRecPtr ptr, int tliIndex,
121121
char *errormsg;
122122
XLogRecPtr endptr;
123123

124-
xlogreader = XLogReaderAllocate(WalSegSz, datadir, NULL);
124+
xlogreader = XLogReaderAllocate(WalSegSz, XLOG_BLCKSZ, datadir, NULL);
125125
if (xlogreader == NULL)
126126
pg_fatal("out of memory");
127127

@@ -181,7 +181,7 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex,
181181
forkptr += SizeOfXLogShortPHD;
182182
}
183183

184-
xlogreader = XLogReaderAllocate(WalSegSz, datadir, NULL);
184+
xlogreader = XLogReaderAllocate(WalSegSz, XLOG_BLCKSZ, datadir, NULL);
185185
if (xlogreader == NULL)
186186
pg_fatal("out of memory");
187187

‎src/bin/pg_waldump/pg_waldump.c

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1050,7 +1050,7 @@ main(int argc, char **argv)
10501050

10511051
/* we have everything we need, start reading */
10521052
xlogreader_state =
1053-
XLogReaderAllocate(WalSegSz, waldir, WALDumpCloseSegment);
1053+
XLogReaderAllocate(WalSegSz, XLOG_BLCKSZ, waldir, WALDumpCloseSegment);
10541054

10551055
if (!xlogreader_state)
10561056
fatal_error("out of memory");

‎src/include/access/xlogreader.h

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -151,6 +151,7 @@ struct XLogReaderState
151151
* ----------------------------------------
152152
*/
153153
/* variables to communicate with page reader */
154+
int pageSize; /* size of a page */
154155
XLogRecPtr readPagePtr; /* page pointer to read */
155156
int32 readLen; /* bytes requested to reader, or actual bytes
156157
* read by reader, which must be larger than
@@ -243,7 +244,7 @@ struct XLogReaderState
243244

244245
/* Get a new XLogReader */
245246
extern XLogReaderState *XLogReaderAllocate(int wal_segment_size,
246-
const char *waldir,
247+
int page_size, const char *waldir,
247248
WALSegmentCleanupCB cleanup_cb);
248249

249250
/* Free an XLogReader */

0 commit comments

Comments
 (0)
0