Re: Timeline following for logical slots

Hi,


On 2016-03-14 20:10:58 -0300, Alvaro Herrera wrote:
> diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
> index fcb0872..7b60b8f 100644
> --- a/src/backend/access/transam/xlogreader.c
> +++ b/src/backend/access/transam/xlogreader.c
> @@ -10,9 +10,11 @@
>   *
>   * NOTES
>   *        See xlogreader.h for more notes on this facility.
> + *
> + *        This file is compiled as both front-end and backend code, so it
> + *        may not use ereport, server-defined static variables, etc.
>   *-------------------------------------------------------------------------
>   */
> -

Huh?

>  #include "postgres.h"
>  
>  #include "access/transam.h"
> @@ -116,6 +118,11 @@ XLogReaderAllocate(XLogPageReadCB pagereadfunc, void *private_data)
>          return NULL;
>      }
>  
> +#ifndef FRONTEND
> +    /* Will be loaded on first read */
> +    state->timelineHistory = NIL;
> +#endif
> +
>      return state;
>  }
>  
> @@ -135,6 +142,10 @@ XLogReaderFree(XLogReaderState *state)
>      pfree(state->errormsg_buf);
>      if (state->readRecordBuf)
>          pfree(state->readRecordBuf);
> +#ifndef FRONTEND
> +    if (state->timelineHistory)
> +        list_free_deep(state->timelineHistory);
> +#endif

Hm. So we don't support timelines following for frontend code, although
it'd be rather helpful for pg_xlogdump. And possibly pg_rewind.


>      pfree(state->readBuf);
>      pfree(state);
>  }
> @@ -208,10 +219,11 @@ XLogReadRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg)
>  
>      if (RecPtr == InvalidXLogRecPtr)
>      {
> +        /* No explicit start point; read the record after the one we just read */
>          RecPtr = state->EndRecPtr;
>  
>          if (state->ReadRecPtr == InvalidXLogRecPtr)
> -            randAccess = true;
> +            randAccess = true;    /* allow readPageTLI to go backwards */

randAccess is doing more than that, so I'm doubtful that comment is an
improvment.


>          /*
>           * RecPtr is pointing to end+1 of the previous WAL record.  If we're
> @@ -223,6 +235,8 @@ XLogReadRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg)
>      else
>      {
>          /*
> +         * Caller supplied a position to start at.
> +         *
>           * In this case, the passed-in record pointer should already be
>           * pointing to a valid record starting position.
>           */
> @@ -309,8 +323,10 @@ XLogReadRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg)
>          /* XXX: more validation should be done here */
>          if (total_len < SizeOfXLogRecord)
>          {
> -            report_invalid_record(state, "invalid record length at %X/%X",
> -                                  (uint32) (RecPtr >> 32), (uint32) RecPtr);
> +            report_invalid_record(state,
> +                        "invalid record length at %X/%X: wanted %lu, got %u",
> +                                  (uint32) (RecPtr >> 32), (uint32) RecPtr,
> +                                  SizeOfXLogRecord, total_len);
>              goto err;
>          }
>          gotheader = false;
> @@ -466,9 +482,7 @@ err:
>       * Invalidate the xlog page we've cached. We might read from a different
>       * source after failure.
>       */
> -    state->readSegNo = 0;
> -    state->readOff = 0;
> -    state->readLen = 0;
> +    XLogReaderInvalCache(state);

I don't think that "cache" is the right way to describe this.


>  #include <unistd.h>
>  
> -#include "miscadmin.h"
> -

spurious change imo.



>  /*
> - * TODO: This is duplicate code with pg_xlogdump, similar to walsender.c, but
> - * we currently don't have the infrastructure (elog!) to share it.
> + * Read 'count' bytes from WAL into 'buf', starting at location 'startptr'
> + * in timeline 'tli'.
> + *
> + * Will open, and keep open, one WAL segment stored in the static file
> + * descriptor 'sendFile'. This means if XLogRead is used once, there will
> + * always be one descriptor left open until the process ends, but never
> + * more than one.
> + *
> + * XXX This is very similar to pg_xlogdump's XLogDumpXLogRead and to XLogRead
> + * in walsender.c but for small differences (such as lack of elog() in
> + * frontend).  Probably these should be merged at some point.
>   */
>  static void
>  XLogRead(char *buf, TimeLineID tli, XLogRecPtr startptr, Size count)
> @@ -648,8 +657,12 @@ XLogRead(char *buf, TimeLineID tli, XLogRecPtr startptr, Size count)
>      XLogRecPtr    recptr;
>      Size        nbytes;
>  
> +    /*
> +     * Cached state across calls.
> +     */

One line?


>      static int    sendFile = -1;
>      static XLogSegNo sendSegNo = 0;
> +    static TimeLineID sendTLI = 0;
>      static uint32 sendOff = 0;
>  
>      p = buf;
> @@ -664,11 +677,12 @@ XLogRead(char *buf, TimeLineID tli, XLogRecPtr startptr, Size count)
>  
>          startoff = recptr % XLogSegSize;
>  
> -        if (sendFile < 0 || !XLByteInSeg(recptr, sendSegNo))
> +        /* Do we need to open a new xlog segment? */
> +        if (sendFile < 0 || !XLByteInSeg(recptr, sendSegNo) ||
> +            sendTLI != tli)
>          {

s/open a new/open a different/?  New imo has connotations that we don't
really want here.


>              char        path[MAXPGPATH];
>  
> -            /* Switch to another logfile segment */
>              if (sendFile >= 0)
>                  close(sendFile);

E.g. you could just have moved the above comment.


>          /* Need to seek in the file? */
>          if (sendOff != startoff)
>          {
>              if (lseek(sendFile, (off_t) startoff, SEEK_SET) < 0)
> -            {
> -                char        path[MAXPGPATH];
> -
> -                XLogFilePath(path, tli, sendSegNo);
> -
>                  ereport(ERROR,
>                          (errcode_for_file_access(),
>                    errmsg("could not seek in log segment %s to offset %u: %m",
> -                         path, startoff)));
> -            }
> +                         XLogFileNameP(tli, sendSegNo), startoff)));
>              sendOff = startoff;
>          }

Not a serious issue, more a general remark: I'm doubtful that going for
palloc in error situations is good practice. This will be allocated in
the current memory context; without access to the emergency error
reserves.


I'm also getting the feeling that the patch is bordering on doing some
relatively random cleanups mixed in with architectural changes. Makes
things a bit harder to review.


> +static void
> +XLogReadDetermineTimeline(XLogReaderState *state)
> +{
> +    /* Read the history on first time through */
> +    if (state->timelineHistory == NIL)
> +        state->timelineHistory = readTimeLineHistory(ThisTimeLineID);
> +
> +    /*
> +     * Are we reading the record immediately following the one we read last
> +     * time?  If not, then don't use the cached timeline info.
> +     */
> +    if (state->currRecPtr != state->EndRecPtr)
> +    {
> +        state->currTLI = 0;
> +        state->currTLIValidUntil = InvalidXLogRecPtr;
> +    }


Hm. So we grow essentially a second version of the last end position and
the randAccess stuff in XLogReadRecord().


> +    if (state->currTLI == 0)
> +    {
> +        /*
> +         * Something changed; work out what timeline this record is on. We
> +         * might read it from the segment on this TLI or, if the segment is
> +         * also contained by newer timelines, the copy from a newer TLI.
> +         */
> +        state->currTLI = tliOfPointInHistory(state->currRecPtr,
> +                                             state->timelineHistory);
> +
> +        /*
> +         * Look for the most recent timeline that's on the same xlog segment
> +         * as this record, since that's the only one we can assume is still
> +         * readable.
> +         */
> +        while (state->currTLI != ThisTimeLineID &&
> +               state->currTLIValidUntil == InvalidXLogRecPtr)
> +        {
> +            XLogRecPtr    tliSwitch;
> +            TimeLineID    nextTLI;
> +
> +            tliSwitch = tliSwitchPoint(state->currTLI, state->timelineHistory,
> +                                       &nextTLI);
> +
> +            /* round ValidUntil down to start of seg containing the switch */
> +            state->currTLIValidUntil =
> +                ((tliSwitch / XLogSegSize) * XLogSegSize);
> +
> +            if (state->currRecPtr >= state->currTLIValidUntil)
> +            {
> +                /*
> +                 * The new currTLI ends on this WAL segment so check the next
> +                 * TLI to see if it's the last one on the segment.
> +                 *
> +                 * If that's the current TLI we'll stop searching.

I don't really understand how we're stopping searching here?

> +                 */
> +                state->currTLI = nextTLI;
> +                state->currTLIValidUntil = InvalidXLogRecPtr;
> +            }
> +        }
> +}


XLogReadDetermineTimeline() doesn't sit quite right with me, I do wonder
whether there's not a simpler way to write this.


> +/*
> + * XLogPageReadCB callback for reading local xlog files
>   *
>   * Public because it would likely be very helpful for someone writing another
>   * output method outside walsender, e.g. in a bgworker.
>   *
> - * TODO: The walsender has it's own version of this, but it relies on the
> + * TODO: The walsender has its own version of this, but it relies on the
>   * walsender's latch being set whenever WAL is flushed. No such infrastructure
>   * exists for normal backends, so we have to do a check/sleep/repeat style of
>   * loop for now.
> @@ -754,46 +897,88 @@ int
>  read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
>      int reqLen, XLogRecPtr targetRecPtr, char *cur_page, TimeLineID *pageTLI)
>  {
> -    XLogRecPtr    flushptr,
> +    XLogRecPtr    read_upto,
>                  loc;
>      int            count;
>  
>      loc = targetPagePtr + reqLen;
> +
> +    /* Make sure enough xlog is available... */
>      while (1)
>      {
>          /*
> -         * TODO: we're going to have to do something more intelligent about
> -         * timelines on standbys. Use readTimeLineHistory() and
> -         * tliOfPointInHistory() to get the proper LSN? For now we'll catch
> -         * that case earlier, but the code and TODO is left in here for when
> -         * that changes.
> +         * Check which timeline to get the record from.
> +         *
> +         * We have to do it each time through the loop because if we're in
> +         * recovery as a cascading standby, the current timeline might've
> +         * become historical.
>           */
> -        if (!RecoveryInProgress())
> +        XLogReadDetermineTimeline(state);
> +
> +        if (state->currTLI == ThisTimeLineID)
>          {
> -            *pageTLI = ThisTimeLineID;
> -            flushptr = GetFlushRecPtr();
> +            /*
> +             * We're reading from the current timeline so we might have to
> +             * wait for the desired record to be generated (or, for a standby,
> +             * received & replayed)
> +             */
> +            if (!RecoveryInProgress())
> +            {
> +                *pageTLI = ThisTimeLineID;
> +                read_upto = GetFlushRecPtr();
> +            }
> +            else
> +                read_upto = GetXLogReplayRecPtr(pageTLI);
> +
> +            if (loc <= read_upto)
> +                break;
> +
> +            CHECK_FOR_INTERRUPTS();
> +            pg_usleep(1000L);
>          }
>          else
> -            flushptr = GetXLogReplayRecPtr(pageTLI);
> +        {
> +            /*
> +             * We're on a historical timeline, so limit reading to the switch
> +             * point where we moved to the next timeline.
> +             */
> +            read_upto = state->currTLIValidUntil;

Hm. Is it ok to not check GetFlushRecPtr/GetXLogReplayRecPtr() here? If
so, how come?

> -        if (loc <= flushptr)
> +            /*
> +             * Setting pageTLI to our wanted record's TLI is slightly wrong;
> +             * the page might begin on an older timeline if it contains a
> +             * timeline switch, since its xlog segment will have been copied
> +             * from the prior timeline. This is pretty harmless though, as
> +             * nothing cares so long as the timeline doesn't go backwards.  We
> +             * should read the page header instead; FIXME someday.
> +             */
> +            *pageTLI = state->currTLI;
> +
> +            /* No need to wait on a historical timeline */
>              break;
> -
> -        CHECK_FOR_INTERRUPTS();
> -        pg_usleep(1000L);
> +        }
>      }
>  
> -    /* more than one block available */
> -    if (targetPagePtr + XLOG_BLCKSZ <= flushptr)
> +    if (targetPagePtr + XLOG_BLCKSZ <= read_upto)
> +    {
> +        /*
> +         * more than one block available; read only that block, have caller
> +         * come back if they need more.
> +         */
>          count = XLOG_BLCKSZ;
> -    /* not enough data there */
> -    else if (targetPagePtr + reqLen > flushptr)
> +    }
> +    else if (targetPagePtr + reqLen > read_upto)
> +    {
> +        /* not enough data there */
>          return -1;
> -    /* part of the page available */
> +    }
>      else
> -        count = flushptr - targetPagePtr;
> +    {
> +        /* enough bytes available to satisfy the request */
> +        count = read_upto - targetPagePtr;
> +    }
>  
> -    XLogRead(cur_page, *pageTLI, targetPagePtr, XLOG_BLCKSZ);
> +    XLogRead(cur_page, *pageTLI, targetPagePtr, count);

When are we reading less than a page? That should afaik never be required.


> +        /*
> +         * We start reading xlog from the restart lsn, even though in
> +         * CreateDecodingContext we set the snapshot builder up using the
> +         * slot's candidate_restart_lsn. This means we might read xlog we
> +         * don't actually decode rows from, but the snapshot builder might
> +         * need it to get to a consistent point. The point we start returning
> +         * data to *users* at is the candidate restart lsn from the decoding
> +         * context.
> +         */

Uh? Where are we using candidate_restart_lsn that way? I seriously doubt
it is - candidate_restart_lsn is about a potential future restart_lsn,
which we can set once we get reception confirmation from the client.


> @@ -299,6 +312,18 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
>              CHECK_FOR_INTERRUPTS();
>          }
>  
> +        /* Make sure timeline lookups use the start of the next record */
> +        startptr = ctx->reader->EndRecPtr;

Huh? startptr isn't used after this, so I'm not sure what this even
means?

> +        /*
> +         * The XLogReader will read a page past the valid end of WAL because
> +         * it doesn't know about timelines. When we switch timelines and ask
> +         * it for the first page on the new timeline it will think it has it
> +         * cached, but it'll have the old partial page and say it can't find
> +         * the next record. So flush the cache.
> +         */
> +        XLogReaderInvalCache(ctx->reader);
> +

dito.


> diff --git a/src/test/modules/decoding_failover/decoding_failover.c
b/src/test/modules/decoding_failover/decoding_failover.c
> new file mode 100644
> index 0000000..669e6c4
> --- /dev/null
> +++ b/src/test/modules/decoding_failover/decoding_failover.c

> +
> +/*
> + * Create a new logical slot, with invalid LSN and xid, directly. This does not
> + * use the snapshot builder or logical decoding machinery. It's only intended
> + * for creating a slot on a replica that mirrors the state of a slot on an
> + * upstream master.
> + *
> + * You should immediately decoding_failover_advance_logical_slot(...) it
> + * after creation.
> + */

Uh. I doubt we want this, even if it's formally located in
src/test/modules.  These comments make it appear not to be only intended
for that, and I have serious doubts about the validity of the concept as
is.


This seems to need some more polishing.


Greetings,

Andres Freund