public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: Amit Kapila <[email protected]>
To: Euler Taveira <[email protected]>
Cc: Masahiko Sawada <[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: Wed, 6 Apr 2022 05:39:05 +0000
Message-ID: <OS3PR01MB6275FD2CE0850FC66512D8F99EE79@OS3PR01MB6275.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAA4eK1+HNmgtWRm2U7EG2T0yWbAvhqSDDFBmKX5+vSgBsRmY=w@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>
On Fri, Apr 1, 2022 at 12:09 AM Amit Kapila <[email protected]> wrote:
> On Fri, Apr 1, 2022 at 8:28 AM Euler Taveira <[email protected]> wrote:
> >
> > On Thu, Mar 31, 2022, at 11:27 PM, Amit Kapila wrote:
> >
> > This is exactly our initial analysis and we have tried a patch on
> > these lines and it has a noticeable overhead. See [1]. Calling this
> > for each change or each skipped change can bring noticeable overhead
> > that is why we decided to call it after a certain threshold (100) of
> > skipped changes. Now, surely as mentioned in my previous reply we can
> > make it generic such that instead of calling this (update_progress
> > function as in the patch) for skipped cases, we call it always. Will
> > that make it better?
> >
> > That's what I have in mind but using a different approach.
> >
> > > The functions CreateInitDecodingContext and CreateDecodingContext
> receives the
> > > update_progress function as a parameter. These functions are called in 2
> > > places: (a) streaming replication protocol (CREATE_REPLICATION_SLOT) and
> (b)
> > > SQL logical decoding functions (pg_logical_*_changes). Case (a) uses
> > > WalSndUpdateProgress as a progress function. Case (b) does not have one
> because
> > > it is not required -- local decoding/communication. There is no custom
> update
> > > progress routine for each output plugin which leads me to the question:
> > > couldn't we encapsulate the update progress call into the callback functions?
> > >
> >
> > Sorry, I don't get your point. What exactly do you mean by this?
> > AFAIS, currently we call this output plugin API in pgoutput functions
> > only, do you intend to get it invoked from a different place?
> >
> > It seems I didn't make myself clear. The callbacks I'm referring to the
> > *_cb_wrapper functions. After every ctx->callbacks.foo_cb() call into a
> > *_cb_wrapper() function, we have something like:
> >
> > if (ctx->progress & PGOUTPUT_PROGRESS_FOO)
> > NewUpdateProgress(ctx, false);
> >
> > The NewUpdateProgress function would contain a logic similar to the
> > update_progress() from the proposed patch. (A different function name here
> just
> > to avoid confusion.)
> >
> > The output plugin is responsible to set ctx->progress with the callback
> > variables (for example, PGOUTPUT_PROGRESS_CHANGE for change_cb())
> that we would
> > like to run NewUpdateProgress.
> >
>
> This sounds like a conflicting approach to what we currently do.
> Currently, OutputPluginUpdateProgress() is called from the xact
> related pgoutput functions like pgoutput_commit_txn(),
> pgoutput_prepare_txn(), pgoutput_commit_prepared_txn(), etc. So, if we
> follow what you are saying then for some of the APIs like
> pgoutput_change/_message/_truncate, we need to set the parameter to
> invoke NewUpdateProgress() which will internally call
> OutputPluginUpdateProgress(), and for the remaining APIs, we will call
> in the corresponding pgoutput_* function. I feel if we want to make it
> more generic than the current patch, it is better to directly call
> what you are referring to here as NewUpdateProgress() in all remaining
> APIs like pgoutput_change/_truncate, etc.
Thanks for your comments.
According to your suggestion, improve the patch to make it more generic.
Attach the new patch.
Regards,
Wang wei
Attachments:
[application/octet-stream] v11-0001-Fix-the-logical-replication-timeout-during-large.patch (10.0K, ../OS3PR01MB6275FD2CE0850FC66512D8F99EE79@OS3PR01MB6275.jpnprd01.prod.outlook.com/2-v11-0001-Fix-the-logical-replication-timeout-during-large.patch)
download | inline diff:
From 5115b185ee548906187b619815cd8d56377888c5 Mon Sep 17 00:00:00 2001
From: wangw <[email protected]>
Date: Wed, 6 Apr 2022 10:56:50 +0800
Subject: [PATCH v11] 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 certan threshold of changes.
---
src/backend/replication/logical/logical.c | 7 +--
src/backend/replication/pgoutput/pgoutput.c | 52 ++++++++++++++++++---
src/backend/replication/walsender.c | 16 +++++--
src/include/replication/logical.h | 3 +-
src/include/replication/output_plugin.h | 3 +-
5 files changed, 67 insertions(+), 14 deletions(-)
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index e1f14aeecb..ea00aee126 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -680,17 +680,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 last_write)
{
if (!ctx->update_progress)
return;
ctx->update_progress(ctx, ctx->write_location, ctx->write_xid,
- skipped_xact);
+ skipped_xact, last_write);
}
/*
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 20d0b1e125..e47662a94e 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -96,6 +96,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_progress(LogicalDecodingContext *ctx);
/*
* Only 3 publication actions are used for row filtering ("insert", "update",
@@ -577,7 +578,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);
+ OutputPluginUpdateProgress(ctx, !sent_begin_txn, true);
pfree(txndata);
txn->output_plugin_private = NULL;
@@ -616,7 +617,7 @@ static void
pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
XLogRecPtr prepare_lsn)
{
- OutputPluginUpdateProgress(ctx, false);
+ OutputPluginUpdateProgress(ctx, false, true);
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_prepare(ctx->out, txn, prepare_lsn);
@@ -630,7 +631,7 @@ static void
pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
XLogRecPtr commit_lsn)
{
- OutputPluginUpdateProgress(ctx, false);
+ OutputPluginUpdateProgress(ctx, false, true);
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_commit_prepared(ctx->out, txn, commit_lsn);
@@ -646,7 +647,7 @@ pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx,
XLogRecPtr prepare_end_lsn,
TimestampTz prepare_time)
{
- OutputPluginUpdateProgress(ctx, false);
+ OutputPluginUpdateProgress(ctx, false, true);
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_rollback_prepared(ctx->out, txn, prepare_end_lsn,
@@ -1379,6 +1380,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
TupleTableSlot *old_slot = NULL;
TupleTableSlot *new_slot = NULL;
+ update_progress(ctx);
+
if (!is_publishable_relation(relation))
return;
@@ -1611,6 +1614,8 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
Oid *relids;
TransactionId xid = InvalidTransactionId;
+ update_progress(ctx);
+
/* Remember the xid for the change in streaming mode. See pgoutput_change. */
if (in_streaming)
xid = change->txn->xid;
@@ -1674,6 +1679,8 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
TransactionId xid = InvalidTransactionId;
+ update_progress(ctx);
+
if (!data->messages)
return;
@@ -1718,6 +1725,8 @@ pgoutput_sequence(LogicalDecodingContext *ctx,
TransactionId xid = InvalidTransactionId;
RelationSyncEntry *relentry;
+ update_progress(ctx);
+
if (!data->sequences)
return;
@@ -1924,7 +1933,7 @@ pgoutput_stream_commit(struct LogicalDecodingContext *ctx,
Assert(!in_streaming);
Assert(rbtxn_is_streamed(txn));
- OutputPluginUpdateProgress(ctx, false);
+ OutputPluginUpdateProgress(ctx, false, true);
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_stream_commit(ctx->out, txn, commit_lsn);
@@ -1945,7 +1954,7 @@ pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx,
{
Assert(rbtxn_is_streamed(txn));
- OutputPluginUpdateProgress(ctx, false);
+ OutputPluginUpdateProgress(ctx, false, true);
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn);
OutputPluginWrite(ctx, true);
@@ -2443,3 +2452,34 @@ 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 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_progress(LogicalDecodingContext *ctx)
+{
+ static int changes_count = 0;
+
+ /*
+ * After continuously processing CHANGES_THRESHOLD changes, update progress
+ * which will also try to send a keepalive message if required.
+ *
+ * We don't want to try sending a keepalive message or updating progress
+ * 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, false, false);
+ changes_count = 0;
+ }
+}
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 75400a53f2..14029a7f26 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 last_write);
static XLogRecPtr WalSndWaitForWal(XLogRecPtr loc);
static void LagTrackerWrite(XLogRecPtr lsn, TimestampTz local_flush_time);
static TimeOffset LagTrackerRead(int head, XLogRecPtr lsn, TimestampTz now);
@@ -1461,13 +1461,17 @@ ProcessPendingWrites(void)
* Write the current position to the lag tracker (see XLogSendPhysical).
*
* When skipping empty transactions, send a keepalive message if necessary.
+ *
+ * If too many changes are processed then try to send a keepalive message to
+ * receiver to avoid timeouts.
*/
static void
WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
- bool skipped_xact)
+ bool skipped_xact, bool last_write)
{
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
@@ -1501,8 +1505,14 @@ 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 */
+ if (pending_writes || (!last_write &&
+ 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..976bcf727c 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 last_write
);
typedef struct LogicalDecodingContext
diff --git a/src/include/replication/output_plugin.h b/src/include/replication/output_plugin.h
index fe85d49a03..31baf36fe2 100644
--- a/src/include/replication/output_plugin.h
+++ b/src/include/replication/output_plugin.h
@@ -270,6 +270,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 last_write);
#endif /* OUTPUT_PLUGIN_H */
--
2.23.0.windows.1
view thread (35+ 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: <OS3PR01MB6275FD2CE0850FC66512D8F99EE79@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