public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: Amit Kapila <[email protected]>
Cc: 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 12:59:55 +0000
Message-ID: <OS3PR01MB627516D7A3A2ED91EEAA510C9EE79@OS3PR01MB6275.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAA4eK1L-MU+YfRUD0J1wBu6CJ2FEaSVh=Kr3PjjyshprUpvkeg@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>
On Wed, Apr 6, 2022 at 1:59 AM Amit Kapila <[email protected]> wrote:
On Wed, Apr 6, 2022 at 4:32 AM Amit Kapila <[email protected]> wrote:
>
Thanks for your comments.
> typedef void (*LogicalOutputPluginWriterUpdateProgress) (struct
> LogicalDecodingContext *lr,
> XLogRecPtr Ptr,
> TransactionId xid,
> - bool skipped_xact
> + bool skipped_xact,
> + bool last_write
>
> In this approach, I don't think we need an additional parameter last_write. Let's
> do the work related to keepalive without a parameter, do you see any problem
> with that?
I agree with you. Modify this point.
> I think this patch doesn't take into account that we call
> OutputPluginUpdateProgress() from APIs like pgoutput_commit_txn(). I
> think we should always call the new function update_progress from
> those existing call sites and arrange the function such that when
> called from xact end APIs like pgoutput_commit_txn(), it always call
> OutputPluginUpdateProgress and make changes_count as 0.
Improve it.
Add two new input to function update_progress.(skipped_xact and end_xact).
Modify the function invoke from OutputPluginUpdateProgress to update_progress.
> Also, let's try to evaluate how it impacts lag functionality for large transactions?
I think this patch will not affect lag functionality. It will updates the lag
field of view pg_stat_replication more frequently.
IIUC, when invoking function WalSndUpdateProgress, it will store the lsn of
change and invoking time in lag_tracker. Then when invoking function
ProcessStandbyReplyMessage, it will calculate the lag field according to the
message from subscriber and the information in lag_tracker. This patch does
not modify this logic, but only increases the frequency of invoking.
Please let me know if I understand wrong.
Attach the new patch.
1. Remove the new function input parameters in this patch(parameter last_write
of WalSndUpdateProgress). [suggestion by Amit-San]
2. Also invoke function update_progress in other xact end APIs like
pgoutput_commit_txn. [suggestion by Amit-San]
Regards,
Wang wei
Attachments:
[application/octet-stream] v12-0001-Fix-the-logical-replication-timeout-during-large.patch (7.3K, ../OS3PR01MB627516D7A3A2ED91EEAA510C9EE79@OS3PR01MB6275.jpnprd01.prod.outlook.com/2-v12-0001-Fix-the-logical-replication-timeout-during-large.patch)
download | inline diff:
From 4ec82fac3283ee5a47bd3d65d1ca94206cdbff36 Mon Sep 17 00:00:00 2001
From: wangw <[email protected]>
Date: Wed, 6 Apr 2022 20:40:40 +0800
Subject: [PATCH v12] 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.
---
src/backend/replication/pgoutput/pgoutput.c | 57 ++++++++++++++++++---
src/backend/replication/walsender.c | 12 ++++-
2 files changed, 62 insertions(+), 7 deletions(-)
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 20d0b1e125..04cb43aefc 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, bool skipped_xact, bool end_xact);
/*
* 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);
+ update_progress(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);
+ update_progress(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);
+ update_progress(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);
+ update_progress(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, false, false);
+
if (!is_publishable_relation(relation))
return;
@@ -1611,6 +1614,8 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
Oid *relids;
TransactionId xid = InvalidTransactionId;
+ update_progress(ctx, false, false);
+
/* 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, false, false);
+
if (!data->messages)
return;
@@ -1718,6 +1725,8 @@ pgoutput_sequence(LogicalDecodingContext *ctx,
TransactionId xid = InvalidTransactionId;
RelationSyncEntry *relentry;
+ update_progress(ctx, false, false);
+
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);
+ update_progress(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);
+ update_progress(ctx, false, true);
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn);
OutputPluginWrite(ctx, true);
@@ -2443,3 +2452,39 @@ 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, bool skipped_xact, bool end_xact)
+{
+ static int changes_count = 0;
+
+ if (end_xact)
+ {
+ /* Update progress tracking at xact end. */
+ OutputPluginUpdateProgress(ctx, skipped_xact);
+ 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
+ else if (++changes_count >= CHANGES_THRESHOLD)
+ {
+ OutputPluginUpdateProgress(ctx, skipped_xact);
+ changes_count = 0;
+ }
+}
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 75400a53f2..503ecd13c6 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1461,6 +1461,9 @@ 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,
@@ -1468,6 +1471,7 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId
{
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 ||
+ (now >= TimestampTzPlusMilliseconds(last_reply_timestamp,
+ wal_sender_timeout / 2)))
+ ProcessPendingWrites();
}
/*
--
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: <OS3PR01MB627516D7A3A2ED91EEAA510C9EE79@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