public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: Masahiko Sawada <[email protected]>
To: Amit Kapila <[email protected]>
Cc: Euler Taveira <[email protected]>
Cc: [email protected] <[email protected]>
Cc: Peter Smith <[email protected]>
Cc: Fabrice Chapuis <[email protected]>
Cc: Simon Riggs <[email protected]>
Cc: Petr Jelinek <[email protected]>
Cc: [email protected] <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Cc: Ajin Cherian <[email protected]>
Subject: RE: Logical replication timeout problem
Date: Mon, 18 Apr 2022 06:16:40 +0000
Message-ID: <OS3PR01MB627561344A2C7ECF68E41D7E9EF39@OS3PR01MB6275.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAD21AoDnyjNnEjgeC6ALKwCNXcqtz7SRPgDA-0HV8gguLdLDHw@mail.gmail.com>
References: <CAA4eK1+fQjndoBOFUn9Wy0hhm3MLyUWEpcT9O7iuCELktfdBiQ@mail.gmail.com>
<CAA4eK1LGnaPuWs2M4sDfpd6JQZjoh4DGAsgUvNW=Or8i9z6K8w@mail.gmail.com>
<OS3PR01MB62756599B78FA7D4C908109A9E179@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAA4eK1JKZrOfCLMVZYfUnm_wfxucDjOzeGwQy3hJ_U9Y3AV05Q@mail.gmail.com>
<OS3PR01MB627542044A4C16511E63D5DA9E1A9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAD21AoCEhZ6D1VtqvxK1vO0=NnJd6phWbijt-w=M5hwka=-c1A@mail.gmail.com>
<OS3PR01MB62750C20AD7AF4F13541FC109E1A9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<TYAPR01MB586636517DED745651D6E75AF51D9@TYAPR01MB5866.jpnprd01.prod.outlook.com>
<OS3PR01MB6275C64F264662E84D2FB7AE9E1D9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<OS3PR01MB6275F0E4E27F5522847589ED9E1E9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<OS3PR01MB6275E0C2B4D9E488AD7CBA209E1F9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAA4eK1JUgO9F5fZb5P8quMYfDbD7SBNApsTsGLf6vYNNetpWTA@mail.gmail.com>
<CAD21AoCQi33mjd=bM0Jhkk3QaV-fHVB_soA7iQmTor2A-1Cx3Q@mail.gmail.com>
<[email protected]>
<CAA4eK1+wSvfDVUSe-4o6APFdO8=9G-HnZJ+01_LQOxqLmhqe5Q@mail.gmail.com>
<[email protected]>
<CAA4eK1+HNmgtWRm2U7EG2T0yWbAvhqSDDFBmKX5+vSgBsRmY=w@mail.gmail.com>
<OS3PR01MB6275FD2CE0850FC66512D8F99EE79@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAA4eK1+A2aesu8YS9Y8bFBCbOUw_UkU1Nj_-V6rfXBiJgCxqKg@mail.gmail.com>
<CAA4eK1L-MU+YfRUD0J1wBu6CJ2FEaSVh=Kr3PjjyshprUpvkeg@mail.gmail.com>
<OS3PR01MB627516D7A3A2ED91EEAA510C9EE79@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAA4eK1L8jDBGqdZU0Fk6LDSpdqE+3wg2AwkgdxwiCqmXHc6aoQ@mail.gmail.com>
<OS3PR01MB62755D216245199554DDC8DB9EEA9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAA4eK1Kj=-5opwBstZtu2CBDMtoQPmDD4c=X5arqx54wq52FNw@mail.gmail.com>
<CAD21AoC-VnFSB8WHaXpD-X1h7Kz0DVs8C+tVSdPGw+b=BgJG1w@mail.gmail.com>
<CAA4eK1Ly6eA4ui60mABgtQki0XRLPcSqjQngMTXHGPjjZOtCHQ@mail.gmail.com>
<CAD21AoDnyjNnEjgeC6ALKwCNXcqtz7SRPgDA-0HV8gguLdLDHw@mail.gmail.com>
On Mon, Apr 18, 2022 at 00:35 PM Masahiko Sawada <[email protected]> wrote:
> On Mon, Apr 18, 2022 at 1:01 PM Amit Kapila <[email protected]> wrote:
> >
> > On Thu, Apr 14, 2022 at 5:50 PM Masahiko Sawada <[email protected]>
> wrote:
> > >
> > > On Wed, Apr 13, 2022 at 7:45 PM Amit Kapila <[email protected]>
> wrote:
> > > >
> > > > On Mon, Apr 11, 2022 at 12:09 PM [email protected]
> > > > <[email protected]> wrote:
> > > > >
> > > > > So I skip tracking lag during a transaction just like the current HEAD.
> > > > > Attach the new patch.
> > > > >
> > > >
> > > > Thanks, please find the updated patch where I have slightly
> > > > modified the comments.
> > > >
> > > > Sawada-San, Euler, do you have any opinion on this approach? I
> > > > personally still prefer the approach implemented in v10 [1]
> > > > especially due to the latest finding by Wang-San that we can't
> > > > update the lag-tracker apart from when it is invoked at the transaction end.
> > > > However, I am fine if we like this approach more.
> > >
> > > Thank you for updating the patch.
> > >
> > > The current patch looks much better than v10 which requires to call
> > > to
> > > update_progress() every path.
> > >
> > > Regarding v15 patch, I'm concerned a bit that the new function name,
> > > update_progress(), is too generic. How about
> > > update_replation_progress() or something more specific name?
> > >
> >
> > Do you intend to say update_replication_progress()? The word
> > 'replation' doesn't make sense to me. I am fine with this suggestion.
>
> Yeah, that was a typo. I meant update_replication_progress().
Thanks for your comments.
> > > Regarding v15 patch, I'm concerned a bit that the new function name,
> > > update_progress(), is too generic. How about
> > > update_replation_progress() or something more specific name?
Improve as suggested. Change the name from update_progress to
update_replication_progress.
> > > ---
> > > + if (end_xact)
> > > + {
> > > + /* Update progress tracking at xact end. */
> > > + OutputPluginUpdateProgress(ctx, skipped_xact, end_xact);
> > > + changes_count = 0;
> > > + return;
> > > + }
> > > +
> > > + /*
> > > + * After continuously processing CHANGES_THRESHOLD changes,
> > > we try to send
> > > + * a keepalive message if required.
> > > + *
> > > + * We don't want to try sending a keepalive message after
> > > processing each
> > > + * change as that can have overhead. Testing reveals that there is no
> > > + * noticeable overhead in doing it after continuously
> > > processing 100 or so
> > > + * changes.
> > > + */
> > > +#define CHANGES_THRESHOLD 100
> > > + if (++changes_count >= CHANGES_THRESHOLD)
> > > + {
> > > + OutputPluginUpdateProgress(ctx, skipped_xact, end_xact);
> > > + changes_count = 0;
> > > + }
> > >
> > > Can we merge two if branches since we do the same things? Or did you
> > > separate them for better readability?
Improve as suggested. Merge two if-branches.
Attach the new patch.
1. Rename the new function(update_progress) to update_replication_progress. [suggestion by Sawada-San]
2. Merge two if-branches in new function update_replication_progress. [suggestion by Sawada-San.]
3. Improve comments to make them clear. [suggestions by Euler-San.]
Regards,
Wang wei
Attachments:
[application/octet-stream] v16-0001-Fix-the-logical-replication-timeout-during-large.patch (11.0K, ../OS3PR01MB627561344A2C7ECF68E41D7E9EF39@OS3PR01MB6275.jpnprd01.prod.outlook.com/2-v16-0001-Fix-the-logical-replication-timeout-during-large.patch)
download | inline diff:
From 610c6000fa3c6e7fc98f8722b2091ff88e24171a Mon Sep 17 00:00:00 2001
From: wangw <[email protected]>
Date: Mon, 18 Apr 2022 14:03:19 +0800
Subject: [PATCH v16] Fix the logical replication timeout during large
transactions.
The problem is that we don't send keep-alive messages for a long time
while processing large transactions during logical replication where we
don't send any data of such transactions. This can happen when the table
modified in the transaction is not published or because all the changes
got filtered. We do try to send the keep_alive if necessary at the end of
the transaction (via WalSndWriteData()) but by that time the
subscriber-side can timeout and exit.
To fix this we try to send the keepalive message if required after
processing certain threshold of changes.
As this requires an exposed API change, we didn't backpatched it.
---
src/backend/replication/logical/logical.c | 7 +--
src/backend/replication/pgoutput/pgoutput.c | 53 ++++++++++++++++++---
src/backend/replication/walsender.c | 30 ++++++++++--
src/include/replication/logical.h | 3 +-
src/include/replication/output_plugin.h | 3 +-
5 files changed, 80 insertions(+), 16 deletions(-)
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 788769dd73..d3fe045c84 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -669,17 +669,18 @@ OutputPluginWrite(struct LogicalDecodingContext *ctx, bool last_write)
}
/*
- * Update progress tracking (if supported).
+ * Update progress tracking and try to send a keepalive message (if supported).
*/
void
OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx,
- bool skipped_xact)
+ bool skipped_xact,
+ bool end_xact)
{
if (!ctx->update_progress)
return;
ctx->update_progress(ctx, ctx->write_location, ctx->write_xid,
- skipped_xact);
+ skipped_xact, end_xact);
}
/*
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index b197bfd565..e64dee9f01 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -91,6 +91,7 @@ static void send_relation_and_attrs(Relation relation, TransactionId xid,
static void send_repl_origin(LogicalDecodingContext *ctx,
RepOriginId origin_id, XLogRecPtr origin_lsn,
bool send_origin);
+static void update_replication_progress(LogicalDecodingContext *ctx, bool skipped_xact, bool end_xact);
/*
* Only 3 publication actions are used for row filtering ("insert", "update",
@@ -558,7 +559,7 @@ pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
* from this transaction has been sent to the downstream.
*/
sent_begin_txn = txndata->sent_begin_txn;
- OutputPluginUpdateProgress(ctx, !sent_begin_txn);
+ update_replication_progress(ctx, !sent_begin_txn, true);
pfree(txndata);
txn->output_plugin_private = NULL;
@@ -597,7 +598,7 @@ static void
pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
XLogRecPtr prepare_lsn)
{
- OutputPluginUpdateProgress(ctx, false);
+ update_replication_progress(ctx, false, true);
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_prepare(ctx->out, txn, prepare_lsn);
@@ -611,7 +612,7 @@ static void
pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
XLogRecPtr commit_lsn)
{
- OutputPluginUpdateProgress(ctx, false);
+ update_replication_progress(ctx, false, true);
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_commit_prepared(ctx->out, txn, commit_lsn);
@@ -627,7 +628,7 @@ pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx,
XLogRecPtr prepare_end_lsn,
TimestampTz prepare_time)
{
- OutputPluginUpdateProgress(ctx, false);
+ update_replication_progress(ctx, false, true);
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_rollback_prepared(ctx->out, txn, prepare_end_lsn,
@@ -1360,6 +1361,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
TupleTableSlot *old_slot = NULL;
TupleTableSlot *new_slot = NULL;
+ update_replication_progress(ctx, false, false);
+
if (!is_publishable_relation(relation))
return;
@@ -1592,6 +1595,8 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
Oid *relids;
TransactionId xid = InvalidTransactionId;
+ update_replication_progress(ctx, false, false);
+
/* Remember the xid for the change in streaming mode. See pgoutput_change. */
if (in_streaming)
xid = change->txn->xid;
@@ -1655,6 +1660,8 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
TransactionId xid = InvalidTransactionId;
+ update_replication_progress(ctx, false, false);
+
if (!data->messages)
return;
@@ -1847,7 +1854,7 @@ pgoutput_stream_commit(struct LogicalDecodingContext *ctx,
Assert(!in_streaming);
Assert(rbtxn_is_streamed(txn));
- OutputPluginUpdateProgress(ctx, false);
+ update_replication_progress(ctx, false, true);
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_stream_commit(ctx->out, txn, commit_lsn);
@@ -1868,7 +1875,7 @@ pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx,
{
Assert(rbtxn_is_streamed(txn));
- OutputPluginUpdateProgress(ctx, false);
+ update_replication_progress(ctx, false, true);
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn);
OutputPluginWrite(ctx, true);
@@ -2361,3 +2368,37 @@ send_repl_origin(LogicalDecodingContext *ctx, RepOriginId origin_id,
}
}
}
+
+/*
+ * Try to update progress and send a keepalive message if too many changes were
+ * processed.
+ *
+ * For a large transaction, if we don't send any change to the downstream for a
+ * long time(exceeds the wal_receiver_timeout of standby) then it can timeout.
+ * This can happen when all or most of the changes are either not published or
+ * got filtered out.
+ */
+static void
+update_replication_progress(LogicalDecodingContext *ctx, bool skipped_xact, bool end_xact)
+{
+ static int changes_count = 0;
+
+ /*
+ * We don't want to try sending a keepalive message after processing each
+ * change as that can have overhead. Tests revealed that there is no
+ * noticeable overhead in doing it after continuously processing 100 or so
+ * changes.
+ */
+#define CHANGES_THRESHOLD 100
+
+ /*
+ * If we are at the end of transaction LSN, update progress tracking.
+ * Otherwise, after continuously processing CHANGES_THRESHOLD changes, we
+ * try to send a keepalive message if required.
+ */
+ if(end_xact || ++changes_count >= CHANGES_THRESHOLD)
+ {
+ OutputPluginUpdateProgress(ctx, skipped_xact, end_xact);
+ changes_count = 0;
+ }
+}
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index be40261393..75364340e1 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -251,7 +251,7 @@ static void WalSndWait(uint32 socket_events, long timeout, uint32 wait_event);
static void WalSndPrepareWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write);
static void WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write);
static void WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
- bool skipped_xact);
+ bool skipped_xact, bool end_xact);
static XLogRecPtr WalSndWaitForWal(XLogRecPtr loc);
static void LagTrackerWrite(XLogRecPtr lsn, TimestampTz local_flush_time);
static TimeOffset LagTrackerRead(int head, XLogRecPtr lsn, TimestampTz now);
@@ -1460,21 +1460,29 @@ ProcessPendingWrites(void)
* Write the current position to the lag tracker (see XLogSendPhysical).
*
* When skipping empty transactions, send a keepalive message if necessary.
+ *
+ * In logical replication, if too many changes are processed then try to send a
+ * keepalive message. It might avoid a timeout in the subscriber.
*/
static void
WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
- bool skipped_xact)
+ bool skipped_xact, bool end_xact)
{
static TimestampTz sendTime = 0;
TimestampTz now = GetCurrentTimestamp();
+ bool pending_writes = false;
/*
* Track lag no more than once per WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS to
* avoid flooding the lag tracker when we commit frequently.
+ *
+ * We don't have a mechanism to get the ack for any LSN other than end xact
+ * LSN from the downstream. So, we track lag only for end of transaction
+ * LSN.
*/
#define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 1000
- if (TimestampDifferenceExceeds(sendTime, now,
- WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS))
+ if (end_xact && TimestampDifferenceExceeds(sendTime, now,
+ WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS))
{
LagTrackerWrite(lsn, now);
sendTime = now;
@@ -1500,8 +1508,20 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId
/* If we have pending write here, make sure it's actually flushed */
if (pq_is_send_pending())
- ProcessPendingWrites();
+ pending_writes = true;
}
+
+ /*
+ * Process pending writes if any or try to send a keepalive if required. We
+ * don't need to try sending keep alive messages at the transaction end as
+ * that will be done at a later point in time. This is required only for
+ * large transactions where we don't send any changes to the downstream and
+ * the receiver can timeout due to that.
+ */
+ if (pending_writes || (!end_xact &&
+ now >= TimestampTzPlusMilliseconds(last_reply_timestamp,
+ wal_sender_timeout / 2)))
+ ProcessPendingWrites();
}
/*
diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h
index a6ef16ad5b..e4c50a193e 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -27,7 +27,8 @@ typedef LogicalOutputPluginWriterWrite LogicalOutputPluginWriterPrepareWrite;
typedef void (*LogicalOutputPluginWriterUpdateProgress) (struct LogicalDecodingContext *lr,
XLogRecPtr Ptr,
TransactionId xid,
- bool skipped_xact
+ bool skipped_xact,
+ bool end_xact
);
typedef struct LogicalDecodingContext
diff --git a/src/include/replication/output_plugin.h b/src/include/replication/output_plugin.h
index 539dc8e697..82929db313 100644
--- a/src/include/replication/output_plugin.h
+++ b/src/include/replication/output_plugin.h
@@ -243,6 +243,7 @@ typedef struct OutputPluginCallbacks
/* Functions in replication/logical/logical.c */
extern void OutputPluginPrepareWrite(struct LogicalDecodingContext *ctx, bool last_write);
extern void OutputPluginWrite(struct LogicalDecodingContext *ctx, bool last_write);
-extern void OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx, bool skipped_xact);
+extern void OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx, bool skipped_xact,
+ bool end_xact);
#endif /* OUTPUT_PLUGIN_H */
--
2.23.0.windows.1
view thread (30+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: RE: Logical replication timeout problem
In-Reply-To: <OS3PR01MB627561344A2C7ECF68E41D7E9EF39@OS3PR01MB6275.jpnprd01.prod.outlook.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox