From 844ae008dbd3f9bc5257b421f0b43bf14ca268ff Mon Sep 17 00:00:00 2001 From: wangw Date: Wed, 30 Mar 2022 15:27:22 +0800 Subject: [PATCH v10] 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 skipping certain threshold of changes. --- src/backend/replication/logical/logical.c | 7 +- src/backend/replication/pgoutput/pgoutput.c | 86 +++++++++++++++++++-- src/backend/replication/walsender.c | 16 +++- src/include/replication/logical.h | 3 +- src/include/replication/output_plugin.h | 3 +- 5 files changed, 101 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..6ce4920d49 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 last_write); /* * 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, @@ -1378,9 +1379,13 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, ReorderBufferChangeType action = change->action; TupleTableSlot *old_slot = NULL; TupleTableSlot *new_slot = NULL; + bool change_sent = false; if (!is_publishable_relation(relation)) + { + update_progress(ctx, false); return; + } /* * Remember the xid for the change in streaming mode. We need to send xid @@ -1398,15 +1403,24 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, { case REORDER_BUFFER_CHANGE_INSERT: if (!relentry->pubactions.pubinsert) + { + update_progress(ctx, false); return; + } break; case REORDER_BUFFER_CHANGE_UPDATE: if (!relentry->pubactions.pubupdate) + { + update_progress(ctx, false); return; + } break; case REORDER_BUFFER_CHANGE_DELETE: if (!relentry->pubactions.pubdelete) + { + update_progress(ctx, false); return; + } break; default: Assert(false); @@ -1465,6 +1479,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, logicalrep_write_insert(ctx->out, xid, targetrel, new_slot, data->binary, relentry->columns); OutputPluginWrite(ctx, true); + change_sent = true; break; case REORDER_BUFFER_CHANGE_UPDATE: if (change->data.tp.oldtuple) @@ -1538,6 +1553,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, } OutputPluginWrite(ctx, true); + change_sent = true; break; case REORDER_BUFFER_CHANGE_DELETE: if (change->data.tp.oldtuple) @@ -1579,6 +1595,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, logicalrep_write_delete(ctx->out, xid, targetrel, old_slot, data->binary); OutputPluginWrite(ctx, true); + change_sent = true; } else elog(DEBUG1, "didn't send DELETE change because of missing oldtuple"); @@ -1587,6 +1604,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Assert(false); } + update_progress(ctx, change_sent); + if (RelationIsValid(ancestor)) { RelationClose(ancestor); @@ -1660,7 +1679,10 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, change->data.truncate.cascade, change->data.truncate.restart_seqs); OutputPluginWrite(ctx, true); + update_progress(ctx, true); } + else + update_progress(ctx, false); MemoryContextSwitchTo(old); MemoryContextReset(data->context); @@ -1675,7 +1697,10 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, TransactionId xid = InvalidTransactionId; if (!data->messages) + { + update_progress(ctx, false); return; + } /* * Remember the xid for the message in streaming mode. See @@ -1706,6 +1731,7 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, sz, message); OutputPluginWrite(ctx, true); + update_progress(ctx, true); } static void @@ -1719,10 +1745,16 @@ pgoutput_sequence(LogicalDecodingContext *ctx, RelationSyncEntry *relentry; if (!data->sequences) + { + update_progress(ctx, false); return; + } if (!is_publishable_relation(relation)) + { + update_progress(ctx, false); return; + } /* * Remember the xid for the message in streaming mode. See @@ -1739,7 +1771,10 @@ pgoutput_sequence(LogicalDecodingContext *ctx, * We handle just REORDER_BUFFER_CHANGE_SEQUENCE here. */ if (!relentry->pubactions.pubsequence) + { + update_progress(ctx, false); return; + } /* * Output BEGIN if we haven't yet. Avoid for non-transactional @@ -1764,6 +1799,7 @@ pgoutput_sequence(LogicalDecodingContext *ctx, log_cnt, is_called); OutputPluginWrite(ctx, true); + update_progress(ctx, true); } /* @@ -1924,7 +1960,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 +1981,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 +2479,41 @@ send_repl_origin(LogicalDecodingContext *ctx, RepOriginId origin_id, } } } + +/* + * Try to update progress and send a keepalive message if too many changes were + * skipped. + * + * 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 last_write) +{ + static int skipped_changes_count = 0; + + /* reset the skipped count after sending a change */ + if (last_write) + { + skipped_changes_count = 0; + return; + } + + /* + * After continuously skipping SKIPPED_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 skipping each change as that can have overhead. Testing reveals + * that there is no noticeable overhead in doing it after continuously + * skipping 100 or so changes. + */ +#define SKIPPED_CHANGES_THRESHOLD 100 + + if (++skipped_changes_count >= SKIPPED_CHANGES_THRESHOLD) + { + OutputPluginUpdateProgress(ctx, false, false); + skipped_changes_count = 0; + } +} diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 75400a53f2..c5f1d36172 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 the last write is skipped 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.18.4