public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: Amit Kapila <[email protected]>
Cc: Masahiko Sawada <[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: Fri, 29 Apr 2022 05:35:58 +0000
Message-ID: <OS3PR01MB62757C3EB48E25D34D54913D9EFC9@OS3PR01MB6275.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAA4eK1Jm7sr=wHXcLNChkPVfjEjKXMUM1oHf-KJncW5gUqvKNQ@mail.gmail.com>
References: <CAA4eK1K6rHbMrKHuqGGULg_31bZmgndd+BL=nT0y3KfMUG6pHw@mail.gmail.com>
<CAA4eK1+KssQYCjKdMd7VfZqd3uPL+yVsS-qX-ocPnGuba3zSGQ@mail.gmail.com>
<OS3PR01MB6275BBCFA72DE68FA863CD0A9EF49@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<OS3PR01MB62754DF6581A3904A78A37CB9EF49@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAA4eK1Jm7sr=wHXcLNChkPVfjEjKXMUM1oHf-KJncW5gUqvKNQ@mail.gmail.com>
On Thur, Apr 28, 2022 at 6:26 PM Amit Kapila <[email protected]> wrote:
> On Thu, Apr 21, 2022 at 3:21 PM [email protected]
> <[email protected]> wrote:
> >
>
> I think it is better to keep the new variable 'end_xact' at the end of
> the struct where it belongs for HEAD. In back branches, we can keep it
> at the place as you have. Apart from that, I have made some cosmetic
> changes and changed a few comments in the attached. Let's use this to
> prepare patches for back-branches.
Thanks for your review and improvement.
I improved the back-branch patches according to your modifications.
Attach the back-branch patches for REL10~REL14.
(Also attach the patch for HEAD, I did not make any changes to this patch.)
BTW, I found Hou-san shared some points. After our discussion, I will update
the patches if required.
Regards,
Wang wei
Attachments:
[application/octet-stream] HEAD_v19-0001-Fix-the-logical-replication-timeout-during-large.patch (12.3K, ../OS3PR01MB62757C3EB48E25D34D54913D9EFC9@OS3PR01MB6275.jpnprd01.prod.outlook.com/2-HEAD_v19-0001-Fix-the-logical-replication-timeout-during-large.patch)
download | inline diff:
From dddfb8efda84ea981315deb529bb0c351d69ee02 Mon Sep 17 00:00:00 2001
From: Amit Kapila <[email protected]>
Date: Thu, 28 Apr 2022 15:19:01 +0530
Subject: [PATCH v19] 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/logical/logical.c | 57 +++++++++++++++++++--
src/backend/replication/pgoutput/pgoutput.c | 6 +++
src/backend/replication/walsender.c | 24 +++++++--
src/include/replication/logical.h | 2 +
4 files changed, 83 insertions(+), 6 deletions(-)
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 788769dd73..985838fc4c 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -669,17 +669,41 @@ OutputPluginWrite(struct LogicalDecodingContext *ctx, bool last_write)
}
/*
- * Update progress tracking (if supported).
+ * Update progress tracking and try to send a keepalive message (if supported).
+ *
+ * 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.
*/
void
OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx,
bool skipped_xact)
{
+ static int changes_count = 0;
+
if (!ctx->update_progress)
return;
- ctx->update_progress(ctx, ctx->write_location, ctx->write_xid,
- skipped_xact);
+ /*
+ * 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 (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD)
+ {
+ ctx->update_progress(ctx, ctx->write_location, ctx->write_xid,
+ skipped_xact);
+ changes_count = 0;
+ }
}
/*
@@ -746,6 +770,7 @@ startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool i
/* set output state */
ctx->accept_writes = false;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.startup_cb(ctx, opt, is_init);
@@ -773,6 +798,7 @@ shutdown_cb_wrapper(LogicalDecodingContext *ctx)
/* set output state */
ctx->accept_writes = false;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.shutdown_cb(ctx);
@@ -808,6 +834,7 @@ begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn)
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->first_lsn;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.begin_cb(ctx, txn);
@@ -839,6 +866,7 @@ commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->end_lsn; /* points to the end of the record */
+ ctx->end_xact = true;
/* do the actual work: call callback */
ctx->callbacks.commit_cb(ctx, txn, commit_lsn);
@@ -879,6 +907,7 @@ begin_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn)
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->first_lsn;
+ ctx->end_xact = false;
/*
* If the plugin supports two-phase commits then begin prepare callback is
@@ -923,6 +952,7 @@ prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->end_lsn; /* points to the end of the record */
+ ctx->end_xact = true;
/*
* If the plugin supports two-phase commits then prepare callback is
@@ -967,6 +997,7 @@ commit_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->end_lsn; /* points to the end of the record */
+ ctx->end_xact = true;
/*
* If the plugin support two-phase commits then commit prepared callback
@@ -1012,6 +1043,7 @@ rollback_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->end_lsn; /* points to the end of the record */
+ ctx->end_xact = true;
/*
* If the plugin support two-phase commits then rollback prepared callback
@@ -1062,6 +1094,8 @@ change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
*/
ctx->write_location = change->lsn;
+ ctx->end_xact = false;
+
ctx->callbacks.change_cb(ctx, txn, relation, change);
/* Pop the error context stack */
@@ -1102,6 +1136,8 @@ truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
*/
ctx->write_location = change->lsn;
+ ctx->end_xact = false;
+
ctx->callbacks.truncate_cb(ctx, txn, nrelations, relations, change);
/* Pop the error context stack */
@@ -1129,6 +1165,7 @@ filter_prepare_cb_wrapper(LogicalDecodingContext *ctx, TransactionId xid,
/* set output state */
ctx->accept_writes = false;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ret = ctx->callbacks.filter_prepare_cb(ctx, xid, gid);
@@ -1159,6 +1196,7 @@ filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
/* set output state */
ctx->accept_writes = false;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id);
@@ -1196,6 +1234,7 @@ message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
ctx->write_location = message_lsn;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix,
@@ -1239,6 +1278,8 @@ stream_start_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
*/
ctx->write_location = first_lsn;
+ ctx->end_xact = false;
+
/* in streaming mode, stream_start_cb is required */
if (ctx->callbacks.stream_start_cb == NULL)
ereport(ERROR,
@@ -1286,6 +1327,8 @@ stream_stop_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
*/
ctx->write_location = last_lsn;
+ ctx->end_xact = false;
+
/* in streaming mode, stream_stop_cb is required */
if (ctx->callbacks.stream_stop_cb == NULL)
ereport(ERROR,
@@ -1325,6 +1368,7 @@ stream_abort_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = abort_lsn;
+ ctx->end_xact = true;
/* in streaming mode, stream_abort_cb is required */
if (ctx->callbacks.stream_abort_cb == NULL)
@@ -1369,6 +1413,7 @@ stream_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->end_lsn;
+ ctx->end_xact = true;
/* in streaming mode with two-phase commits, stream_prepare_cb is required */
if (ctx->callbacks.stream_prepare_cb == NULL)
@@ -1409,6 +1454,7 @@ stream_commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->end_lsn;
+ ctx->end_xact = true;
/* in streaming mode, stream_commit_cb is required */
if (ctx->callbacks.stream_commit_cb == NULL)
@@ -1457,6 +1503,8 @@ stream_change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
*/
ctx->write_location = change->lsn;
+ ctx->end_xact = false;
+
/* in streaming mode, stream_change_cb is required */
if (ctx->callbacks.stream_change_cb == NULL)
ereport(ERROR,
@@ -1501,6 +1549,7 @@ stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
ctx->write_location = message_lsn;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.stream_message_cb(ctx, txn, message_lsn, transactional, prefix,
@@ -1549,6 +1598,8 @@ stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
*/
ctx->write_location = change->lsn;
+ ctx->end_xact = false;
+
ctx->callbacks.stream_truncate_cb(ctx, txn, nrelations, relations, change);
/* Pop the error context stack */
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index b197bfd565..4e2531601f 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -1360,6 +1360,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
TupleTableSlot *old_slot = NULL;
TupleTableSlot *new_slot = NULL;
+ OutputPluginUpdateProgress(ctx, false);
+
if (!is_publishable_relation(relation))
return;
@@ -1592,6 +1594,8 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
Oid *relids;
TransactionId xid = InvalidTransactionId;
+ OutputPluginUpdateProgress(ctx, false);
+
/* Remember the xid for the change in streaming mode. See pgoutput_change. */
if (in_streaming)
xid = change->txn->xid;
@@ -1655,6 +1659,8 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
TransactionId xid = InvalidTransactionId;
+ OutputPluginUpdateProgress(ctx, false);
+
if (!data->messages)
return;
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 63a818140b..79fffe849b 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1482,14 +1482,20 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId
{
static TimestampTz sendTime = 0;
TimestampTz now = GetCurrentTimestamp();
+ bool pending_writes = false;
+ bool end_xact = ctx->end_xact;
/*
* 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;
@@ -1515,8 +1521,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..edadacd589 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -107,6 +107,8 @@ typedef struct LogicalDecodingContext
bool prepared_write;
XLogRecPtr write_location;
TransactionId write_xid;
+ /* Are we processing the end LSN of a transaction? */
+ bool end_xact;
} LogicalDecodingContext;
--
2.28.0.windows.1
[application/octet-stream] REL14_v3-0001-Fix-the-logical-replication-timeout-during-large-.patch (12.5K, ../OS3PR01MB62757C3EB48E25D34D54913D9EFC9@OS3PR01MB6275.jpnprd01.prod.outlook.com/3-REL14_v3-0001-Fix-the-logical-replication-timeout-during-large-.patch)
download | inline diff:
From 7ebc888c03095829b22def22ef782bdd14a7570b Mon Sep 17 00:00:00 2001
From: wangw <[email protected]>
Date: Thu, 21 Apr 2022 09:20:52 +0800
Subject: [PATCH v3] 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. 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/logical/logical.c | 54 ++++++++++++++++++++-
src/backend/replication/pgoutput/pgoutput.c | 6 +++
src/backend/replication/walsender.c | 36 ++++++++++++--
src/include/replication/logical.h | 3 ++
4 files changed, 93 insertions(+), 6 deletions(-)
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index d536a5f3ba..9db7363a22 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -649,15 +649,38 @@ OutputPluginWrite(struct LogicalDecodingContext *ctx, bool last_write)
}
/*
- * Update progress tracking (if supported).
+ * Update progress tracking and try to send a keepalive message (if supported).
+ *
+ * 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 not published.
*/
void
OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx)
{
+ static int changes_count = 0;
+
if (!ctx->update_progress)
return;
- ctx->update_progress(ctx, ctx->write_location, ctx->write_xid);
+ /*
+ * 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 (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD)
+ {
+ ctx->update_progress(ctx, ctx->write_location, ctx->write_xid);
+ changes_count = 0;
+ }
}
/*
@@ -724,6 +747,7 @@ startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool i
/* set output state */
ctx->accept_writes = false;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.startup_cb(ctx, opt, is_init);
@@ -751,6 +775,7 @@ shutdown_cb_wrapper(LogicalDecodingContext *ctx)
/* set output state */
ctx->accept_writes = false;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.shutdown_cb(ctx);
@@ -786,6 +811,7 @@ begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn)
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->first_lsn;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.begin_cb(ctx, txn);
@@ -817,6 +843,7 @@ commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->end_lsn; /* points to the end of the record */
+ ctx->end_xact = true;
/* do the actual work: call callback */
ctx->callbacks.commit_cb(ctx, txn, commit_lsn);
@@ -857,6 +884,7 @@ begin_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn)
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->first_lsn;
+ ctx->end_xact = false;
/*
* If the plugin supports two-phase commits then begin prepare callback is
@@ -901,6 +929,7 @@ prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->end_lsn; /* points to the end of the record */
+ ctx->end_xact = true;
/*
* If the plugin supports two-phase commits then prepare callback is
@@ -945,6 +974,7 @@ commit_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->end_lsn; /* points to the end of the record */
+ ctx->end_xact = true;
/*
* If the plugin support two-phase commits then commit prepared callback
@@ -990,6 +1020,7 @@ rollback_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->end_lsn; /* points to the end of the record */
+ ctx->end_xact = true;
/*
* If the plugin support two-phase commits then rollback prepared callback
@@ -1040,6 +1071,8 @@ change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
*/
ctx->write_location = change->lsn;
+ ctx->end_xact = false;
+
ctx->callbacks.change_cb(ctx, txn, relation, change);
/* Pop the error context stack */
@@ -1080,6 +1113,8 @@ truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
*/
ctx->write_location = change->lsn;
+ ctx->end_xact = false;
+
ctx->callbacks.truncate_cb(ctx, txn, nrelations, relations, change);
/* Pop the error context stack */
@@ -1107,6 +1142,7 @@ filter_prepare_cb_wrapper(LogicalDecodingContext *ctx, TransactionId xid,
/* set output state */
ctx->accept_writes = false;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ret = ctx->callbacks.filter_prepare_cb(ctx, xid, gid);
@@ -1137,6 +1173,7 @@ filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
/* set output state */
ctx->accept_writes = false;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id);
@@ -1174,6 +1211,7 @@ message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
ctx->write_location = message_lsn;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix,
@@ -1217,6 +1255,8 @@ stream_start_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
*/
ctx->write_location = first_lsn;
+ ctx->end_xact = false;
+
/* in streaming mode, stream_start_cb is required */
if (ctx->callbacks.stream_start_cb == NULL)
ereport(ERROR,
@@ -1264,6 +1304,8 @@ stream_stop_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
*/
ctx->write_location = last_lsn;
+ ctx->end_xact = false;
+
/* in streaming mode, stream_stop_cb is required */
if (ctx->callbacks.stream_stop_cb == NULL)
ereport(ERROR,
@@ -1303,6 +1345,7 @@ stream_abort_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = abort_lsn;
+ ctx->end_xact = true;
/* in streaming mode, stream_abort_cb is required */
if (ctx->callbacks.stream_abort_cb == NULL)
@@ -1347,6 +1390,7 @@ stream_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->end_lsn;
+ ctx->end_xact = true;
/* in streaming mode with two-phase commits, stream_prepare_cb is required */
if (ctx->callbacks.stream_prepare_cb == NULL)
@@ -1387,6 +1431,7 @@ stream_commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->end_lsn;
+ ctx->end_xact = true;
/* in streaming mode, stream_commit_cb is required */
if (ctx->callbacks.stream_commit_cb == NULL)
@@ -1435,6 +1480,8 @@ stream_change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
*/
ctx->write_location = change->lsn;
+ ctx->end_xact = false;
+
/* in streaming mode, stream_change_cb is required */
if (ctx->callbacks.stream_change_cb == NULL)
ereport(ERROR,
@@ -1479,6 +1526,7 @@ stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
ctx->write_location = message_lsn;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.stream_message_cb(ctx, txn, message_lsn, transactional, prefix,
@@ -1527,6 +1575,8 @@ stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
*/
ctx->write_location = change->lsn;
+ ctx->end_xact = false;
+
ctx->callbacks.stream_truncate_cb(ctx, txn, nrelations, relations, change);
/* Pop the error context stack */
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 2ea540ce4d..31b585a6fa 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -535,6 +535,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
TransactionId xid = InvalidTransactionId;
Relation ancestor = NULL;
+ OutputPluginUpdateProgress(ctx);
+
if (!is_publishable_relation(relation))
return;
@@ -677,6 +679,8 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
Oid *relids;
TransactionId xid = InvalidTransactionId;
+ OutputPluginUpdateProgress(ctx);
+
/* Remember the xid for the change in streaming mode. See pgoutput_change. */
if (in_streaming)
xid = change->txn->xid;
@@ -735,6 +739,8 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
TransactionId xid = InvalidTransactionId;
+ OutputPluginUpdateProgress(ctx);
+
if (!data->messages)
return;
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 3b245c619f..28a09b3de4 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -240,6 +240,7 @@ static void ProcessStandbyMessage(void);
static void ProcessStandbyReplyMessage(void);
static void ProcessStandbyHSFeedbackMessage(void);
static void ProcessRepliesIfAny(void);
+static void ProcessPendingWrites(void);
static void WalSndKeepalive(bool requestReply);
static void WalSndKeepaliveIfNecessary(void);
static void WalSndCheckTimeOut(void);
@@ -1288,6 +1289,16 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
}
/* If we have pending write here, go to slow path */
+ ProcessPendingWrites();
+}
+
+/*
+ * Wait until there is no pending write. Also process replies from the other
+ * side and check timeouts during that.
+ */
+static void
+ProcessPendingWrites(void)
+{
for (;;)
{
long sleeptime;
@@ -1342,18 +1353,35 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId
{
static TimestampTz sendTime = 0;
TimestampTz now = GetCurrentTimestamp();
+ bool end_xact = ctx->end_xact;
/*
* 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,
+ if (end_xact && TimestampDifferenceExceeds(sendTime, now,
WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS))
- return;
+ {
+ LagTrackerWrite(lsn, now);
+ sendTime = now;
+ }
- LagTrackerWrite(lsn, now);
- sendTime = now;
+ /*
+ * 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 (!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 af551d6f4e..25490f221f 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -49,6 +49,9 @@ typedef struct LogicalDecodingContext
*/
bool fast_forward;
+ /* Are we processing the end LSN of a transaction? */
+ bool end_xact;
+
OutputPluginCallbacks callbacks;
OutputPluginOptions options;
--
2.23.0.windows.1
[application/octet-stream] REL13_v2-0001-Fix-the-logical-replication-timeout-during-large-.patch (8.3K, ../OS3PR01MB62757C3EB48E25D34D54913D9EFC9@OS3PR01MB6275.jpnprd01.prod.outlook.com/4-REL13_v2-0001-Fix-the-logical-replication-timeout-during-large-.patch)
download | inline diff:
From 6312c6b094bdd97e820597fed4074168bceb7e1b Mon Sep 17 00:00:00 2001
From: wangw <[email protected]>
Date: Thu, 21 Apr 2022 15:01:02 +0800
Subject: [PATCH v2] 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. 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/logical/logical.c | 37 +++++++++++++++++++--
src/backend/replication/pgoutput/pgoutput.c | 4 +++
src/backend/replication/walsender.c | 36 +++++++++++++++++---
src/include/replication/logical.h | 3 ++
4 files changed, 74 insertions(+), 6 deletions(-)
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 61902be3b0..79a87e4650 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -546,15 +546,38 @@ OutputPluginWrite(struct LogicalDecodingContext *ctx, bool last_write)
}
/*
- * Update progress tracking (if supported).
+ * Update progress tracking and try to send a keepalive message (if supported).
+ *
+ * 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 not published.
*/
void
OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx)
{
+ static int changes_count = 0;
+
if (!ctx->update_progress)
return;
- ctx->update_progress(ctx, ctx->write_location, ctx->write_xid);
+ /*
+ * 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 (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD)
+ {
+ ctx->update_progress(ctx, ctx->write_location, ctx->write_xid);
+ changes_count = 0;
+ }
}
/*
@@ -622,6 +645,7 @@ startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool i
/* set output state */
ctx->accept_writes = false;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.startup_cb(ctx, opt, is_init);
@@ -649,6 +673,7 @@ shutdown_cb_wrapper(LogicalDecodingContext *ctx)
/* set output state */
ctx->accept_writes = false;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.shutdown_cb(ctx);
@@ -684,6 +709,7 @@ begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn)
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->first_lsn;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.begin_cb(ctx, txn);
@@ -715,6 +741,7 @@ commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->end_lsn; /* points to the end of the record */
+ ctx->end_xact = true;
/* do the actual work: call callback */
ctx->callbacks.commit_cb(ctx, txn, commit_lsn);
@@ -754,6 +781,8 @@ change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
*/
ctx->write_location = change->lsn;
+ ctx->end_xact = false;
+
ctx->callbacks.change_cb(ctx, txn, relation, change);
/* Pop the error context stack */
@@ -794,6 +823,8 @@ truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
*/
ctx->write_location = change->lsn;
+ ctx->end_xact = false;
+
ctx->callbacks.truncate_cb(ctx, txn, nrelations, relations, change);
/* Pop the error context stack */
@@ -820,6 +851,7 @@ filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
/* set output state */
ctx->accept_writes = false;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id);
@@ -857,6 +889,7 @@ message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
ctx->write_location = message_lsn;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix,
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 79735da21a..2ae00a70cb 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -385,6 +385,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
RelationSyncEntry *relentry;
Relation ancestor = NULL;
+ OutputPluginUpdateProgress(ctx);
+
if (!is_publishable_relation(relation))
return;
@@ -514,6 +516,8 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
int nrelids;
Oid *relids;
+ OutputPluginUpdateProgress(ctx);
+
old = MemoryContextSwitchTo(data->context);
relids = palloc0(nrelations * sizeof(Oid));
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index fe0d2ee2ed..e2b4950688 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -240,6 +240,7 @@ static void ProcessStandbyMessage(void);
static void ProcessStandbyReplyMessage(void);
static void ProcessStandbyHSFeedbackMessage(void);
static void ProcessRepliesIfAny(void);
+static void ProcessPendingWrites(void);
static void WalSndKeepalive(bool requestReply);
static void WalSndKeepaliveIfNecessary(void);
static void WalSndCheckTimeOut(void);
@@ -1295,6 +1296,16 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
}
/* If we have pending write here, go to slow path */
+ ProcessPendingWrites();
+}
+
+/*
+ * Wait until there is no pending write. Also process replies from the other
+ * side and check timeouts during that.
+ */
+static void
+ProcessPendingWrites(void)
+{
for (;;)
{
int wakeEvents;
@@ -1354,18 +1365,35 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId
{
static TimestampTz sendTime = 0;
TimestampTz now = GetCurrentTimestamp();
+ bool end_xact = ctx->end_xact;
/*
* 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,
+ if (end_xact && TimestampDifferenceExceeds(sendTime, now,
WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS))
- return;
+ {
+ LagTrackerWrite(lsn, now);
+ sendTime = now;
+ }
- LagTrackerWrite(lsn, now);
- sendTime = now;
+ /*
+ * 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 (!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 c2f2475e5d..14d71fd136 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -49,6 +49,9 @@ typedef struct LogicalDecodingContext
*/
bool fast_forward;
+ /* Are we processing the end LSN of a transaction? */
+ bool end_xact;
+
OutputPluginCallbacks callbacks;
OutputPluginOptions options;
--
2.23.0.windows.1
[application/octet-stream] REL12-REL11_v2-0001-Fix-the-logical-replication-timeout-during-large-.patch (8.2K, ../OS3PR01MB62757C3EB48E25D34D54913D9EFC9@OS3PR01MB6275.jpnprd01.prod.outlook.com/5-REL12-REL11_v2-0001-Fix-the-logical-replication-timeout-during-large-.patch)
download | inline diff:
From 47a620477ff906e8737ef39d2ea607d375a130c1 Mon Sep 17 00:00:00 2001
From: wangw <[email protected]>
Date: Thu, 21 Apr 2022 15:06:07 +0800
Subject: [PATCH v2] 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. 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/logical/logical.c | 37 +++++++++++++++++++--
src/backend/replication/pgoutput/pgoutput.c | 4 +++
src/backend/replication/walsender.c | 36 +++++++++++++++++---
src/include/replication/logical.h | 3 ++
4 files changed, 74 insertions(+), 6 deletions(-)
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index a0fc480646..6884f62d95 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -548,15 +548,38 @@ OutputPluginWrite(struct LogicalDecodingContext *ctx, bool last_write)
}
/*
- * Update progress tracking (if supported).
+ * Update progress tracking and try to send a keepalive message (if supported).
+ *
+ * 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 not published.
*/
void
OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx)
{
+ static int changes_count = 0;
+
if (!ctx->update_progress)
return;
- ctx->update_progress(ctx, ctx->write_location, ctx->write_xid);
+ /*
+ * 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 (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD)
+ {
+ ctx->update_progress(ctx, ctx->write_location, ctx->write_xid);
+ changes_count = 0;
+ }
}
/*
@@ -624,6 +647,7 @@ startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool i
/* set output state */
ctx->accept_writes = false;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.startup_cb(ctx, opt, is_init);
@@ -651,6 +675,7 @@ shutdown_cb_wrapper(LogicalDecodingContext *ctx)
/* set output state */
ctx->accept_writes = false;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.shutdown_cb(ctx);
@@ -686,6 +711,7 @@ begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn)
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->first_lsn;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.begin_cb(ctx, txn);
@@ -717,6 +743,7 @@ commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->end_lsn; /* points to the end of the record */
+ ctx->end_xact = true;
/* do the actual work: call callback */
ctx->callbacks.commit_cb(ctx, txn, commit_lsn);
@@ -756,6 +783,8 @@ change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
*/
ctx->write_location = change->lsn;
+ ctx->end_xact = false;
+
ctx->callbacks.change_cb(ctx, txn, relation, change);
/* Pop the error context stack */
@@ -796,6 +825,8 @@ truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
*/
ctx->write_location = change->lsn;
+ ctx->end_xact = false;
+
ctx->callbacks.truncate_cb(ctx, txn, nrelations, relations, change);
/* Pop the error context stack */
@@ -822,6 +853,7 @@ filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
/* set output state */
ctx->accept_writes = false;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id);
@@ -859,6 +891,7 @@ message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
ctx->write_location = message_lsn;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix,
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index d317fd7006..cff24fb44f 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -309,6 +309,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
MemoryContext old;
RelationSyncEntry *relentry;
+ OutputPluginUpdateProgress(ctx);
+
if (!is_publishable_relation(relation))
return;
@@ -389,6 +391,8 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
int nrelids;
Oid *relids;
+ OutputPluginUpdateProgress(ctx);
+
old = MemoryContextSwitchTo(data->context);
relids = palloc0(nrelations * sizeof(Oid));
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 3698135e49..122c365e29 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -244,6 +244,7 @@ static void ProcessStandbyMessage(void);
static void ProcessStandbyReplyMessage(void);
static void ProcessStandbyHSFeedbackMessage(void);
static void ProcessRepliesIfAny(void);
+static void ProcessPendingWrites(void);
static void WalSndKeepalive(bool requestReply);
static void WalSndKeepaliveIfNecessary(void);
static void WalSndCheckTimeOut(void);
@@ -1214,6 +1215,16 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
}
/* If we have pending write here, go to slow path */
+ ProcessPendingWrites();
+}
+
+/*
+ * Wait until there is no pending write. Also process replies from the other
+ * side and check timeouts during that.
+ */
+static void
+ProcessPendingWrites(void)
+{
for (;;)
{
int wakeEvents;
@@ -1273,18 +1284,35 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId
{
static TimestampTz sendTime = 0;
TimestampTz now = GetCurrentTimestamp();
+ bool end_xact = ctx->end_xact;
/*
* 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,
+ if (end_xact && TimestampDifferenceExceeds(sendTime, now,
WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS))
- return;
+ {
+ LagTrackerWrite(lsn, now);
+ sendTime = now;
+ }
- LagTrackerWrite(lsn, now);
- sendTime = now;
+ /*
+ * 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 (!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 31c796b765..718080d54a 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -50,6 +50,9 @@ typedef struct LogicalDecodingContext
*/
bool fast_forward;
+ /* Are we processing the end LSN of a transaction? */
+ bool end_xact;
+
OutputPluginCallbacks callbacks;
OutputPluginOptions options;
--
2.23.0.windows.1
[application/octet-stream] REL10_v2-0001-Fix-the-logical-replication-timeout-during-large-.patch (7.7K, ../OS3PR01MB62757C3EB48E25D34D54913D9EFC9@OS3PR01MB6275.jpnprd01.prod.outlook.com/6-REL10_v2-0001-Fix-the-logical-replication-timeout-during-large-.patch)
download | inline diff:
From c4fc9251b07cd3a528bf56d56a1a1405c9d4b513 Mon Sep 17 00:00:00 2001
From: wangw <[email protected]>
Date: Thu, 21 Apr 2022 15:53:29 +0800
Subject: [PATCH v2] 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. 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/logical/logical.c | 35 ++++++++++++++++++--
src/backend/replication/pgoutput/pgoutput.c | 2 ++
src/backend/replication/walsender.c | 36 ++++++++++++++++++---
src/include/replication/logical.h | 1 +
4 files changed, 68 insertions(+), 6 deletions(-)
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 1851a626f9..b51614e3fb 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -515,15 +515,38 @@ OutputPluginWrite(struct LogicalDecodingContext *ctx, bool last_write)
}
/*
- * Update progress tracking (if supported).
+ * Update progress tracking and try to send a keepalive message (if supported).
+ *
+ * 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 not published.
*/
void
OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx)
{
+ static int changes_count = 0;
+
if (!ctx->update_progress)
return;
- ctx->update_progress(ctx, ctx->write_location, ctx->write_xid);
+ /*
+ * 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 (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD)
+ {
+ ctx->update_progress(ctx, ctx->write_location, ctx->write_xid);
+ changes_count = 0;
+ }
}
/*
@@ -589,6 +612,7 @@ startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool i
/* set output state */
ctx->accept_writes = false;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.startup_cb(ctx, opt, is_init);
@@ -614,6 +638,7 @@ shutdown_cb_wrapper(LogicalDecodingContext *ctx)
/* set output state */
ctx->accept_writes = false;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.shutdown_cb(ctx);
@@ -647,6 +672,7 @@ begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn)
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->first_lsn;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.begin_cb(ctx, txn);
@@ -676,6 +702,7 @@ commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn->xid;
ctx->write_location = txn->end_lsn; /* points to the end of the record */
+ ctx->end_xact = true;
/* do the actual work: call callback */
ctx->callbacks.commit_cb(ctx, txn, commit_lsn);
@@ -713,6 +740,8 @@ change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
*/
ctx->write_location = change->lsn;
+ ctx->end_xact = false;
+
ctx->callbacks.change_cb(ctx, txn, relation, change);
/* Pop the error context stack */
@@ -737,6 +766,7 @@ filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
/* set output state */
ctx->accept_writes = false;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id);
@@ -772,6 +802,7 @@ message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
ctx->accept_writes = true;
ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
ctx->write_location = message_lsn;
+ ctx->end_xact = false;
/* do the actual work: call callback */
ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix,
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 529a5c0b48..12db2ad4fd 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -264,6 +264,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
MemoryContext old;
RelationSyncEntry *relentry;
+ OutputPluginUpdateProgress(ctx);
+
if (!is_publishable_relation(relation))
return;
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 3c9b9f5376..4dbe620fe6 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -243,6 +243,7 @@ static void ProcessStandbyMessage(void);
static void ProcessStandbyReplyMessage(void);
static void ProcessStandbyHSFeedbackMessage(void);
static void ProcessRepliesIfAny(void);
+static void ProcessPendingWrites(void);
static void WalSndKeepalive(bool requestReply);
static void WalSndKeepaliveIfNecessary(void);
static void WalSndCheckTimeOut(void);
@@ -1190,6 +1191,16 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
}
/* If we have pending write here, go to slow path */
+ ProcessPendingWrites();
+}
+
+/*
+ * Wait until there is no pending write. Also process replies from the other
+ * side and check timeouts during that.
+ */
+static void
+ProcessPendingWrites(void)
+{
for (;;)
{
int wakeEvents;
@@ -1256,18 +1267,35 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId
{
static TimestampTz sendTime = 0;
TimestampTz now = GetCurrentTimestamp();
+ bool end_xact = ctx->end_xact;
/*
* 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,
+ if (end_xact && TimestampDifferenceExceeds(sendTime, now,
WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS))
- return;
+ {
+ LagTrackerWrite(lsn, now);
+ sendTime = now;
+ }
- LagTrackerWrite(lsn, now);
- sendTime = now;
+ /*
+ * 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 (!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 7f0e0fa881..b25a935bc0 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -80,6 +80,7 @@ typedef struct LogicalDecodingContext
*/
bool accept_writes;
bool prepared_write;
+ bool end_xact;
XLogRecPtr write_location;
TransactionId write_xid;
} LogicalDecodingContext;
--
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: <OS3PR01MB62757C3EB48E25D34D54913D9EFC9@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