On 10 May 2016 at 09:50, Craig Ringer <cr...@2ndquadrant.com> wrote:
> I outlined how I think WalSndWaitForWal() should be handled earlier. Test
> streamingDoneReceiving
> and streamingDoneSending in logical_read_xlog_page and return -1.
>
OK, so thinking about this some more, I see why you've added the callback
within the reorder buffer code. You want to stop processing of a
transaction after we've decoded the commit and are looping over the changes
within ReorderBufferCommit(...), which doesn't know anything about the
walsender. So we could loop for a long time within WalSndLoop ->
XLogSendLogical -> LogicalDecodingProcessRecord if the record is a commit,
as we process each change and send it to the client.
So aborting processing and returning between individual changes in
ReorderBufferCommit(...) seems very reasonable. I agree that some kind of
callback is needed because the walsender uses static globals to control its
send/receive stop logic. I don't like the naming "is_active"; maybe reverse
the sense and call it "stop_decoding_cb" ? Or "continue_decoding_cb"?
Unsure.
Anyway, here's a draft patch that does the parts other than the reorder
buffer processing stop. It passes 'make check' and the src/test/recovery
tests, but I haven't written anything to verify the client-initiated abort
handling. You have test code for this and I'd be interested to see the
results.
This patch doesn't attempt to allow decoding to be aborted during
processing of an xlog record, including a commit. So it'll still attempt to
send whole transactions. But it should allow clients to send CopyDone when
we're waiting for new WAL to be generated and return to command mode then.
I think it's worth making the next step, where you allow reorder buffer
commit processing to be interrupted, into a separate patch on top of this
one. They're two separate changes IMO.
--
Craig Ringer http://www.2ndQuadrant.com/
PostgreSQL Development, 24x7 Support, Training & Services
From 14c0ac1aacfacb3b6b42bf0c5e453ab4e989aa2c Mon Sep 17 00:00:00 2001
From: Craig Ringer <cr...@2ndquadrant.com>
Date: Tue, 10 May 2016 10:34:10 +0800
Subject: [PATCH] Respect client-initiated CopyDone in 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 926a247..4e2164e 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;
/*
@@ -1789,7 +1799,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)
{
@@ -1852,10 +1869,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 */
@@ -2911,7 +2933,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 14c0ac1aacfacb3b6b42bf0c5e453ab4e989aa2c Mon Sep 17 00:00:00 2001
From: Craig Ringer <cr...@2ndquadrant.com>
Date: Tue, 10 May 2016 10:34:10 +0800
Subject: [PATCH] Respect client-initiated CopyDone in 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 926a247..4e2164e 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;
/*
@@ -1789,7 +1799,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)
{
@@ -1852,10 +1869,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 */
@@ -2911,7 +2933,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
--
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
