Commit 0ab9d1c4 authored by Heikki Linnakangas's avatar Heikki Linnakangas

Replace XLogRecPtr struct with a 64-bit integer.

This simplifies code that needs to do arithmetic on XLogRecPtrs.

To avoid changing on-disk format of data pages, the LSN on data pages is
still stored in the old format. That should keep pg_upgrade happy. However,
we have XLogRecPtrs embedded in the control file, and in the structs that
are sent over the replication protocol, so this changes breaks compatibility
of pg_basebackup and server. I didn't do anything about this in this patch,
per discussion on -hackers, the right thing to do would to be to change the
replication protocol to be architecture-independent, so that you could use
a newer version of pg_receivexlog, for example, against an older server
version.
parent 061e7efb
...@@ -206,7 +206,8 @@ page_header(PG_FUNCTION_ARGS) ...@@ -206,7 +206,8 @@ page_header(PG_FUNCTION_ARGS)
/* Extract information from the page header */ /* Extract information from the page header */
lsn = PageGetLSN(page); lsn = PageGetLSN(page);
snprintf(lsnchar, sizeof(lsnchar), "%X/%X", lsn.xlogid, lsn.xrecoff); snprintf(lsnchar, sizeof(lsnchar), "%X/%X",
(uint32) (lsn >> 32), (uint32) lsn);
values[0] = CStringGetTextDatum(lsnchar); values[0] = CStringGetTextDatum(lsnchar);
values[1] = UInt16GetDatum(PageGetTLI(page)); values[1] = UInt16GetDatum(PageGetTLI(page));
......
...@@ -197,7 +197,7 @@ gistplacetopage(Relation rel, Size freespace, GISTSTATE *giststate, ...@@ -197,7 +197,7 @@ gistplacetopage(Relation rel, Size freespace, GISTSTATE *giststate,
SplitedPageLayout *dist = NULL, SplitedPageLayout *dist = NULL,
*ptr; *ptr;
BlockNumber oldrlink = InvalidBlockNumber; BlockNumber oldrlink = InvalidBlockNumber;
GistNSN oldnsn = {0, 0}; GistNSN oldnsn = 0;
SplitedPageLayout rootpg; SplitedPageLayout rootpg;
BlockNumber blkno = BufferGetBlockNumber(buffer); BlockNumber blkno = BufferGetBlockNumber(buffer);
bool is_rootsplit; bool is_rootsplit;
...@@ -488,7 +488,7 @@ gistdoinsert(Relation r, IndexTuple itup, Size freespace, GISTSTATE *giststate) ...@@ -488,7 +488,7 @@ gistdoinsert(Relation r, IndexTuple itup, Size freespace, GISTSTATE *giststate)
/* Start from the root */ /* Start from the root */
firststack.blkno = GIST_ROOT_BLKNO; firststack.blkno = GIST_ROOT_BLKNO;
firststack.lsn.xrecoff = 0; firststack.lsn = 0;
firststack.parent = NULL; firststack.parent = NULL;
firststack.downlinkoffnum = InvalidOffsetNumber; firststack.downlinkoffnum = InvalidOffsetNumber;
state.stack = stack = &firststack; state.stack = stack = &firststack;
......
...@@ -706,13 +706,7 @@ gistoptions(PG_FUNCTION_ARGS) ...@@ -706,13 +706,7 @@ gistoptions(PG_FUNCTION_ARGS)
XLogRecPtr XLogRecPtr
GetXLogRecPtrForTemp(void) GetXLogRecPtrForTemp(void)
{ {
static XLogRecPtr counter = {0, 1}; static XLogRecPtr counter = 1;
counter++;
counter.xrecoff++;
if (counter.xrecoff == 0)
{
counter.xlogid++;
counter.xrecoff++;
}
return counter; return counter;
} }
...@@ -24,9 +24,6 @@ ...@@ -24,9 +24,6 @@
#include "access/transam.h" #include "access/transam.h"
#include "utils/snapmgr.h" #include "utils/snapmgr.h"
/* Handy constant for an invalid xlog recptr */
const XLogRecPtr InvalidXLogRecPtr = {0, 0};
/* /*
* Single-item cache for results of TransactionLogFetch. It's worth having * Single-item cache for results of TransactionLogFetch. It's worth having
* such a cache because we frequently find ourselves repeatedly checking the * such a cache because we frequently find ourselves repeatedly checking the
......
...@@ -333,8 +333,7 @@ MarkAsPreparing(TransactionId xid, const char *gid, ...@@ -333,8 +333,7 @@ MarkAsPreparing(TransactionId xid, const char *gid,
gxact->prepared_at = prepared_at; gxact->prepared_at = prepared_at;
/* initialize LSN to 0 (start of WAL) */ /* initialize LSN to 0 (start of WAL) */
gxact->prepare_lsn.xlogid = 0; gxact->prepare_lsn = 0;
gxact->prepare_lsn.xrecoff = 0;
gxact->owner = owner; gxact->owner = owner;
gxact->locking_xid = xid; gxact->locking_xid = xid;
gxact->valid = false; gxact->valid = false;
......
...@@ -951,7 +951,7 @@ RecordTransactionCommit(void) ...@@ -951,7 +951,7 @@ RecordTransactionCommit(void)
if (XLogStandbyInfoActive()) if (XLogStandbyInfoActive())
nmsgs = xactGetCommittedInvalidationMessages(&invalMessages, nmsgs = xactGetCommittedInvalidationMessages(&invalMessages,
&RelcacheInitFileInval); &RelcacheInitFileInval);
wrote_xlog = (XactLastRecEnd.xrecoff != 0); wrote_xlog = (XactLastRecEnd != 0);
/* /*
* If we haven't been assigned an XID yet, we neither can, nor do we want * If we haven't been assigned an XID yet, we neither can, nor do we want
...@@ -1200,7 +1200,7 @@ RecordTransactionCommit(void) ...@@ -1200,7 +1200,7 @@ RecordTransactionCommit(void)
SyncRepWaitForLSN(XactLastRecEnd); SyncRepWaitForLSN(XactLastRecEnd);
/* Reset XactLastRecEnd until the next transaction writes something */ /* Reset XactLastRecEnd until the next transaction writes something */
XactLastRecEnd.xrecoff = 0; XactLastRecEnd = 0;
cleanup: cleanup:
/* Clean up local data */ /* Clean up local data */
...@@ -1402,7 +1402,7 @@ RecordTransactionAbort(bool isSubXact) ...@@ -1402,7 +1402,7 @@ RecordTransactionAbort(bool isSubXact)
{ {
/* Reset XactLastRecEnd until the next transaction writes something */ /* Reset XactLastRecEnd until the next transaction writes something */
if (!isSubXact) if (!isSubXact)
XactLastRecEnd.xrecoff = 0; XactLastRecEnd = 0;
return InvalidTransactionId; return InvalidTransactionId;
} }
...@@ -1501,7 +1501,7 @@ RecordTransactionAbort(bool isSubXact) ...@@ -1501,7 +1501,7 @@ RecordTransactionAbort(bool isSubXact)
/* Reset XactLastRecEnd until the next transaction writes something */ /* Reset XactLastRecEnd until the next transaction writes something */
if (!isSubXact) if (!isSubXact)
XactLastRecEnd.xrecoff = 0; XactLastRecEnd = 0;
/* And clean up local data */ /* And clean up local data */
if (rels) if (rels)
...@@ -2170,7 +2170,7 @@ PrepareTransaction(void) ...@@ -2170,7 +2170,7 @@ PrepareTransaction(void)
*/ */
/* Reset XactLastRecEnd until the next transaction writes something */ /* Reset XactLastRecEnd until the next transaction writes something */
XactLastRecEnd.xrecoff = 0; XactLastRecEnd = 0;
/* /*
* Let others know about no transaction in progress by me. This has to be * Let others know about no transaction in progress by me. This has to be
......
This diff is collapsed.
...@@ -57,7 +57,7 @@ pg_start_backup(PG_FUNCTION_ARGS) ...@@ -57,7 +57,7 @@ pg_start_backup(PG_FUNCTION_ARGS)
startpoint = do_pg_start_backup(backupidstr, fast, NULL); startpoint = do_pg_start_backup(backupidstr, fast, NULL);
snprintf(startxlogstr, sizeof(startxlogstr), "%X/%X", snprintf(startxlogstr, sizeof(startxlogstr), "%X/%X",
startpoint.xlogid, startpoint.xrecoff); (uint32) (startpoint >> 32), (uint32) startpoint);
PG_RETURN_TEXT_P(cstring_to_text(startxlogstr)); PG_RETURN_TEXT_P(cstring_to_text(startxlogstr));
} }
...@@ -83,7 +83,7 @@ pg_stop_backup(PG_FUNCTION_ARGS) ...@@ -83,7 +83,7 @@ pg_stop_backup(PG_FUNCTION_ARGS)
stoppoint = do_pg_stop_backup(NULL, true); stoppoint = do_pg_stop_backup(NULL, true);
snprintf(stopxlogstr, sizeof(stopxlogstr), "%X/%X", snprintf(stopxlogstr, sizeof(stopxlogstr), "%X/%X",
stoppoint.xlogid, stoppoint.xrecoff); (uint32) (stoppoint >> 32), (uint32) stoppoint);
PG_RETURN_TEXT_P(cstring_to_text(stopxlogstr)); PG_RETURN_TEXT_P(cstring_to_text(stopxlogstr));
} }
...@@ -113,7 +113,7 @@ pg_switch_xlog(PG_FUNCTION_ARGS) ...@@ -113,7 +113,7 @@ pg_switch_xlog(PG_FUNCTION_ARGS)
* As a convenience, return the WAL location of the switch record * As a convenience, return the WAL location of the switch record
*/ */
snprintf(location, sizeof(location), "%X/%X", snprintf(location, sizeof(location), "%X/%X",
switchpoint.xlogid, switchpoint.xrecoff); (uint32) (switchpoint >> 32), (uint32) switchpoint);
PG_RETURN_TEXT_P(cstring_to_text(location)); PG_RETURN_TEXT_P(cstring_to_text(location));
} }
...@@ -158,7 +158,7 @@ pg_create_restore_point(PG_FUNCTION_ARGS) ...@@ -158,7 +158,7 @@ pg_create_restore_point(PG_FUNCTION_ARGS)
* As a convenience, return the WAL location of the restore point record * As a convenience, return the WAL location of the restore point record
*/ */
snprintf(location, sizeof(location), "%X/%X", snprintf(location, sizeof(location), "%X/%X",
restorepoint.xlogid, restorepoint.xrecoff); (uint32) (restorepoint >> 32), (uint32) restorepoint);
PG_RETURN_TEXT_P(cstring_to_text(location)); PG_RETURN_TEXT_P(cstring_to_text(location));
} }
...@@ -184,7 +184,7 @@ pg_current_xlog_location(PG_FUNCTION_ARGS) ...@@ -184,7 +184,7 @@ pg_current_xlog_location(PG_FUNCTION_ARGS)
current_recptr = GetXLogWriteRecPtr(); current_recptr = GetXLogWriteRecPtr();
snprintf(location, sizeof(location), "%X/%X", snprintf(location, sizeof(location), "%X/%X",
current_recptr.xlogid, current_recptr.xrecoff); (uint32) (current_recptr >> 32), (uint32) current_recptr);
PG_RETURN_TEXT_P(cstring_to_text(location)); PG_RETURN_TEXT_P(cstring_to_text(location));
} }
...@@ -208,7 +208,7 @@ pg_current_xlog_insert_location(PG_FUNCTION_ARGS) ...@@ -208,7 +208,7 @@ pg_current_xlog_insert_location(PG_FUNCTION_ARGS)
current_recptr = GetXLogInsertRecPtr(); current_recptr = GetXLogInsertRecPtr();
snprintf(location, sizeof(location), "%X/%X", snprintf(location, sizeof(location), "%X/%X",
current_recptr.xlogid, current_recptr.xrecoff); (uint32) (current_recptr >> 32), (uint32) current_recptr);
PG_RETURN_TEXT_P(cstring_to_text(location)); PG_RETURN_TEXT_P(cstring_to_text(location));
} }
...@@ -226,11 +226,11 @@ pg_last_xlog_receive_location(PG_FUNCTION_ARGS) ...@@ -226,11 +226,11 @@ pg_last_xlog_receive_location(PG_FUNCTION_ARGS)
recptr = GetWalRcvWriteRecPtr(NULL); recptr = GetWalRcvWriteRecPtr(NULL);
if (recptr.xlogid == 0 && recptr.xrecoff == 0) if (recptr == 0)
PG_RETURN_NULL(); PG_RETURN_NULL();
snprintf(location, sizeof(location), "%X/%X", snprintf(location, sizeof(location), "%X/%X",
recptr.xlogid, recptr.xrecoff); (uint32) (recptr >> 32), (uint32) recptr);
PG_RETURN_TEXT_P(cstring_to_text(location)); PG_RETURN_TEXT_P(cstring_to_text(location));
} }
...@@ -248,11 +248,11 @@ pg_last_xlog_replay_location(PG_FUNCTION_ARGS) ...@@ -248,11 +248,11 @@ pg_last_xlog_replay_location(PG_FUNCTION_ARGS)
recptr = GetXLogReplayRecPtr(NULL); recptr = GetXLogReplayRecPtr(NULL);
if (recptr.xlogid == 0 && recptr.xrecoff == 0) if (recptr == 0)
PG_RETURN_NULL(); PG_RETURN_NULL();
snprintf(location, sizeof(location), "%X/%X", snprintf(location, sizeof(location), "%X/%X",
recptr.xlogid, recptr.xrecoff); (uint32) (recptr >> 32), (uint32) recptr);
PG_RETURN_TEXT_P(cstring_to_text(location)); PG_RETURN_TEXT_P(cstring_to_text(location));
} }
...@@ -269,8 +269,8 @@ pg_xlogfile_name_offset(PG_FUNCTION_ARGS) ...@@ -269,8 +269,8 @@ pg_xlogfile_name_offset(PG_FUNCTION_ARGS)
{ {
text *location = PG_GETARG_TEXT_P(0); text *location = PG_GETARG_TEXT_P(0);
char *locationstr; char *locationstr;
unsigned int uxlogid; uint32 hi,
unsigned int uxrecoff; lo;
XLogSegNo xlogsegno; XLogSegNo xlogsegno;
uint32 xrecoff; uint32 xrecoff;
XLogRecPtr locationpoint; XLogRecPtr locationpoint;
...@@ -294,14 +294,12 @@ pg_xlogfile_name_offset(PG_FUNCTION_ARGS) ...@@ -294,14 +294,12 @@ pg_xlogfile_name_offset(PG_FUNCTION_ARGS)
validate_xlog_location(locationstr); validate_xlog_location(locationstr);
if (sscanf(locationstr, "%X/%X", &uxlogid, &uxrecoff) != 2) if (sscanf(locationstr, "%X/%X", &hi, &lo) != 2)
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE), (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("could not parse transaction log location \"%s\"", errmsg("could not parse transaction log location \"%s\"",
locationstr))); locationstr)));
locationpoint = ((uint64) hi) << 32 | lo;
locationpoint.xlogid = uxlogid;
locationpoint.xrecoff = uxrecoff;
/* /*
* Construct a tuple descriptor for the result row. This must match this * Construct a tuple descriptor for the result row. This must match this
...@@ -327,7 +325,7 @@ pg_xlogfile_name_offset(PG_FUNCTION_ARGS) ...@@ -327,7 +325,7 @@ pg_xlogfile_name_offset(PG_FUNCTION_ARGS)
/* /*
* offset * offset
*/ */
xrecoff = locationpoint.xrecoff % XLogSegSize; xrecoff = locationpoint % XLogSegSize;
values[1] = UInt32GetDatum(xrecoff); values[1] = UInt32GetDatum(xrecoff);
isnull[1] = false; isnull[1] = false;
...@@ -351,8 +349,8 @@ pg_xlogfile_name(PG_FUNCTION_ARGS) ...@@ -351,8 +349,8 @@ pg_xlogfile_name(PG_FUNCTION_ARGS)
{ {
text *location = PG_GETARG_TEXT_P(0); text *location = PG_GETARG_TEXT_P(0);
char *locationstr; char *locationstr;
unsigned int uxlogid; uint32 hi,
unsigned int uxrecoff; lo;
XLogSegNo xlogsegno; XLogSegNo xlogsegno;
XLogRecPtr locationpoint; XLogRecPtr locationpoint;
char xlogfilename[MAXFNAMELEN]; char xlogfilename[MAXFNAMELEN];
...@@ -367,14 +365,12 @@ pg_xlogfile_name(PG_FUNCTION_ARGS) ...@@ -367,14 +365,12 @@ pg_xlogfile_name(PG_FUNCTION_ARGS)
validate_xlog_location(locationstr); validate_xlog_location(locationstr);
if (sscanf(locationstr, "%X/%X", &uxlogid, &uxrecoff) != 2) if (sscanf(locationstr, "%X/%X", &hi, &lo) != 2)
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE), (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("could not parse transaction log location \"%s\"", errmsg("could not parse transaction log location \"%s\"",
locationstr))); locationstr)));
locationpoint = ((uint64) hi) << 32 | lo;
locationpoint.xlogid = uxlogid;
locationpoint.xrecoff = uxrecoff;
XLByteToPrevSeg(locationpoint, xlogsegno); XLByteToPrevSeg(locationpoint, xlogsegno);
XLogFileName(xlogfilename, ThisTimeLineID, xlogsegno); XLogFileName(xlogfilename, ThisTimeLineID, xlogsegno);
...@@ -514,6 +510,8 @@ pg_xlog_location_diff(PG_FUNCTION_ARGS) ...@@ -514,6 +510,8 @@ pg_xlog_location_diff(PG_FUNCTION_ARGS)
Numeric result; Numeric result;
uint64 bytes1, uint64 bytes1,
bytes2; bytes2;
uint32 hi,
lo;
/* /*
* Read and parse input * Read and parse input
...@@ -524,17 +522,20 @@ pg_xlog_location_diff(PG_FUNCTION_ARGS) ...@@ -524,17 +522,20 @@ pg_xlog_location_diff(PG_FUNCTION_ARGS)
validate_xlog_location(str1); validate_xlog_location(str1);
validate_xlog_location(str2); validate_xlog_location(str2);
if (sscanf(str1, "%X/%X", &loc1.xlogid, &loc1.xrecoff) != 2) if (sscanf(str1, "%X/%X", &hi, &lo) != 2)
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE), (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("could not parse transaction log location \"%s\"", str1))); errmsg("could not parse transaction log location \"%s\"", str1)));
if (sscanf(str2, "%X/%X", &loc2.xlogid, &loc2.xrecoff) != 2) loc1 = ((uint64) hi) << 32 | lo;
if (sscanf(str2, "%X/%X", &hi, &lo) != 2)
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE), (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("could not parse transaction log location \"%s\"", str2))); errmsg("could not parse transaction log location \"%s\"", str2)));
loc2 = ((uint64) hi) << 32 | lo;
bytes1 = (((uint64)loc1.xlogid) << 32L) + loc1.xrecoff; bytes1 = (uint64) loc1;
bytes2 = (((uint64)loc2.xlogid) << 32L) + loc2.xrecoff; bytes2 = (uint64) loc2;
/* /*
* result = bytes1 - bytes2. * result = bytes1 - bytes2.
......
...@@ -631,7 +631,7 @@ CheckArchiveTimeout(void) ...@@ -631,7 +631,7 @@ CheckArchiveTimeout(void)
* If the returned pointer points exactly to a segment boundary, * If the returned pointer points exactly to a segment boundary,
* assume nothing happened. * assume nothing happened.
*/ */
if ((switchpoint.xrecoff % XLogSegSize) != 0) if ((switchpoint % XLogSegSize) != 0)
ereport(DEBUG1, ereport(DEBUG1,
(errmsg("transaction log switch forced (archive_timeout=%d)", (errmsg("transaction log switch forced (archive_timeout=%d)",
XLogArchiveTimeout))); XLogArchiveTimeout)));
...@@ -778,10 +778,7 @@ IsCheckpointOnSchedule(double progress) ...@@ -778,10 +778,7 @@ IsCheckpointOnSchedule(double progress)
if (!RecoveryInProgress()) if (!RecoveryInProgress())
{ {
recptr = GetInsertRecPtr(); recptr = GetInsertRecPtr();
elapsed_xlogs = elapsed_xlogs = (((double) (recptr - ckpt_start_recptr)) / XLogSegSize) / CheckPointSegments;
(((double) ((uint64) (recptr.xlogid - ckpt_start_recptr.xlogid) << 32L)) +
((double) recptr.xrecoff - (double) ckpt_start_recptr.xrecoff) / XLogSegSize) /
CheckPointSegments;
if (progress < elapsed_xlogs) if (progress < elapsed_xlogs)
{ {
......
...@@ -497,7 +497,7 @@ SendXlogRecPtrResult(XLogRecPtr ptr) ...@@ -497,7 +497,7 @@ SendXlogRecPtrResult(XLogRecPtr ptr)
StringInfoData buf; StringInfoData buf;
char str[MAXFNAMELEN]; char str[MAXFNAMELEN];
snprintf(str, sizeof(str), "%X/%X", ptr.xlogid, ptr.xrecoff); snprintf(str, sizeof(str), "%X/%X", (uint32) (ptr >> 32), (uint32) ptr);
pq_beginmessage(&buf, 'T'); /* RowDescription */ pq_beginmessage(&buf, 'T'); /* RowDescription */
pq_sendint(&buf, 1, 2); /* 1 field */ pq_sendint(&buf, 1, 2); /* 1 field */
......
...@@ -156,7 +156,7 @@ libpqrcv_connect(char *conninfo, XLogRecPtr startpoint) ...@@ -156,7 +156,7 @@ libpqrcv_connect(char *conninfo, XLogRecPtr startpoint)
/* Start streaming from the point requested by startup process */ /* Start streaming from the point requested by startup process */
snprintf(cmd, sizeof(cmd), "START_REPLICATION %X/%X", snprintf(cmd, sizeof(cmd), "START_REPLICATION %X/%X",
startpoint.xlogid, startpoint.xrecoff); (uint32) (startpoint >> 32), (uint32) startpoint);
res = libpqrcv_PQexec(cmd); res = libpqrcv_PQexec(cmd);
if (PQresultStatus(res) != PGRES_COPY_BOTH) if (PQresultStatus(res) != PGRES_COPY_BOTH)
{ {
......
...@@ -72,8 +72,11 @@ START_REPLICATION { return K_START_REPLICATION; } ...@@ -72,8 +72,11 @@ START_REPLICATION { return K_START_REPLICATION; }
" " ; " " ;
{hexdigit}+\/{hexdigit}+ { {hexdigit}+\/{hexdigit}+ {
if (sscanf(yytext, "%X/%X", &yylval.recptr.xlogid, &yylval.recptr.xrecoff) != 2) uint32 hi,
lo;
if (sscanf(yytext, "%X/%X", &hi, &lo) != 2)
yyerror("invalid streaming start location"); yyerror("invalid streaming start location");
yylval.recptr = ((uint64) hi) << 32 | lo;
return RECPTR; return RECPTR;
} }
......
...@@ -145,7 +145,7 @@ SyncRepWaitForLSN(XLogRecPtr XactCommitLSN) ...@@ -145,7 +145,7 @@ SyncRepWaitForLSN(XLogRecPtr XactCommitLSN)
new_status = (char *) palloc(len + 32 + 1); new_status = (char *) palloc(len + 32 + 1);
memcpy(new_status, old_status, len); memcpy(new_status, old_status, len);
sprintf(new_status + len, " waiting for %X/%X", sprintf(new_status + len, " waiting for %X/%X",
XactCommitLSN.xlogid, XactCommitLSN.xrecoff); (uint32) (XactCommitLSN >> 32), (uint32) XactCommitLSN);
set_ps_display(new_status, false); set_ps_display(new_status, false);
new_status[len] = '\0'; /* truncate off " waiting ..." */ new_status[len] = '\0'; /* truncate off " waiting ..." */
} }
...@@ -255,8 +255,7 @@ SyncRepWaitForLSN(XLogRecPtr XactCommitLSN) ...@@ -255,8 +255,7 @@ SyncRepWaitForLSN(XLogRecPtr XactCommitLSN)
*/ */
Assert(SHMQueueIsDetached(&(MyProc->syncRepLinks))); Assert(SHMQueueIsDetached(&(MyProc->syncRepLinks)));
MyProc->syncRepState = SYNC_REP_NOT_WAITING; MyProc->syncRepState = SYNC_REP_NOT_WAITING;
MyProc->waitLSN.xlogid = 0; MyProc->waitLSN = 0;
MyProc->waitLSN.xrecoff = 0;
if (new_status) if (new_status)
{ {
...@@ -440,12 +439,8 @@ SyncRepReleaseWaiters(void) ...@@ -440,12 +439,8 @@ SyncRepReleaseWaiters(void)
LWLockRelease(SyncRepLock); LWLockRelease(SyncRepLock);
elog(DEBUG3, "released %d procs up to write %X/%X, %d procs up to flush %X/%X", elog(DEBUG3, "released %d procs up to write %X/%X, %d procs up to flush %X/%X",
numwrite, numwrite, (uint32) (MyWalSnd->write >> 32), (uint32) MyWalSnd->write,
MyWalSnd->write.xlogid, numflush, (uint32) (MyWalSnd->flush >> 32), (uint32) MyWalSnd->flush);
MyWalSnd->write.xrecoff,
numflush,
MyWalSnd->flush.xlogid,
MyWalSnd->flush.xrecoff);
/* /*
* If we are managing the highest priority standby, though we weren't * If we are managing the highest priority standby, though we weren't
...@@ -630,8 +625,7 @@ SyncRepQueueIsOrderedByLSN(int mode) ...@@ -630,8 +625,7 @@ SyncRepQueueIsOrderedByLSN(int mode)
Assert(mode >= 0 && mode < NUM_SYNC_REP_WAIT_MODE); Assert(mode >= 0 && mode < NUM_SYNC_REP_WAIT_MODE);
lastLSN.xlogid = 0; lastLSN = 0;
lastLSN.xrecoff = 0;
proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue[mode]), proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue[mode]),
&(WalSndCtl->SyncRepQueue[mode]), &(WalSndCtl->SyncRepQueue[mode]),
......
...@@ -516,7 +516,7 @@ XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr) ...@@ -516,7 +516,7 @@ XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr)
} }
/* Calculate the start offset of the received logs */ /* Calculate the start offset of the received logs */
startoff = recptr.xrecoff % XLogSegSize; startoff = recptr % XLogSegSize;
if (startoff + nbytes > XLogSegSize) if (startoff + nbytes > XLogSegSize)
segbytes = XLogSegSize - startoff; segbytes = XLogSegSize - startoff;
...@@ -601,8 +601,8 @@ XLogWalRcvFlush(bool dying) ...@@ -601,8 +601,8 @@ XLogWalRcvFlush(bool dying)
char activitymsg[50]; char activitymsg[50];
snprintf(activitymsg, sizeof(activitymsg), "streaming %X/%X", snprintf(activitymsg, sizeof(activitymsg), "streaming %X/%X",
LogstreamResult.Write.xlogid, (uint32) (LogstreamResult.Write >> 32),
LogstreamResult.Write.xrecoff); (uint32) LogstreamResult.Write);
set_ps_display(activitymsg, false); set_ps_display(activitymsg, false);
} }
...@@ -657,9 +657,9 @@ XLogWalRcvSendReply(void) ...@@ -657,9 +657,9 @@ XLogWalRcvSendReply(void)
reply_message.sendTime = now; reply_message.sendTime = now;
elog(DEBUG2, "sending write %X/%X flush %X/%X apply %X/%X", elog(DEBUG2, "sending write %X/%X flush %X/%X apply %X/%X",
reply_message.write.xlogid, reply_message.write.xrecoff, (uint32) (reply_message.write >> 32), (uint32) reply_message.write,
reply_message.flush.xlogid, reply_message.flush.xrecoff, (uint32) (reply_message.flush >> 32), (uint32) reply_message.flush,
reply_message.apply.xlogid, reply_message.apply.xrecoff); (uint32) (reply_message.apply >> 32), (uint32) reply_message.apply);
/* Prepend with the message type and send it. */ /* Prepend with the message type and send it. */
buf[0] = 'r'; buf[0] = 'r';
......
...@@ -185,8 +185,8 @@ RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo) ...@@ -185,8 +185,8 @@ RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo)
* being created by XLOG streaming, which might cause trouble later on if * being created by XLOG streaming, which might cause trouble later on if
* the segment is e.g archived. * the segment is e.g archived.
*/ */
if (recptr.xrecoff % XLogSegSize != 0) if (recptr % XLogSegSize != 0)
recptr.xrecoff -= recptr.xrecoff % XLogSegSize; recptr -= recptr % XLogSegSize;
SpinLockAcquire(&walrcv->mutex); SpinLockAcquire(&walrcv->mutex);
...@@ -204,8 +204,7 @@ RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo) ...@@ -204,8 +204,7 @@ RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo)
* If this is the first startup of walreceiver, we initialize receivedUpto * If this is the first startup of walreceiver, we initialize receivedUpto
* and latestChunkStart to receiveStart. * and latestChunkStart to receiveStart.
*/ */
if (walrcv->receiveStart.xlogid == 0 && if (walrcv->receiveStart == 0)
walrcv->receiveStart.xrecoff == 0)
{ {
walrcv->receivedUpto = recptr; walrcv->receivedUpto = recptr;
walrcv->latestChunkStart = recptr; walrcv->latestChunkStart = recptr;
......
...@@ -94,7 +94,7 @@ static uint32 sendOff = 0; ...@@ -94,7 +94,7 @@ static uint32 sendOff = 0;
* How far have we sent WAL already? This is also advertised in * How far have we sent WAL already? This is also advertised in
* MyWalSnd->sentPtr. (Actually, this is the next WAL location to send.) * MyWalSnd->sentPtr. (Actually, this is the next WAL location to send.)
*/ */
static XLogRecPtr sentPtr = {0, 0}; static XLogRecPtr sentPtr = 0;
/* /*
* Buffer for processing reply messages. * Buffer for processing reply messages.
...@@ -300,8 +300,7 @@ IdentifySystem(void) ...@@ -300,8 +300,7 @@ IdentifySystem(void)
logptr = am_cascading_walsender ? GetStandbyFlushRecPtr() : GetInsertRecPtr(); logptr = am_cascading_walsender ? GetStandbyFlushRecPtr() : GetInsertRecPtr();
snprintf(xpos, sizeof(xpos), "%X/%X", snprintf(xpos, sizeof(xpos), "%X/%X", (uint32) (logptr >> 32), (uint32) logptr);
logptr.xlogid, logptr.xrecoff);
/* Send a RowDescription message */ /* Send a RowDescription message */
pq_beginmessage(&buf, 'T'); pq_beginmessage(&buf, 'T');
...@@ -613,9 +612,9 @@ ProcessStandbyReplyMessage(void) ...@@ -613,9 +612,9 @@ ProcessStandbyReplyMessage(void)
pq_copymsgbytes(&reply_message, (char *) &reply, sizeof(StandbyReplyMessage)); pq_copymsgbytes(&reply_message, (char *) &reply, sizeof(StandbyReplyMessage));
elog(DEBUG2, "write %X/%X flush %X/%X apply %X/%X", elog(DEBUG2, "write %X/%X flush %X/%X apply %X/%X",
reply.write.xlogid, reply.write.xrecoff, (uint32) (reply.write << 32), (uint32) reply.write,
reply.flush.xlogid, reply.flush.xrecoff, (uint32) (reply.flush << 32), (uint32) reply.flush,
reply.apply.xlogid, reply.apply.xrecoff); (uint32) (reply.apply << 32), (uint32) reply.apply);
/* /*
* Update shared state for this WalSender process based on reply data from * Update shared state for this WalSender process based on reply data from
...@@ -990,7 +989,7 @@ retry: ...@@ -990,7 +989,7 @@ retry:
int segbytes; int segbytes;
int readbytes; int readbytes;
startoff = recptr.xrecoff % XLogSegSize; startoff = recptr % XLogSegSize;
if (sendFile < 0 || !XLByteInSeg(recptr, sendSegNo)) if (sendFile < 0 || !XLByteInSeg(recptr, sendSegNo))
{ {
...@@ -1156,12 +1155,6 @@ XLogSend(char *msgbuf, bool *caughtup) ...@@ -1156,12 +1155,6 @@ XLogSend(char *msgbuf, bool *caughtup)
startptr = sentPtr; startptr = sentPtr;
endptr = startptr; endptr = startptr;
XLByteAdvance(endptr, MAX_SEND_SIZE); XLByteAdvance(endptr, MAX_SEND_SIZE);
if (endptr.xlogid != startptr.xlogid)
{
/* Don't cross a logfile boundary within one message */
Assert(endptr.xlogid == startptr.xlogid + 1);
endptr.xrecoff = 0;
}
/* if we went beyond SendRqstPtr, back off */ /* if we went beyond SendRqstPtr, back off */
if (XLByteLE(SendRqstPtr, endptr)) if (XLByteLE(SendRqstPtr, endptr))
...@@ -1172,14 +1165,11 @@ XLogSend(char *msgbuf, bool *caughtup) ...@@ -1172,14 +1165,11 @@ XLogSend(char *msgbuf, bool *caughtup)
else else
{ {
/* round down to page boundary. */ /* round down to page boundary. */
endptr.xrecoff -= (endptr.xrecoff % XLOG_BLCKSZ); endptr -= (endptr % XLOG_BLCKSZ);
*caughtup = false; *caughtup = false;
} }
if (endptr.xrecoff == 0) nbytes = endptr - startptr;
nbytes = 0x100000000L - (uint64) startptr.xrecoff;
else
nbytes = endptr.xrecoff - startptr.xrecoff;
Assert(nbytes <= MAX_SEND_SIZE); Assert(nbytes <= MAX_SEND_SIZE);
/* /*
...@@ -1223,7 +1213,7 @@ XLogSend(char *msgbuf, bool *caughtup) ...@@ -1223,7 +1213,7 @@ XLogSend(char *msgbuf, bool *caughtup)
char activitymsg[50]; char activitymsg[50];
snprintf(activitymsg, sizeof(activitymsg), "streaming %X/%X", snprintf(activitymsg, sizeof(activitymsg), "streaming %X/%X",
sentPtr.xlogid, sentPtr.xrecoff); (uint32) (sentPtr >> 32), (uint32) sentPtr);
set_ps_display(activitymsg, false); set_ps_display(activitymsg, false);
} }
...@@ -1565,25 +1555,25 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS) ...@@ -1565,25 +1555,25 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
values[1] = CStringGetTextDatum(WalSndGetStateString(state)); values[1] = CStringGetTextDatum(WalSndGetStateString(state));
snprintf(location, sizeof(location), "%X/%X", snprintf(location, sizeof(location), "%X/%X",
sentPtr.xlogid, sentPtr.xrecoff); (uint32) (sentPtr >> 32), (uint32) sentPtr);
values[2] = CStringGetTextDatum(location); values[2] = CStringGetTextDatum(location);
if (write.xlogid == 0 && write.xrecoff == 0) if (write == 0)
nulls[3] = true; nulls[3] = true;
snprintf(location, sizeof(location), "%X/%X", snprintf(location, sizeof(location), "%X/%X",
write.xlogid, write.xrecoff); (uint32) (write >> 32), (uint32) write);
values[3] = CStringGetTextDatum(location); values[3] = CStringGetTextDatum(location);
if (flush.xlogid == 0 && flush.xrecoff == 0) if (flush == 0)
nulls[4] = true; nulls[4] = true;
snprintf(location, sizeof(location), "%X/%X", snprintf(location, sizeof(location), "%X/%X",
flush.xlogid, flush.xrecoff); (uint32) (flush >> 32), (uint32) flush);
values[4] = CStringGetTextDatum(location); values[4] = CStringGetTextDatum(location);
if (apply.xlogid == 0 && apply.xrecoff == 0) if (apply == 0)
nulls[5] = true; nulls[5] = true;
snprintf(location, sizeof(location), "%X/%X", snprintf(location, sizeof(location), "%X/%X",
apply.xlogid, apply.xrecoff); (uint32) (apply >> 32), (uint32) apply);
values[5] = CStringGetTextDatum(location); values[5] = CStringGetTextDatum(location);
values[6] = Int32GetDatum(sync_priority[i]); values[6] = Int32GetDatum(sync_priority[i]);
......
...@@ -936,7 +936,7 @@ LogCurrentRunningXacts(RunningTransactions CurrRunningXacts) ...@@ -936,7 +936,7 @@ LogCurrentRunningXacts(RunningTransactions CurrRunningXacts)
elog(trace_recovery(DEBUG2), elog(trace_recovery(DEBUG2),
"snapshot of %u running transactions overflowed (lsn %X/%X oldest xid %u latest complete %u next xid %u)", "snapshot of %u running transactions overflowed (lsn %X/%X oldest xid %u latest complete %u next xid %u)",
CurrRunningXacts->xcnt, CurrRunningXacts->xcnt,
recptr.xlogid, recptr.xrecoff, (uint32) (recptr >> 32), (uint32) recptr,
CurrRunningXacts->oldestRunningXid, CurrRunningXacts->oldestRunningXid,
CurrRunningXacts->latestCompletedXid, CurrRunningXacts->latestCompletedXid,
CurrRunningXacts->nextXid); CurrRunningXacts->nextXid);
...@@ -944,7 +944,7 @@ LogCurrentRunningXacts(RunningTransactions CurrRunningXacts) ...@@ -944,7 +944,7 @@ LogCurrentRunningXacts(RunningTransactions CurrRunningXacts)
elog(trace_recovery(DEBUG2), elog(trace_recovery(DEBUG2),
"snapshot of %u running transaction ids (lsn %X/%X oldest xid %u latest complete %u next xid %u)", "snapshot of %u running transaction ids (lsn %X/%X oldest xid %u latest complete %u next xid %u)",
CurrRunningXacts->xcnt, CurrRunningXacts->xcnt,
recptr.xlogid, recptr.xrecoff, (uint32) (recptr >> 32), (uint32) recptr,
CurrRunningXacts->oldestRunningXid, CurrRunningXacts->oldestRunningXid,
CurrRunningXacts->latestCompletedXid, CurrRunningXacts->latestCompletedXid,
CurrRunningXacts->nextXid); CurrRunningXacts->nextXid);
......
...@@ -376,8 +376,7 @@ InitProcess(void) ...@@ -376,8 +376,7 @@ InitProcess(void)
MyProc->recoveryConflictPending = false; MyProc->recoveryConflictPending = false;
/* Initialize fields for sync rep */ /* Initialize fields for sync rep */
MyProc->waitLSN.xlogid = 0; MyProc->waitLSN = 0;
MyProc->waitLSN.xrecoff = 0;
MyProc->syncRepState = SYNC_REP_NOT_WAITING; MyProc->syncRepState = SYNC_REP_NOT_WAITING;
SHMQueueElemInit(&(MyProc->syncRepLinks)); SHMQueueElemInit(&(MyProc->syncRepLinks));
......
...@@ -162,6 +162,8 @@ reached_end_position(XLogRecPtr segendpos, uint32 timeline, bool segment_finishe ...@@ -162,6 +162,8 @@ reached_end_position(XLogRecPtr segendpos, uint32 timeline, bool segment_finishe
if (r == 1) if (r == 1)
{ {
char xlogend[64]; char xlogend[64];
uint32 hi,
lo;
MemSet(xlogend, 0, sizeof(xlogend)); MemSet(xlogend, 0, sizeof(xlogend));
r = read(bgpipe[0], xlogend, sizeof(xlogend)); r = read(bgpipe[0], xlogend, sizeof(xlogend));
...@@ -172,12 +174,13 @@ reached_end_position(XLogRecPtr segendpos, uint32 timeline, bool segment_finishe ...@@ -172,12 +174,13 @@ reached_end_position(XLogRecPtr segendpos, uint32 timeline, bool segment_finishe
exit(1); exit(1);
} }
if (sscanf(xlogend, "%X/%X", &xlogendptr.xlogid, &xlogendptr.xrecoff) != 2) if (sscanf(xlogend, "%X/%X", &hi, &lo) != 2)
{ {
fprintf(stderr, _("%s: could not parse xlog end position \"%s\"\n"), fprintf(stderr, _("%s: could not parse xlog end position \"%s\"\n"),
progname, xlogend); progname, xlogend);
exit(1); exit(1);
} }
xlogendptr = ((uint64) hi) << 32 | lo;
has_xlogendptr = 1; has_xlogendptr = 1;
/* /*
...@@ -207,9 +210,7 @@ reached_end_position(XLogRecPtr segendpos, uint32 timeline, bool segment_finishe ...@@ -207,9 +210,7 @@ reached_end_position(XLogRecPtr segendpos, uint32 timeline, bool segment_finishe
* At this point we have an end pointer, so compare it to the current * At this point we have an end pointer, so compare it to the current
* position to figure out if it's time to stop. * position to figure out if it's time to stop.
*/ */
if (segendpos.xlogid > xlogendptr.xlogid || if (segendpos >= xlogendptr)
(segendpos.xlogid == xlogendptr.xlogid &&
segendpos.xrecoff >= xlogendptr.xrecoff))
return true; return true;
/* /*
...@@ -255,20 +256,23 @@ static void ...@@ -255,20 +256,23 @@ static void
StartLogStreamer(char *startpos, uint32 timeline, char *sysidentifier) StartLogStreamer(char *startpos, uint32 timeline, char *sysidentifier)
{ {
logstreamer_param *param; logstreamer_param *param;
uint32 hi,
lo;
param = xmalloc0(sizeof(logstreamer_param)); param = xmalloc0(sizeof(logstreamer_param));
param->timeline = timeline; param->timeline = timeline;
param->sysidentifier = sysidentifier; param->sysidentifier = sysidentifier;
/* Convert the starting position */ /* Convert the starting position */
if (sscanf(startpos, "%X/%X", &param->startptr.xlogid, &param->startptr.xrecoff) != 2) if (sscanf(startpos, "%X/%X", &hi, &lo) != 2)
{ {
fprintf(stderr, _("%s: invalid format of xlog location: %s\n"), fprintf(stderr, _("%s: invalid format of xlog location: %s\n"),
progname, startpos); progname, startpos);
disconnect_and_exit(1); disconnect_and_exit(1);
} }
param->startptr = ((uint64) hi) << 32 | lo;
/* Round off to even segment position */ /* Round off to even segment position */
param->startptr.xrecoff -= param->startptr.xrecoff % XLOG_SEG_SIZE; param->startptr -= param->startptr % XLOG_SEG_SIZE;
#ifndef WIN32 #ifndef WIN32
/* Create our background pipe */ /* Create our background pipe */
......
...@@ -77,7 +77,9 @@ stop_streaming(XLogRecPtr segendpos, uint32 timeline, bool segment_finished) ...@@ -77,7 +77,9 @@ stop_streaming(XLogRecPtr segendpos, uint32 timeline, bool segment_finished)
{ {
if (verbose && segment_finished) if (verbose && segment_finished)
fprintf(stderr, _("%s: finished segment at %X/%X (timeline %u)\n"), fprintf(stderr, _("%s: finished segment at %X/%X (timeline %u)\n"),
progname, segendpos.xlogid, segendpos.xrecoff, timeline); progname,
(uint32) (segendpos >> 32), (uint32) segendpos,
timeline);
if (time_to_abort) if (time_to_abort)
{ {
...@@ -212,6 +214,8 @@ StreamLog(void) ...@@ -212,6 +214,8 @@ StreamLog(void)
PGresult *res; PGresult *res;
uint32 timeline; uint32 timeline;
XLogRecPtr startpos; XLogRecPtr startpos;
uint32 hi,
lo;
/* /*
* Connect in replication mode to the server * Connect in replication mode to the server
...@@ -239,12 +243,13 @@ StreamLog(void) ...@@ -239,12 +243,13 @@ StreamLog(void)
disconnect_and_exit(1); disconnect_and_exit(1);
} }
timeline = atoi(PQgetvalue(res, 0, 1)); timeline = atoi(PQgetvalue(res, 0, 1));
if (sscanf(PQgetvalue(res, 0, 2), "%X/%X", &startpos.xlogid, &startpos.xrecoff) != 2) if (sscanf(PQgetvalue(res, 0, 2), "%X/%X", &hi, &lo) != 2)
{ {
fprintf(stderr, _("%s: could not parse log start position from value \"%s\"\n"), fprintf(stderr, _("%s: could not parse log start position from value \"%s\"\n"),
progname, PQgetvalue(res, 0, 2)); progname, PQgetvalue(res, 0, 2));
disconnect_and_exit(1); disconnect_and_exit(1);
} }
startpos = ((uint64) hi) << 32 | lo;
PQclear(res); PQclear(res);
/* /*
...@@ -255,14 +260,16 @@ StreamLog(void) ...@@ -255,14 +260,16 @@ StreamLog(void)
/* /*
* Always start streaming at the beginning of a segment * Always start streaming at the beginning of a segment
*/ */
startpos.xrecoff -= startpos.xrecoff % XLOG_SEG_SIZE; startpos -= startpos % XLOG_SEG_SIZE;
/* /*
* Start the replication * Start the replication
*/ */
if (verbose) if (verbose)
fprintf(stderr, _("%s: starting log streaming at %X/%X (timeline %u)\n"), fprintf(stderr, _("%s: starting log streaming at %X/%X (timeline %u)\n"),
progname, startpos.xlogid, startpos.xrecoff, timeline); progname,
(uint32) (startpos >> 32), (uint32) startpos,
timeline);
ReceiveXlogStream(conn, startpos, timeline, NULL, basedir, ReceiveXlogStream(conn, startpos, timeline, NULL, basedir,
stop_streaming, stop_streaming,
......
...@@ -38,8 +38,6 @@ ...@@ -38,8 +38,6 @@
#define STREAMING_HEADER_SIZE (1+sizeof(WalDataMessageHeader)) #define STREAMING_HEADER_SIZE (1+sizeof(WalDataMessageHeader))
#define STREAMING_KEEPALIVE_SIZE (1+sizeof(PrimaryKeepaliveMessage)) #define STREAMING_KEEPALIVE_SIZE (1+sizeof(PrimaryKeepaliveMessage))
const XLogRecPtr InvalidXLogRecPtr = {0, 0};
/* /*
* Open a new WAL file in the specified directory. Store the name * Open a new WAL file in the specified directory. Store the name
* (not including the full directory) in namebuf. Assumes there is * (not including the full directory) in namebuf. Assumes there is
...@@ -310,7 +308,8 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi ...@@ -310,7 +308,8 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
} }
/* Initiate the replication stream at specified location */ /* Initiate the replication stream at specified location */
snprintf(query, sizeof(query), "START_REPLICATION %X/%X", startpos.xlogid, startpos.xrecoff); snprintf(query, sizeof(query), "START_REPLICATION %X/%X",
(uint32) (startpos >> 32), (uint32) startpos);
res = PQexec(conn, query); res = PQexec(conn, query);
if (PQresultStatus(res) != PGRES_COPY_BOTH) if (PQresultStatus(res) != PGRES_COPY_BOTH)
{ {
...@@ -471,7 +470,7 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi ...@@ -471,7 +470,7 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
/* Extract WAL location for this block */ /* Extract WAL location for this block */
memcpy(&blockpos, copybuf + 1, 8); memcpy(&blockpos, copybuf + 1, 8);
xlogoff = blockpos.xrecoff % XLOG_SEG_SIZE; xlogoff = blockpos % XLOG_SEG_SIZE;
/* /*
* Verify that the initial location in the stream matches where we * Verify that the initial location in the stream matches where we
...@@ -543,7 +542,7 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi ...@@ -543,7 +542,7 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
xlogoff += bytes_to_write; xlogoff += bytes_to_write;
/* Did we reach the end of a WAL segment? */ /* Did we reach the end of a WAL segment? */
if (blockpos.xrecoff % XLOG_SEG_SIZE == 0) if (blockpos % XLOG_SEG_SIZE == 0)
{ {
if (!close_walfile(walfile, basedir, current_walfile_name, false)) if (!close_walfile(walfile, basedir, current_walfile_name, false))
/* Error message written in close_walfile() */ /* Error message written in close_walfile() */
......
...@@ -193,14 +193,14 @@ main(int argc, char *argv[]) ...@@ -193,14 +193,14 @@ main(int argc, char *argv[])
printf(_("pg_control last modified: %s\n"), printf(_("pg_control last modified: %s\n"),
pgctime_str); pgctime_str);
printf(_("Latest checkpoint location: %X/%X\n"), printf(_("Latest checkpoint location: %X/%X\n"),
ControlFile.checkPoint.xlogid, (uint32) (ControlFile.checkPoint >> 32),
ControlFile.checkPoint.xrecoff); (uint32) ControlFile.checkPoint);
printf(_("Prior checkpoint location: %X/%X\n"), printf(_("Prior checkpoint location: %X/%X\n"),
ControlFile.prevCheckPoint.xlogid, (uint32) (ControlFile.prevCheckPoint >> 32),
ControlFile.prevCheckPoint.xrecoff); (uint32) ControlFile.prevCheckPoint);
printf(_("Latest checkpoint's REDO location: %X/%X\n"), printf(_("Latest checkpoint's REDO location: %X/%X\n"),
ControlFile.checkPointCopy.redo.xlogid, (uint32) (ControlFile.checkPointCopy.redo >> 32),
ControlFile.checkPointCopy.redo.xrecoff); (uint32) ControlFile.checkPointCopy.redo);
printf(_("Latest checkpoint's TimeLineID: %u\n"), printf(_("Latest checkpoint's TimeLineID: %u\n"),
ControlFile.checkPointCopy.ThisTimeLineID); ControlFile.checkPointCopy.ThisTimeLineID);
printf(_("Latest checkpoint's full_page_writes: %s\n"), printf(_("Latest checkpoint's full_page_writes: %s\n"),
...@@ -223,14 +223,14 @@ main(int argc, char *argv[]) ...@@ -223,14 +223,14 @@ main(int argc, char *argv[])
printf(_("Time of latest checkpoint: %s\n"), printf(_("Time of latest checkpoint: %s\n"),
ckpttime_str); ckpttime_str);
printf(_("Minimum recovery ending location: %X/%X\n"), printf(_("Minimum recovery ending location: %X/%X\n"),
ControlFile.minRecoveryPoint.xlogid, (uint32) (ControlFile.minRecoveryPoint >> 32),
ControlFile.minRecoveryPoint.xrecoff); (uint32) ControlFile.minRecoveryPoint);
printf(_("Backup start location: %X/%X\n"), printf(_("Backup start location: %X/%X\n"),
ControlFile.backupStartPoint.xlogid, (uint32) (ControlFile.backupStartPoint >> 32),
ControlFile.backupStartPoint.xrecoff); (uint32) ControlFile.backupStartPoint);
printf(_("Backup end location: %X/%X\n"), printf(_("Backup end location: %X/%X\n"),
ControlFile.backupEndPoint.xlogid, (uint32) (ControlFile.backupEndPoint >> 32),
ControlFile.backupEndPoint.xrecoff); (uint32) ControlFile.backupEndPoint);
printf(_("End-of-backup record required: %s\n"), printf(_("End-of-backup record required: %s\n"),
ControlFile.backupEndRequired ? _("yes") : _("no")); ControlFile.backupEndRequired ? _("yes") : _("no"));
printf(_("Current wal_level setting: %s\n"), printf(_("Current wal_level setting: %s\n"),
......
...@@ -463,8 +463,7 @@ GuessControlValues(void) ...@@ -463,8 +463,7 @@ GuessControlValues(void)
ControlFile.system_identifier = sysidentifier; ControlFile.system_identifier = sysidentifier;
ControlFile.checkPointCopy.redo.xlogid = 0; ControlFile.checkPointCopy.redo = SizeOfXLogLongPHD;
ControlFile.checkPointCopy.redo.xrecoff = SizeOfXLogLongPHD;
ControlFile.checkPointCopy.ThisTimeLineID = 1; ControlFile.checkPointCopy.ThisTimeLineID = 1;
ControlFile.checkPointCopy.fullPageWrites = false; ControlFile.checkPointCopy.fullPageWrites = false;
ControlFile.checkPointCopy.nextXidEpoch = 0; ControlFile.checkPointCopy.nextXidEpoch = 0;
...@@ -611,14 +610,10 @@ RewriteControlFile(void) ...@@ -611,14 +610,10 @@ RewriteControlFile(void)
ControlFile.state = DB_SHUTDOWNED; ControlFile.state = DB_SHUTDOWNED;
ControlFile.time = (pg_time_t) time(NULL); ControlFile.time = (pg_time_t) time(NULL);
ControlFile.checkPoint = ControlFile.checkPointCopy.redo; ControlFile.checkPoint = ControlFile.checkPointCopy.redo;
ControlFile.prevCheckPoint.xlogid = 0; ControlFile.prevCheckPoint = 0;
ControlFile.prevCheckPoint.xrecoff = 0; ControlFile.minRecoveryPoint = 0;
ControlFile.minRecoveryPoint.xlogid = 0; ControlFile.backupStartPoint = 0;
ControlFile.minRecoveryPoint.xrecoff = 0; ControlFile.backupEndPoint = 0;
ControlFile.backupStartPoint.xlogid = 0;
ControlFile.backupStartPoint.xrecoff = 0;
ControlFile.backupEndPoint.xlogid = 0;
ControlFile.backupEndPoint.xrecoff = 0;
ControlFile.backupEndRequired = false; ControlFile.backupEndRequired = false;
/* /*
...@@ -714,8 +709,7 @@ FindEndOfXLOG(void) ...@@ -714,8 +709,7 @@ FindEndOfXLOG(void)
* numbering according to the old xlog seg size. * numbering according to the old xlog seg size.
*/ */
segs_per_xlogid = (0x100000000L / ControlFile.xlog_seg_size); segs_per_xlogid = (0x100000000L / ControlFile.xlog_seg_size);
newXlogSegNo = ((uint64) ControlFile.checkPointCopy.redo.xlogid) * segs_per_xlogid newXlogSegNo = ControlFile.checkPointCopy.redo / ControlFile.xlog_seg_size;
+ (ControlFile.checkPointCopy.redo.xrecoff / ControlFile.xlog_seg_size);
/* /*
* Scan the pg_xlog directory to find existing WAL segment files. We * Scan the pg_xlog directory to find existing WAL segment files. We
...@@ -919,10 +913,7 @@ WriteEmptyXLOG(void) ...@@ -919,10 +913,7 @@ WriteEmptyXLOG(void)
page->xlp_magic = XLOG_PAGE_MAGIC; page->xlp_magic = XLOG_PAGE_MAGIC;
page->xlp_info = XLP_LONG_HEADER; page->xlp_info = XLP_LONG_HEADER;
page->xlp_tli = ControlFile.checkPointCopy.ThisTimeLineID; page->xlp_tli = ControlFile.checkPointCopy.ThisTimeLineID;
page->xlp_pageaddr.xlogid = page->xlp_pageaddr = ControlFile.checkPointCopy.redo - SizeOfXLogLongPHD;
ControlFile.checkPointCopy.redo.xlogid;
page->xlp_pageaddr.xrecoff =
ControlFile.checkPointCopy.redo.xrecoff - SizeOfXLogLongPHD;
longpage = (XLogLongPageHeader) page; longpage = (XLogLongPageHeader) page;
longpage->xlp_sysid = ControlFile.system_identifier; longpage->xlp_sysid = ControlFile.system_identifier;
longpage->xlp_seg_size = XLogSegSize; longpage->xlp_seg_size = XLogSegSize;
...@@ -930,8 +921,7 @@ WriteEmptyXLOG(void) ...@@ -930,8 +921,7 @@ WriteEmptyXLOG(void)
/* Insert the initial checkpoint record */ /* Insert the initial checkpoint record */
record = (XLogRecord *) ((char *) page + SizeOfXLogLongPHD); record = (XLogRecord *) ((char *) page + SizeOfXLogLongPHD);
record->xl_prev.xlogid = 0; record->xl_prev = 0;
record->xl_prev.xrecoff = 0;
record->xl_xid = InvalidTransactionId; record->xl_xid = InvalidTransactionId;
record->xl_tot_len = SizeOfXLogRecord + sizeof(CheckPoint); record->xl_tot_len = SizeOfXLogRecord + sizeof(CheckPoint);
record->xl_len = sizeof(CheckPoint); record->xl_len = sizeof(CheckPoint);
......
...@@ -139,10 +139,6 @@ extern bool TransactionStartedDuringRecovery(void); ...@@ -139,10 +139,6 @@ extern bool TransactionStartedDuringRecovery(void);
/* in transam/varsup.c */ /* in transam/varsup.c */
extern PGDLLIMPORT VariableCache ShmemVariableCache; extern PGDLLIMPORT VariableCache ShmemVariableCache;
/* in transam/transam.c */
extern const XLogRecPtr InvalidXLogRecPtr;
/* /*
* prototypes for functions in transam/transam.c * prototypes for functions in transam/transam.c
*/ */
......
...@@ -51,7 +51,7 @@ typedef struct BkpBlock ...@@ -51,7 +51,7 @@ typedef struct BkpBlock
/* /*
* Each page of XLOG file has a header like this: * Each page of XLOG file has a header like this:
*/ */
#define XLOG_PAGE_MAGIC 0xD074 /* can be used as WAL version indicator */ #define XLOG_PAGE_MAGIC 0xD075 /* can be used as WAL version indicator */
typedef struct XLogPageHeaderData typedef struct XLogPageHeaderData
{ {
...@@ -113,10 +113,7 @@ typedef XLogLongPageHeaderData *XLogLongPageHeader; ...@@ -113,10 +113,7 @@ typedef XLogLongPageHeaderData *XLogLongPageHeader;
#define XLogSegmentsPerXLogId (0x100000000L / XLOG_SEG_SIZE) #define XLogSegmentsPerXLogId (0x100000000L / XLOG_SEG_SIZE)
#define XLogSegNoOffsetToRecPtr(segno, offset, dest) \ #define XLogSegNoOffsetToRecPtr(segno, offset, dest) \
do { \ (dest) = (segno) * XLOG_SEG_SIZE + (offset)
(dest).xlogid = (segno) / XLogSegmentsPerXLogId; \
(dest).xrecoff = ((segno) % XLogSegmentsPerXLogId) * XLOG_SEG_SIZE + (offset); \
} while (0)
/* /*
* Macros for manipulating XLOG pointers * Macros for manipulating XLOG pointers
...@@ -125,8 +122,8 @@ typedef XLogLongPageHeaderData *XLogLongPageHeader; ...@@ -125,8 +122,8 @@ typedef XLogLongPageHeaderData *XLogLongPageHeader;
/* Align a record pointer to next page */ /* Align a record pointer to next page */
#define NextLogPage(recptr) \ #define NextLogPage(recptr) \
do { \ do { \
if ((recptr).xrecoff % XLOG_BLCKSZ != 0) \ if ((recptr) % XLOG_BLCKSZ != 0) \
XLByteAdvance(recptr, (XLOG_BLCKSZ - (recptr).xrecoff % XLOG_BLCKSZ)); \ XLByteAdvance(recptr, (XLOG_BLCKSZ - (recptr) % XLOG_BLCKSZ)); \
} while (0) } while (0)
/* /*
...@@ -135,14 +132,13 @@ typedef XLogLongPageHeaderData *XLogLongPageHeader; ...@@ -135,14 +132,13 @@ typedef XLogLongPageHeaderData *XLogLongPageHeader;
* For XLByteToSeg, do the computation at face value. For XLByteToPrevSeg, * For XLByteToSeg, do the computation at face value. For XLByteToPrevSeg,
* a boundary byte is taken to be in the previous segment. This is suitable * a boundary byte is taken to be in the previous segment. This is suitable
* for deciding which segment to write given a pointer to a record end, * for deciding which segment to write given a pointer to a record end,
* for example. (We can assume xrecoff is not zero, since no valid recptr * for example.
* can have that.)
*/ */
#define XLByteToSeg(xlrp, logSegNo) \ #define XLByteToSeg(xlrp, logSegNo) \
logSegNo = ((uint64) (xlrp).xlogid * XLogSegmentsPerXLogId) + (xlrp).xrecoff / XLogSegSize logSegNo = (xlrp) / XLogSegSize
#define XLByteToPrevSeg(xlrp, logSegNo) \ #define XLByteToPrevSeg(xlrp, logSegNo) \
logSegNo = ((uint64) (xlrp).xlogid * XLogSegmentsPerXLogId) + ((xlrp).xrecoff - 1) / XLogSegSize logSegNo = ((xlrp) - 1) / XLogSegSize
/* /*
* Is an XLogRecPtr within a particular XLOG segment? * Is an XLogRecPtr within a particular XLOG segment?
...@@ -151,20 +147,15 @@ typedef XLogLongPageHeaderData *XLogLongPageHeader; ...@@ -151,20 +147,15 @@ typedef XLogLongPageHeaderData *XLogLongPageHeader;
* a boundary byte is taken to be in the previous segment. * a boundary byte is taken to be in the previous segment.
*/ */
#define XLByteInSeg(xlrp, logSegNo) \ #define XLByteInSeg(xlrp, logSegNo) \
(((xlrp).xlogid) == (logSegNo) / XLogSegmentsPerXLogId && \ (((xlrp) / XLogSegSize) == (logSegNo))
((xlrp).xrecoff / XLogSegSize) == (logSegNo) % XLogSegmentsPerXLogId)
#define XLByteInPrevSeg(xlrp, logSegNo) \ #define XLByteInPrevSeg(xlrp, logSegNo) \
(((xlrp).xrecoff == 0) ? \ ((((xlrp) - 1) / XLogSegSize) == (logSegNo))
(((xlrp).xlogid - 1) == (logSegNo) / XLogSegmentsPerXLogId && \
((uint32) 0xffffffff) / XLogSegSize == (logSegNo) % XLogSegmentsPerXLogId) : \ /* Check if an XLogRecPtr value is in a plausible range */
((xlrp).xlogid) == (logSegNo) / XLogSegmentsPerXLogId && \ #define XRecOffIsValid(xlrp) \
(((xlrp).xrecoff - 1) / XLogSegSize) == (logSegNo) % XLogSegmentsPerXLogId) ((xlrp) % XLOG_BLCKSZ >= SizeOfXLogShortPHD && \
(XLOG_BLCKSZ - (xlrp) % XLOG_BLCKSZ) >= SizeOfXLogRecord)
/* Check if an xrecoff value is in a plausible range */
#define XRecOffIsValid(xrecoff) \
((xrecoff) % XLOG_BLCKSZ >= SizeOfXLogShortPHD && \
(XLOG_BLCKSZ - (xrecoff) % XLOG_BLCKSZ) >= SizeOfXLogRecord)
/* /*
* The XLog directory and control file (relative to $PGDATA) * The XLog directory and control file (relative to $PGDATA)
......
...@@ -17,55 +17,30 @@ ...@@ -17,55 +17,30 @@
/* /*
* Pointer to a location in the XLOG. These pointers are 64 bits wide, * Pointer to a location in the XLOG. These pointers are 64 bits wide,
* because we don't want them ever to overflow. * because we don't want them ever to overflow.
*
* NOTE: xrecoff == 0 is used to indicate an invalid pointer. This is OK
* because we use page headers in the XLOG, so no XLOG record can start
* right at the beginning of a file.
*
* NOTE: the "log file number" is somewhat misnamed, since the actual files
* making up the XLOG are much smaller than 4Gb. Each actual file is an
* XLogSegSize-byte "segment" of a logical log file having the indicated
* xlogid. The log file number and segment number together identify a
* physical XLOG file. Segment number and offset within the physical file
* are computed from xrecoff div and mod XLogSegSize.
*/ */
typedef struct XLogRecPtr typedef uint64 XLogRecPtr;
{
uint32 xlogid; /* log file #, 0 based */
uint32 xrecoff; /* byte offset of location in log file */
} XLogRecPtr;
#define XLogRecPtrIsInvalid(r) ((r).xrecoff == 0)
/*
* Zero is used indicate an invalid pointer. Bootstrap skips the first possible
* WAL segment, initializing the first WAL page at XLOG_SEG_SIZE, so no XLOG
* record can begin at zero.
*/
#define InvalidXLogRecPtr 0
#define XLogRecPtrIsInvalid(r) ((r) == InvalidXLogRecPtr)
/* /*
* Macros for comparing XLogRecPtrs * Macros for comparing XLogRecPtrs
*
* Beware of passing expressions with side-effects to these macros,
* since the arguments may be evaluated multiple times.
*/ */
#define XLByteLT(a, b) \ #define XLByteLT(a, b) ((a) < (b))
((a).xlogid < (b).xlogid || \ #define XLByteLE(a, b) ((a) <= (b))
((a).xlogid == (b).xlogid && (a).xrecoff < (b).xrecoff)) #define XLByteEQ(a, b) ((a) == (b))
#define XLByteLE(a, b) \
((a).xlogid < (b).xlogid || \
((a).xlogid == (b).xlogid && (a).xrecoff <= (b).xrecoff))
#define XLByteEQ(a, b) \
((a).xlogid == (b).xlogid && (a).xrecoff == (b).xrecoff)
/* /*
* Macro for advancing a record pointer by the specified number of bytes. * Macro for advancing a record pointer by the specified number of bytes.
*/ */
#define XLByteAdvance(recptr, nbytes) \ #define XLByteAdvance(recptr, nbytes) \
do { \ (recptr) += nbytes \
uint32 oldxrecoff = (recptr).xrecoff; \
(recptr).xrecoff += nbytes; \
if ((recptr).xrecoff < oldxrecoff) \
(recptr).xlogid += 1; /* xrecoff wrapped around */ \
} while (0)
/* /*
* XLogSegNo - physical log file sequence number. * XLogSegNo - physical log file sequence number.
......
...@@ -21,7 +21,7 @@ ...@@ -21,7 +21,7 @@
/* Version identifier for this pg_control format */ /* Version identifier for this pg_control format */
#define PG_CONTROL_VERSION 922 #define PG_CONTROL_VERSION 931
/* /*
* Body of CheckPoint XLOG records. This is declared here because we keep * Body of CheckPoint XLOG records. This is declared here because we keep
......
...@@ -119,10 +119,18 @@ typedef uint16 LocationIndex; ...@@ -119,10 +119,18 @@ typedef uint16 LocationIndex;
* On the high end, we can only support pages up to 32KB because lp_off/lp_len * On the high end, we can only support pages up to 32KB because lp_off/lp_len
* are 15 bits. * are 15 bits.
*/ */
/* for historical reasons, the LSN is stored as two 32-bit values. */
typedef struct
{
uint32 xlogid; /* high bits */
uint32 xrecoff; /* low bits */
} PageXLogRecPtr;
typedef struct PageHeaderData typedef struct PageHeaderData
{ {
/* XXX LSN is member of *any* block, not only page-organized ones */ /* XXX LSN is member of *any* block, not only page-organized ones */
XLogRecPtr pd_lsn; /* LSN: next byte after last byte of xlog PageXLogRecPtr pd_lsn; /* LSN: next byte after last byte of xlog
* record for last change to this page */ * record for last change to this page */
uint16 pd_tli; /* least significant bits of the TimeLineID uint16 pd_tli; /* least significant bits of the TimeLineID
* containing the LSN */ * containing the LSN */
...@@ -314,9 +322,10 @@ typedef PageHeaderData *PageHeader; ...@@ -314,9 +322,10 @@ typedef PageHeaderData *PageHeader;
* Additional macros for access to page headers * Additional macros for access to page headers
*/ */
#define PageGetLSN(page) \ #define PageGetLSN(page) \
(((PageHeader) (page))->pd_lsn) ((uint64) ((PageHeader) (page))->pd_lsn.xlogid << 32 | ((PageHeader) (page))->pd_lsn.xrecoff)
#define PageSetLSN(page, lsn) \ #define PageSetLSN(page, lsn) \
(((PageHeader) (page))->pd_lsn = (lsn)) (((PageHeader) (page))->pd_lsn.xlogid = (uint32) ((lsn) >> 32), \
((PageHeader) (page))->pd_lsn.xrecoff = (uint32) (lsn))
/* NOTE: only the 16 least significant bits are stored */ /* NOTE: only the 16 least significant bits are stored */
#define PageGetTLI(page) \ #define PageGetTLI(page) \
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment