Re: Reviving the "Stopping logical replication protocol" patch fromVladimir Gordichuk

Hi,

On 2018-12-03 06:38:43 -0500, Dave Cramer wrote:
> From 4d023cfc1fed0b5852b4da1aad6a32549b03ce26 Mon Sep 17 00:00:00 2001
> From: Dave Cramer <davecramer@gmail.com>
> Date: Fri, 30 Nov 2018 18:23:49 -0500
> Subject: [PATCH 1/5] Respect client initiated CopyDone in walsender
>
> ---
>  src/backend/replication/walsender.c | 36 ++++++++++++++++++++++++++++++------
>  1 file changed, 30 insertions(+), 6 deletions(-)
>
> diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
> index 46edb52..93f2648 100644
> --- a/src/backend/replication/walsender.c
> +++ b/src/backend/replication/walsender.c
> @@ -770,6 +770,14 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req
>       sendTimeLineValidUpto = state->currTLIValidUntil;
>       sendTimeLineNextTLI = state->nextTLI;
> 
> +     /*
> +     * If the client sent CopyDone while we were waiting,
> +     * bail out so we can wind up the decoding session.
> +     */
> +     if (streamingDoneSending)
> +             return -1;
> +
> +      /* more than one block available */
>       /* make sure we have enough WAL available */
>       flushptr = WalSndWaitForWal(targetPagePtr + reqLen);
> 
> @@ -1341,8 +1349,12 @@ WalSndWaitForWal(XLogRecPtr loc)
>                * It's important to do this check after the recomputation of
>                * RecentFlushPtr, so we can send all remaining data before shutting
>                * down.
> -              */
> -             if (got_STOPPING)
> +              *
> +              * We'll also exit here if the client sent CopyDone because it wants
> +              * to return to command mode.
> +             */
> +
> +             if (got_STOPPING || streamingDoneReceiving)
>                       break;
> 
>               /*
> @@ -2095,7 +2107,14 @@ WalSndCheckTimeOut(void)
>       }
>  }
> 
> -/* Main loop of walsender process that streams the WAL over Copy messages. */
> +/*
> + * Main loop of walsender process that streams the WAL over Copy messages.
> + *
> + * The send_data callback must enqueue complete CopyData messages to libpq
> + * using pq_putmessage_noblock or similar, since the walsender loop may send
> + * CopyDone then exit and return to command mode in response to a client
> + * CopyDone between calls to send_data.
> + */

Wait, how is it ok to end CopyDone before all the pending data has been
sent out?



> diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
> index 23466ba..66b6e90 100644
> --- a/src/backend/replication/logical/reorderbuffer.c
> +++ b/src/backend/replication/logical/reorderbuffer.c
> @@ -1497,7 +1497,9 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
>               rb->begin(rb, txn);
> 
>               iterstate = ReorderBufferIterTXNInit(rb, txn);
> -             while ((change = ReorderBufferIterTXNNext(rb, iterstate)) != NULL)
> +             while ((change = ReorderBufferIterTXNNext(rb, iterstate)) != NULL &&
> +                        (rb->continue_decoding_cb == NULL ||
> +                             rb->continue_decoding_cb()))
>               {
>                       Relation        relation = NULL;
>                       Oid                     reloid;

> @@ -1774,8 +1776,11 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
>               ReorderBufferIterTXNFinish(rb, iterstate);
>               iterstate = NULL;
> 
> -             /* call commit callback */
> -             rb->commit(rb, txn, commit_lsn);
> +             if (rb->continue_decoding_cb == NULL || rb->continue_decoding_cb())
> +             {
> +                     /* call commit callback */
> +                     rb->commit(rb, txn, commit_lsn);
> +             }


I'm doubtful it's ok to simply stop in the middle of a transaction.



> @@ -1194,17 +1212,6 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
> 
>       CHECK_FOR_INTERRUPTS();
> 
> -     /* Try to flush pending output to the client */
> -     if (pq_flush_if_writable() != 0)
> -             WalSndShutdown();
> -
> -     /* Try taking fast path unless we get too close to walsender timeout. */
> -     if (now < TimestampTzPlusMilliseconds(last_reply_timestamp,
> -                                                                               wal_sender_timeout / 2) &&
> -             !pq_is_send_pending())
> -     {
> -             return;
> -     }

As somebody else commented on the thread, I'm also doubtful this is
ok. This'll introduce significant additional blocking unless I'm missing
something?



>       /* If we have pending write here, go to slow path */
>       for (;;)
> @@ -1358,7 +1365,14 @@ WalSndWaitForWal(XLogRecPtr loc)
>                       break;
> 
>               /*
> -              * We only send regular messages to the client for full decoded
> +              * If we have received CopyDone from the client, sent CopyDone
> +              * ourselves, it's time to exit streaming.
> +              */
> +             if (!IsStreamingActive()) {
> +                     break;
> +             }

Wrong formatting.


I wonder if the saner approach here isn't to support query cancellations
or something of that vein, and then handle the error.

Greetings,

Andres Freund