On 25 August 2016 at 13:04, Craig Ringer <cr...@2ndquadrant.com> wrote:
> By the way, I now think that the second part of your patch, to allow
> interruption during ReorderBufferCommit processing, is also very
> desirable.
I've updated your patch, rebasing it on top of 10.0 master and
splitting it into two pieces. I thought you were planning on following
up with an update to the second part, but maybe that was unclear from
our prior discussion, so I'm doing this to help get this moving again.
The first patch is what I posted earlier - the part of your patch that
allows the walsender to exit COPY BOTH mode and return to command mode
in response to a client-initiated CopyDone when it's in the
WalSndLoop. For logical decoding this means that it will notice a
client CopyDone when it's between transactions or while it's ingesting
transaction changes. It won't react to CopyDone while it's in
ReorderBufferCommit and sending a transaction to an output plugin
though.
The second piece is the other half of your original patch. Currently
unmodified from what you wrote. It adds a hook in ReorderBufferCommit
that calls back into the walsender to check for new client input and
interrupt processing. I haven't yet investigated this in detail.
Review comments on the 2nd patch, i.e. the 2nd half of your original patch:
* Other places in logical decoding use the CB suffix for callback
types. This should do the same.
* I'm not too keen on the name is_active for the callback. We
discussed the name continue_decoding_cb in our prior conversation.
* Instead of having LogicalContextAlwaysActive, would it be better to
test if the callback pointer is null and skip it if so?
* Lots of typos in LogicalContextAlwaysActive comments, and wording is unclear
* comment added to arguments of pg_logical_slot_get_changes_guts is
not in PostgreSQL style - it goes way wide on the line. Probably
better to put the comment above the call and mention which argument it
refers to.
* A comment somewhere is needed - probably on the IsStreamingActive()
callback - to point readers at the fact that WalSndWriteData() calls
ProcessRepliesIfAny() and that's how IsStreamingActive can get set. I
had to look at the email discussion history to remind myself how this
worked when I was re-reading the code.
* I'm not keen on
typedef ReorderBufferIsActive LogicalDecondingContextIsActive;
I think instead a single callback name that encompasses both should
be used. ContinueDecodingCB ?
* There are no tests. I don't see any really simple way to test this, though.
I suggest that you apply these updated/split patches to a fresh copy
of master then see if you can update it based on this feedback.
Something like:
git checkout master
git pull
git checkout -b stop-decoding-v10
git am /path/to/0001-Respect-client-initiated-CopyDone-in-walsender.patch
git am /path/to/0002-Second-part-of-Vladimir-Gordiychuk-s-patch.patch
then modify the code per the above, and "git commit --amend" the
changes and send an updated set of patches with "git format-patch -2"
.
I am setting this patch as "waiting on author" in the commitfest.
--
Craig Ringer http://www.2ndQuadrant.com/
PostgreSQL Development, 24x7 Support, Training & Services
From 81b3f61b0235e1b936e3336e43ce55c00bc230c4 Mon Sep 17 00:00:00 2001
From: Craig Ringer <cr...@2ndquadrant.com>
Date: Tue, 10 May 2016 10:34:10 +0800
Subject: [PATCH 1/2] Respect client-initiated CopyDone in walsender
The walsender never looked for CopyDone sent by the client unless it
had already decided it was done sending data and dispatched its own
CopyDone message.
Check for client-initiated CopyDone when in COPY BOTH mode, returning to
command mode. In logical decoding this will allow the client to end a logical
decoding session between transactions without just unilaterally closing its
connection.
This change does not allow a client to end COPY BOTH session in the middle of
processing a logical decoding block.
TODO effect on physical walsender?
---
src/backend/replication/walsender.c | 32 +++++++++++++++++++++++++++-----
1 file changed, 27 insertions(+), 5 deletions(-)
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 1ea2a5c..c43310c 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -759,6 +759,13 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req
/* make sure we have enough WAL available */
flushptr = WalSndWaitForWal(targetPagePtr + reqLen);
+ /*
+ * 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 */
if (targetPagePtr + XLOG_BLCKSZ <= flushptr)
count = XLOG_BLCKSZ;
@@ -1220,8 +1227,11 @@ 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.
+ *
+ * We'll also exit here if the client sent CopyDone because it wants
+ * to return to command mode.
*/
- if (walsender_ready_to_stop)
+ if (walsender_ready_to_stop || streamingDoneReceiving)
break;
/*
@@ -1787,7 +1797,14 @@ WalSndCheckTimeOut(TimestampTz now)
}
}
-/* 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.
+ */
static void
WalSndLoop(WalSndSendDataCallback send_data)
{
@@ -1850,10 +1867,15 @@ WalSndLoop(WalSndSendDataCallback send_data)
* some more. If there is some, we don't bother to call send_data
* again until we've flushed it ... but we'd better assume we are not
* caught up.
+ *
+ * If we're trying to finish sending and exit we shouldn't enqueue more
+ * data to libpq. We need to finish writing out whatever we already
+ * have in libpq's send buffer to maintain protocol sync so we still
+ * need to loop until it's flushed.
*/
- if (!pq_is_send_pending())
+ if (!pq_is_send_pending() && !streamingDoneSending)
send_data();
- else
+ else if (!streamingDoneSending)
WalSndCaughtUp = false;
/* Try to flush pending output to the client */
@@ -2909,7 +2931,7 @@ WalSndKeepaliveIfNecessary(TimestampTz now)
if (wal_sender_timeout <= 0 || last_reply_timestamp <= 0)
return;
- if (waiting_for_ping_response)
+ if (waiting_for_ping_response || streamingDoneSending)
return;
/*
--
2.5.5
From 43de23ddafb17950add6ffff585e501e5ce4c54b Mon Sep 17 00:00:00 2001
From: Craig Ringer <cr...@2ndquadrant.com>
Date: Mon, 5 Sep 2016 22:19:03 +0800
Subject: [PATCH 2/2] Second part of Vladimir Gordiychuk's patch
This component allows streaming to be interrupted during ReorderBufferCommit.
TODO: explain, write up, document
---
src/backend/replication/logical/logical.c | 15 ++++++----
src/backend/replication/logical/logicalfuncs.c | 15 +++++++++-
src/backend/replication/logical/reorderbuffer.c | 9 ++++--
src/backend/replication/slotfuncs.c | 2 +-
src/backend/replication/walsender.c | 39 +++++++++++++------------
src/include/replication/logical.h | 13 +++++++--
src/include/replication/reorderbuffer.h | 8 +++++
7 files changed, 70 insertions(+), 31 deletions(-)
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 1512be5..818b7cf 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -116,7 +116,8 @@ StartupDecodingContext(List *output_plugin_options,
TransactionId xmin_horizon,
XLogPageReadCB read_page,
LogicalOutputPluginWriterPrepareWrite prepare_write,
- LogicalOutputPluginWriterWrite do_write)
+ LogicalOutputPluginWriterWrite do_write,
+ ReorderBufferIsActive is_active)
{
ReplicationSlot *slot;
MemoryContext context,
@@ -180,6 +181,7 @@ StartupDecodingContext(List *output_plugin_options,
ctx->reorder->apply_change = change_cb_wrapper;
ctx->reorder->commit = commit_cb_wrapper;
ctx->reorder->message = message_cb_wrapper;
+ ctx->reorder->is_active = is_active;
ctx->out = makeStringInfo();
ctx->prepare_write = prepare_write;
@@ -212,7 +214,8 @@ CreateInitDecodingContext(char *plugin,
List *output_plugin_options,
XLogPageReadCB read_page,
LogicalOutputPluginWriterPrepareWrite prepare_write,
- LogicalOutputPluginWriterWrite do_write)
+ LogicalOutputPluginWriterWrite do_write,
+ LogicalDecondingContextIsActive is_active)
{
TransactionId xmin_horizon = InvalidTransactionId;
ReplicationSlot *slot;
@@ -288,7 +291,7 @@ CreateInitDecodingContext(char *plugin,
ReplicationSlotSave();
ctx = StartupDecodingContext(NIL, InvalidXLogRecPtr, xmin_horizon,
- read_page, prepare_write, do_write);
+ read_page, prepare_write, do_write, is_active);
/* call output plugin initialization callback */
old_context = MemoryContextSwitchTo(ctx->context);
@@ -328,7 +331,8 @@ CreateDecodingContext(XLogRecPtr start_lsn,
List *output_plugin_options,
XLogPageReadCB read_page,
LogicalOutputPluginWriterPrepareWrite prepare_write,
- LogicalOutputPluginWriterWrite do_write)
+ LogicalOutputPluginWriterWrite do_write,
+ LogicalDecondingContextIsActive is_active)
{
LogicalDecodingContext *ctx;
ReplicationSlot *slot;
@@ -378,7 +382,8 @@ CreateDecodingContext(XLogRecPtr start_lsn,
ctx = StartupDecodingContext(output_plugin_options,
start_lsn, InvalidTransactionId,
- read_page, prepare_write, do_write);
+ read_page, prepare_write, do_write,
+ is_active);
/* call output plugin initialization callback */
old_context = MemoryContextSwitchTo(ctx->context);
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 4e4c8cd..e0efd4c 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -104,6 +104,17 @@ LogicalOutputWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xi
p->returned_rows++;
}
+/**
+ * Stab function that necessary for LogicalDecondign context.
+ * Function always return true and it means that decoding WALs
+ * can't be interrupt in contrast of logical replication.
+ */
+static bool
+LogicalContextAlwaysActive(void)
+{
+ return true;
+}
+
static void
check_permissions(void)
{
@@ -249,7 +260,9 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
options,
logical_read_local_xlog_page,
LogicalOutputPrepareWrite,
- LogicalOutputWrite);
+ LogicalOutputWrite,
+ LogicalContextAlwaysActive /* converting to Datum(sql api) can't be interrupted in contrast of replication*/
+ );
MemoryContextSwitchTo(oldcontext);
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 9b430b9..26b8e71 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -1417,7 +1417,7 @@ 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->is_active())
{
Relation relation = NULL;
Oid reloid;
@@ -1643,8 +1643,11 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
ReorderBufferIterTXNFinish(rb, iterstate);
iterstate = NULL;
- /* call commit callback */
- rb->commit(rb, txn, commit_lsn);
+ if(rb->is_active())
+ {
+ /* call commit callback */
+ rb->commit(rb, txn, commit_lsn);
+ }
/* this is just a sanity check against bad output plugin behaviour */
if (GetCurrentTransactionIdIfAny() != InvalidTransactionId)
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index f908761..976bc0c 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -127,7 +127,7 @@ pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
*/
ctx = CreateInitDecodingContext(
NameStr(*plugin), NIL,
- logical_read_local_xlog_page, NULL, NULL);
+ logical_read_local_xlog_page, NULL, NULL, NULL);
/* build initial snapshot, might take a while */
DecodingContextFindStartpoint(ctx);
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index c43310c..9f4c7b5 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -218,6 +218,8 @@ static XLogRecPtr WalSndWaitForWal(XLogRecPtr loc);
static void XLogRead(char *buf, XLogRecPtr startptr, Size count);
+static bool IsStreamingActive(void);
+
/* Initialize walsender process before entering the main command loop */
void
@@ -823,7 +825,8 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
ctx = CreateInitDecodingContext(cmd->plugin, NIL,
logical_read_xlog_page,
- WalSndPrepareWrite, WalSndWriteData);
+ WalSndPrepareWrite, WalSndWriteData,
+ IsStreamingActive);
/*
* Signal that we don't need the timeout mechanism. We're just
@@ -1002,7 +1005,7 @@ StartLogicalReplication(StartReplicationCmd *cmd)
logical_decoding_ctx = CreateDecodingContext(
cmd->startpoint, cmd->options,
logical_read_xlog_page,
- WalSndPrepareWrite, WalSndWriteData);
+ WalSndPrepareWrite, WalSndWriteData, IsStreamingActive);
/* Start reading WAL from the oldest required WAL. */
logical_startptr = MyReplicationSlot->data.restart_lsn;
@@ -1093,14 +1096,6 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
memcpy(&ctx->out->data[1 + sizeof(int64) + sizeof(int64)],
tmpbuf.data, sizeof(int64));
- /* fast path */
- /* Try to flush pending output to the client */
- if (pq_flush_if_writable() != 0)
- WalSndShutdown();
-
- if (!pq_is_send_pending())
- return;
-
for (;;)
{
int wakeEvents;
@@ -1235,7 +1230,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;
+ }
+
+ /* We only send regular messages to the client for full decoded
* transactions, but a synchronous replication and walsender shutdown
* possibly are waiting for a later location. So we send pings
* containing the flush location every now and then.
@@ -1797,14 +1799,7 @@ WalSndCheckTimeOut(TimestampTz now)
}
}
-/*
- * 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.
- */
+/* Main loop of walsender process that streams the WAL over Copy messages. */
static void
WalSndLoop(WalSndSendDataCallback send_data)
{
@@ -2951,3 +2946,9 @@ WalSndKeepaliveIfNecessary(TimestampTz now)
WalSndShutdown();
}
}
+
+static
+bool IsStreamingActive(void)
+{
+ return !streamingDoneReceiving && !streamingDoneSending;
+}
diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h
index 947000e..ba54ddc 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -26,6 +26,13 @@ typedef void (*LogicalOutputPluginWriterWrite) (
typedef LogicalOutputPluginWriterWrite LogicalOutputPluginWriterPrepareWrite;
+/*
+ * Callback function that allow interrupt logical replication during decoding.
+ * Function return true if decoding can be continue decode, but if function return false
+ * logical decoding will stop as soon as possible.
+ */
+typedef ReorderBufferIsActive LogicalDecondingContextIsActive;
+
typedef struct LogicalDecodingContext
{
/* memory context this is all allocated in */
@@ -81,13 +88,15 @@ extern LogicalDecodingContext *CreateInitDecodingContext(char *plugin,
List *output_plugin_options,
XLogPageReadCB read_page,
LogicalOutputPluginWriterPrepareWrite prepare_write,
- LogicalOutputPluginWriterWrite do_write);
+ LogicalOutputPluginWriterWrite do_write,
+ LogicalDecondingContextIsActive is_active);
extern LogicalDecodingContext *CreateDecodingContext(
XLogRecPtr start_lsn,
List *output_plugin_options,
XLogPageReadCB read_page,
LogicalOutputPluginWriterPrepareWrite prepare_write,
- LogicalOutputPluginWriterWrite do_write);
+ LogicalOutputPluginWriterWrite do_write,
+ LogicalDecondingContextIsActive is_active);
extern void DecodingContextFindStartpoint(LogicalDecodingContext *ctx);
extern bool DecodingContextReady(LogicalDecodingContext *ctx);
extern void FreeDecodingContext(LogicalDecodingContext *ctx);
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 9e209ae..b0916d7 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -292,6 +292,9 @@ typedef void (*ReorderBufferMessageCB) (
const char *prefix, Size sz,
const char *message);
+/* callback signature for check decoding status */
+typedef bool (*ReorderBufferIsActive) (void);
+
struct ReorderBuffer
{
/*
@@ -321,6 +324,11 @@ struct ReorderBuffer
ReorderBufferMessageCB message;
/*
+ * Callback to define status of decoding. Return false if decoding not necessary continue
+ */
+ ReorderBufferIsActive is_active;
+
+ /*
* Pointer that will be passed untouched to the callbacks.
*/
void *private_data;
--
2.5.5
--
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
