public inbox for [email protected]help / color / mirror / Atom feed
[PATCH v11 1/5] Correctly update contfol file at the end of archive recovery 35+ messages / 9 participants [nested] [flat]
* [PATCH v11 1/5] Correctly update contfol file at the end of archive recovery @ 2022-03-04 04:18 Kyotaro Horiguchi <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Kyotaro Horiguchi @ 2022-03-04 04:18 UTC (permalink / raw) CreateRestartPoint runs WAL file cleanup basing on the checkpoint just have finished in the function. If the database has exited DB_IN_ARCHIVE_RECOVERY state when the function is going to update control file, the function refrains from updating the file at all then proceeds to WAL cleanup having the latest REDO LSN, which is now inconsistent with the control file. As the result, the succeeding cleanup procedure overly removes WAL files against the control file and leaves unrecoverable database until the next checkpoint finishes. Along with that fix, we remove a dead code path for the case some other process ran a simultaneous checkpoint. It seems like just a preventive measure but it's no longer useful because we are sure that checkpoint is performed only by checkpointer except single process mode. --- src/backend/access/transam/xlog.c | 72 ++++++++++++++++++++----------- 1 file changed, 47 insertions(+), 25 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 0d2bd7a357..3987aa81de 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -6899,6 +6899,9 @@ CreateRestartPoint(int flags) XLogSegNo _logSegNo; TimestampTz xtime; + /* we don't assume concurrent checkpoint/restartpoint to run */ + Assert (!IsUnderPostmaster || MyBackendType == B_CHECKPOINTER); + /* Get a local copy of the last safe checkpoint record. */ SpinLockAcquire(&XLogCtl->info_lck); lastCheckPointRecPtr = XLogCtl->lastCheckPointRecPtr; @@ -6964,7 +6967,7 @@ CreateRestartPoint(int flags) /* Also update the info_lck-protected copy */ SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->RedoRecPtr = lastCheckPoint.redo; + XLogCtl->RedoRecPtr = RedoRecPtr; SpinLockRelease(&XLogCtl->info_lck); /* @@ -6983,7 +6986,10 @@ CreateRestartPoint(int flags) /* Update the process title */ update_checkpoint_display(flags, true, false); - CheckPointGuts(lastCheckPoint.redo, flags); + CheckPointGuts(RedoRecPtr, flags); + + /* Update pg_control */ + LWLockAcquire(ControlFileLock, LW_EXCLUSIVE); /* * Remember the prior checkpoint's redo ptr for @@ -6991,30 +6997,29 @@ CreateRestartPoint(int flags) */ PriorRedoPtr = ControlFile->checkPointCopy.redo; + Assert (PriorRedoPtr < RedoRecPtr); + + ControlFile->checkPoint = lastCheckPointRecPtr; + ControlFile->checkPointCopy = lastCheckPoint; + + /* Update control file using current time */ + ControlFile->time = (pg_time_t) time(NULL); + /* - * Update pg_control, using current time. Check that it still shows - * DB_IN_ARCHIVE_RECOVERY state and an older checkpoint, else do nothing; - * this is a quick hack to make sure nothing really bad happens if somehow - * we get here after the end-of-recovery checkpoint. + * Ensure minRecoveryPoint is past the checkpoint record while archive + * recovery is still ongoing. Normally, this will have happened already + * while writing out dirty buffers, but not necessarily - e.g. because no + * buffers were dirtied. We do this because a non-exclusive base backup + * uses minRecoveryPoint to determine which WAL files must be included in + * the backup, and the file (or files) containing the checkpoint record + * must be included, at a minimum. Note that for an ordinary restart of + * recovery there's no value in having the minimum recovery point any + * earlier than this anyway, because redo will begin just after the + * checkpoint record. This is a quick hack to make sure nothing really bad + * happens if somehow we get here after the end-of-recovery checkpoint. */ - LWLockAcquire(ControlFileLock, LW_EXCLUSIVE); - if (ControlFile->state == DB_IN_ARCHIVE_RECOVERY && - ControlFile->checkPointCopy.redo < lastCheckPoint.redo) + if (ControlFile->state == DB_IN_ARCHIVE_RECOVERY) { - ControlFile->checkPoint = lastCheckPointRecPtr; - ControlFile->checkPointCopy = lastCheckPoint; - - /* - * Ensure minRecoveryPoint is past the checkpoint record. Normally, - * this will have happened already while writing out dirty buffers, - * but not necessarily - e.g. because no buffers were dirtied. We do - * this because a non-exclusive base backup uses minRecoveryPoint to - * determine which WAL files must be included in the backup, and the - * file (or files) containing the checkpoint record must be included, - * at a minimum. Note that for an ordinary restart of recovery there's - * no value in having the minimum recovery point any earlier than this - * anyway, because redo will begin just after the checkpoint record. - */ if (ControlFile->minRecoveryPoint < lastCheckPointEndPtr) { ControlFile->minRecoveryPoint = lastCheckPointEndPtr; @@ -7026,8 +7031,25 @@ CreateRestartPoint(int flags) } if (flags & CHECKPOINT_IS_SHUTDOWN) ControlFile->state = DB_SHUTDOWNED_IN_RECOVERY; - UpdateControlFile(); } + else + { + /* recovery mode is not supposed to end during shutdown restartpoint */ + Assert((flags & CHECKPOINT_IS_SHUTDOWN) == 0); + + /* + * Aarchive recovery has ended. Crash recovery ever after should + * always recover to the end of WAL + */ + ControlFile->minRecoveryPoint = InvalidXLogRecPtr; + ControlFile->minRecoveryPointTLI = 0; + + /* also update local copy */ + LocalMinRecoveryPoint = InvalidXLogRecPtr; + LocalMinRecoveryPointTLI = 0; + } + + UpdateControlFile(); LWLockRelease(ControlFileLock); /* @@ -7104,7 +7126,7 @@ CreateRestartPoint(int flags) xtime = GetLatestXTime(); ereport((log_checkpoints ? LOG : DEBUG2), (errmsg("recovery restart point at %X/%X", - LSN_FORMAT_ARGS(lastCheckPoint.redo)), + LSN_FORMAT_ARGS(RedoRecPtr)), xtime ? errdetail("Last completed transaction was at log time %s.", timestamptz_to_str(xtime)) : 0)); -- 2.27.0 ----Next_Part(Fri_Mar__4_14_10_38_2022_481)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v11-0002-Add-checkpoint-and-redo-LSN-to-LogCheckpointEnd-.patch" ^ permalink raw reply [nested|flat] 35+ messages in thread
* RE: Logical replication timeout problem @ 2022-03-22 01:55 [email protected] <[email protected]> 0 siblings, 1 reply; 35+ messages in thread From: [email protected] @ 2022-03-22 01:55 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Masahiko Sawada <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Mon, Mar 21, 2022 at 1:31 PM Amit Kapila <[email protected]> wrote: > Thanks for your comments. > On Fri, Mar 18, 2022 at 4:20 PM Amit Kapila <[email protected]> wrote: > > > > On Fri, Mar 18, 2022 at 10:43 AM [email protected] > > <[email protected]> wrote: > > > > > > On Thu, Mar 17, 2022 at 7:52 PM Masahiko Sawada > <[email protected]> wrote: > > > > > > > > > > Attach the new patch. > > > > > > > * > > case REORDER_BUFFER_CHANGE_INVALIDATION: > > - /* Execute the invalidation messages locally */ > > - ReorderBufferExecuteInvalidations( > > - change->data.inval.ninvalidations, > > - change->data.inval.invalidations); > > - break; > > + { > > + LogicalDecodingContext *ctx = rb->private_data; > > + > > + /* Try to send a keepalive message. */ > > + UpdateProgress(ctx, true); > > > > Calling UpdateProgress() here appears adhoc to me especially because > > it calls OutputPluginUpdateProgress which appears to be called only > > from plugin API. Am, I missing something? Also why the same handling > > is missed in other similar messages like > > REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID where we don't call > any > > plug-in API? Yes, you are right. And I invoke in case REORDER_BUFFER_CHANGE_INVALIDATION because I think every DDL will modify the catalog then get into this case. So I only invoke function UpdateProgress here to handle DDL. > > I am not sure what is a good way to achieve this but one idea that > > occurred to me was shall we invent a new callback > > ReorderBufferSkipChangeCB similar to ReorderBufferApplyChangeCB and > > then pgoutput can register its API where we can have the logic similar > > to what you have in UpdateProgress()? If we do so, then all the > > cuurent callers of UpdateProgress in pgoutput can also call that API. > > What do you think? > > > Another idea could be that we leave the DDL case for now as anyway > there is very less chance of timeout for skipping DDLs and we may > later need to even backpatch this bug-fix which would be another > reason to not make such invasive changes. We can handle the DDL case > if required separately. Yes, I think a new callback function would be nice. Yes, as you said, maybe we could fix the usecase that found the problem in the first place. Then make further modifications on the master branch. Modify the patch. Currently only DML related code remains. > > * Why don't you have a quick exit like below code in WalSndWriteData? > > /* Try taking fast path unless we get too close to walsender timeout. */ if (now > > < TimestampTzPlusMilliseconds(last_reply_timestamp, > > wal_sender_timeout / 2) && > > !pq_is_send_pending()) > > { > > return; > > } Fixed. I missed this so adding it in the new patch. > > * Can we rename variable 'is_send' to 'change_sent'? Improve the the name of this variable.(From 'is_send' to 'change_sent') Attach the new patch. [suggestion by Amit-San.] 1. Remove DDL related code. Handle the DDL case later separately if need. 2. Fix a missing.(In function WalSndUpdateProgress) 3. Improve variable names. (From 'is_send' to 'change_sent') 4. Fix some comments.(Above and inside the function WalSndUpdateProgress.) Regards, Wang wei Attachments: [application/octet-stream] v4-0001-Fix-the-timeout-of-subscriber-in-long-transaction.patch (11.7K, ../../OS3PR01MB62756599B78FA7D4C908109A9E179@OS3PR01MB6275.jpnprd01.prod.outlook.com/2-v4-0001-Fix-the-timeout-of-subscriber-in-long-transaction.patch) download | inline diff: From 19d39a500e14af67bd987a4e2d1826aa4b8efc46 Mon Sep 17 00:00:00 2001 From: wangw <[email protected]> Date: Mon, 21 Mar 2022 14:04:10 +0800 Subject: [PATCH v4] Fix the timeout of subscriber in long transactions. 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 (say because the table modified in the transaction is not published) and then subscriber will timeout. So in this case, send keepalive message to the subscriber. --- src/backend/replication/logical/logical.c | 45 +++++++++++++++++++-- src/backend/replication/pgoutput/pgoutput.c | 38 ++++++++++++++--- src/backend/replication/walsender.c | 41 ++++++++++++++++--- src/include/replication/logical.h | 4 +- src/include/replication/output_plugin.h | 2 +- 5 files changed, 113 insertions(+), 17 deletions(-) diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 934aa13f2d..ae7ff14c95 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -680,15 +680,15 @@ 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) +OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx, bool send_keep_alive) { if (!ctx->update_progress) return; - ctx->update_progress(ctx, ctx->write_location, ctx->write_xid); + ctx->update_progress(ctx, ctx->write_location, ctx->write_xid, send_keep_alive); } /* @@ -1930,3 +1930,42 @@ UpdateDecodingStats(LogicalDecodingContext *ctx) rb->totalTxns = 0; rb->totalBytes = 0; } + +/* + * Try to send a keepalive message if too many changes were skipped. + * + * When we loop through changes in a transaction(see ReorderBufferProcessTXN), + * if no message is sent to standby for a long time during a large transaction, + * we should send a keepalive message to ensure that the standby will not + * timeout. + */ +void +UpdateProgress(LogicalDecodingContext *ctx, bool skipped) +{ + static int skipped_changes_count = 0; + + /* + * skipped_changes_count is reset when processing changes that do not + * need to be skipped. + */ + if (!skipped) + { + skipped_changes_count = 0; + return; + } + + /* + * After continuously skipping SKIPPED_CHANGES_THRESHOLD changes, try to send a + * keepalive message. + */ + #define SKIPPED_CHANGES_THRESHOLD 100 + + if (++skipped_changes_count >= SKIPPED_CHANGES_THRESHOLD) + { + /* Try to send a keepalive message. */ + OutputPluginUpdateProgress(ctx, true); + + /* After trying to send a keepalive message, reset the flag. */ + skipped_changes_count = 0; + } +} diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 5fddab3a3d..c9d931888d 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -475,7 +475,7 @@ static void pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit(ctx->out, txn, commit_lsn); @@ -506,7 +506,7 @@ static void pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_prepare(ctx->out, txn, prepare_lsn); @@ -520,7 +520,7 @@ static void pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit_prepared(ctx->out, txn, commit_lsn); @@ -536,7 +536,7 @@ pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_rollback_prepared(ctx->out, txn, prepare_end_lsn, @@ -1149,9 +1149,14 @@ 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)) + { + /* Try to send a keepalive message. */ + UpdateProgress(ctx, true); return; + } /* * Remember the xid for the change in streaming mode. We need to send xid @@ -1169,15 +1174,27 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, { case REORDER_BUFFER_CHANGE_INSERT: if (!relentry->pubactions.pubinsert) + { + /* Try to send a keepalive message. */ + UpdateProgress(ctx, true); return; + } break; case REORDER_BUFFER_CHANGE_UPDATE: if (!relentry->pubactions.pubupdate) + { + /* Try to send a keepalive message. */ + UpdateProgress(ctx, true); return; + } break; case REORDER_BUFFER_CHANGE_DELETE: if (!relentry->pubactions.pubdelete) + { + /* Try to send a keepalive message. */ + UpdateProgress(ctx, true); return; + } break; default: Assert(false); @@ -1226,6 +1243,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, logicalrep_write_insert(ctx->out, xid, targetrel, new_slot, data->binary); OutputPluginWrite(ctx, true); + change_sent = true; break; case REORDER_BUFFER_CHANGE_UPDATE: if (change->data.tp.oldtuple) @@ -1293,6 +1311,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, } OutputPluginWrite(ctx, true); + change_sent = true; break; case REORDER_BUFFER_CHANGE_DELETE: if (change->data.tp.oldtuple) @@ -1330,6 +1349,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"); @@ -1338,6 +1358,12 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Assert(false); } + /* + * Reset the counter for skipped changes if change_sent is true, otherwise try to + * send a keepalive message. + */ + UpdateProgress(ctx, !change_sent); + if (RelationIsValid(ancestor)) { RelationClose(ancestor); @@ -1598,7 +1624,7 @@ pgoutput_stream_commit(struct LogicalDecodingContext *ctx, Assert(!in_streaming); Assert(rbtxn_is_streamed(txn)); - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_commit(ctx->out, txn, commit_lsn); @@ -1619,7 +1645,7 @@ pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx, { Assert(rbtxn_is_streamed(txn)); - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn); OutputPluginWrite(ctx, true); diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 2d0292a092..82daaa06a2 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -249,7 +249,7 @@ static long WalSndComputeSleeptime(TimestampTz now); 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); +static void WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool send_keep_alive); static XLogRecPtr WalSndWaitForWal(XLogRecPtr loc); static void LagTrackerWrite(XLogRecPtr lsn, TimestampTz local_flush_time); static TimeOffset LagTrackerRead(int head, XLogRecPtr lsn, TimestampTz now); @@ -1446,25 +1446,54 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, /* * LogicalDecodingContext 'update_progress' callback. * * Write the current position to the lag tracker (see XLogSendPhysical). + * Try to send a keepalive message to standby if send_keep_alive is true. */ static void -WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid) +WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool send_keep_alive) { - static TimestampTz sendTime = 0; + static TimestampTz trackTime = 0; TimestampTz now = GetCurrentTimestamp(); + if (send_keep_alive) + { + /* + * If the standby does not receive any message from the primary for + * more than (wal_receiver_timeout / 2), the standby will send a + * message requesting a reply to the primary. If receive this message, + * reply immediately to avoid timeout. + */ + + if (now < TimestampTzPlusMilliseconds(last_reply_timestamp, + wal_sender_timeout / 2) && + !pq_is_send_pending()) + return; + + /* Check for input from the client. */ + ProcessRepliesIfAny(); + + /* die if timeout was reached */ + WalSndCheckTimeOut(); + + /* Send keepalive if the time has come */ + WalSndKeepaliveIfNecessary(); + + /* Try to flush pending output to the client */ + if (pq_flush_if_writable() != 0) + WalSndShutdown(); + } + /* * Track lag no more than once per WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS to * avoid flooding the lag tracker when we commit frequently. */ #define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 1000 - if (!TimestampDifferenceExceeds(sendTime, now, + if (!TimestampDifferenceExceeds(trackTime, now, WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) return; LagTrackerWrite(lsn, now); - sendTime = now; + trackTime = now; } /* diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h index 1097cc9799..1d8ab2a56b 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -26,7 +26,8 @@ typedef LogicalOutputPluginWriterWrite LogicalOutputPluginWriterPrepareWrite; typedef void (*LogicalOutputPluginWriterUpdateProgress) (struct LogicalDecodingContext *lr, XLogRecPtr Ptr, - TransactionId xid + TransactionId xid, + bool send_keep_alive ); typedef struct LogicalDecodingContext @@ -140,5 +141,6 @@ extern bool filter_prepare_cb_wrapper(LogicalDecodingContext *ctx, extern bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id); extern void ResetLogicalStreamingState(void); extern void UpdateDecodingStats(LogicalDecodingContext *ctx); +extern void UpdateProgress(LogicalDecodingContext *ctx, bool skipped); #endif diff --git a/src/include/replication/output_plugin.h b/src/include/replication/output_plugin.h index a16bebf76c..ed802b58ef 100644 --- a/src/include/replication/output_plugin.h +++ b/src/include/replication/output_plugin.h @@ -270,6 +270,6 @@ 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); +extern void OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx, bool send_keep_alive); #endif /* OUTPUT_PLUGIN_H */ -- 2.27.0 ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2022-03-24 10:32 Amit Kapila <[email protected]> parent: [email protected] <[email protected]> 0 siblings, 2 replies; 35+ messages in thread From: Amit Kapila @ 2022-03-24 10:32 UTC (permalink / raw) To: [email protected] <[email protected]>; +Cc: Masahiko Sawada <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Tue, Mar 22, 2022 at 7:25 AM [email protected] <[email protected]> wrote: > > Attach the new patch. > It seems by mistake you have removed the changes from pgoutput_message and pgoutput_truncate functions. I have added those back. Additionally, I made a few other changes: (a) moved the function UpdateProgress to pgoutput.c as it is not used outside it, (b) change the new parameter in plugin API from 'send_keep_alive' to 'last_write' to make it look similar to WalSndPrepareWrite and WalSndWriteData, (c) made a number of changes in WalSndUpdateProgress API, it is better to move keep-alive code after lag track code because we do process replies at that time and there it will compute the lag; (d) changed/added comments in the code. Do let me know what you think of the attached? -- With Regards, Amit Kapila. Attachments: [application/octet-stream] v5-0001-Fix-the-logical-replication-timeout-during-large-.patch (12.6K, ../../CAA4eK1JKZrOfCLMVZYfUnm_wfxucDjOzeGwQy3hJ_U9Y3AV05Q@mail.gmail.com/2-v5-0001-Fix-the-logical-replication-timeout-during-large-.patch) download | inline diff: From 1b7b2ccb8cbb0ffb79b806815b5e3d8ef8b29b2f Mon Sep 17 00:00:00 2001 From: wangw <[email protected]> Date: Mon, 21 Mar 2022 14:04:10 +0800 Subject: [PATCH v5] 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 | 6 +-- src/backend/replication/pgoutput/pgoutput.c | 78 ++++++++++++++++++++++++++--- src/backend/replication/walsender.c | 46 ++++++++++++++--- src/include/replication/logical.h | 3 +- src/include/replication/output_plugin.h | 2 +- 5 files changed, 117 insertions(+), 18 deletions(-) diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 934aa13..922b16c 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -680,15 +680,15 @@ 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) +OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx, bool last_write) { if (!ctx->update_progress) return; - ctx->update_progress(ctx, ctx->write_location, ctx->write_xid); + ctx->update_progress(ctx, ctx->write_location, ctx->write_xid, last_write); } /* diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 5fddab3..0186871 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -89,6 +89,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", @@ -475,7 +476,7 @@ static void pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit(ctx->out, txn, commit_lsn); @@ -506,7 +507,7 @@ static void pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_prepare(ctx->out, txn, prepare_lsn); @@ -520,7 +521,7 @@ static void pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit_prepared(ctx->out, txn, commit_lsn); @@ -536,7 +537,7 @@ pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_rollback_prepared(ctx->out, txn, prepare_end_lsn, @@ -1149,9 +1150,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 @@ -1169,15 +1174,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); @@ -1226,6 +1240,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, logicalrep_write_insert(ctx->out, xid, targetrel, new_slot, data->binary); OutputPluginWrite(ctx, true); + change_sent = true; break; case REORDER_BUFFER_CHANGE_UPDATE: if (change->data.tp.oldtuple) @@ -1293,6 +1308,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, } OutputPluginWrite(ctx, true); + change_sent = true; break; case REORDER_BUFFER_CHANGE_DELETE: if (change->data.tp.oldtuple) @@ -1330,6 +1346,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"); @@ -1338,6 +1355,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Assert(false); } + update_progress(ctx, change_sent); + if (RelationIsValid(ancestor)) { RelationClose(ancestor); @@ -1405,6 +1424,11 @@ 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); @@ -1420,7 +1444,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 @@ -1438,6 +1465,7 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, sz, message); OutputPluginWrite(ctx, true); + update_progress(ctx, true); } /* @@ -1598,7 +1626,7 @@ pgoutput_stream_commit(struct LogicalDecodingContext *ctx, Assert(!in_streaming); Assert(rbtxn_is_streamed(txn)); - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_commit(ctx->out, txn, commit_lsn); @@ -1619,7 +1647,7 @@ pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx, { Assert(rbtxn_is_streamed(txn)); - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn); OutputPluginWrite(ctx, true); @@ -2102,3 +2130,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); + skipped_changes_count = 0; + } +} diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 2d0292a..bfd6e7c 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -249,7 +249,7 @@ static long WalSndComputeSleeptime(TimestampTz now); 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); +static void WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, 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); @@ -1447,9 +1447,13 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, * LogicalDecodingContext 'update_progress' callback. * * Write the current position to the lag tracker (see XLogSendPhysical). + * + * 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) +WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, + bool last_write) { static TimestampTz sendTime = 0; TimestampTz now = GetCurrentTimestamp(); @@ -1459,12 +1463,40 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId * avoid flooding the lag tracker when we commit frequently. */ #define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 1000 - if (!TimestampDifferenceExceeds(sendTime, now, - WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) - return; + if (TimestampDifferenceExceeds(sendTime, now, + WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) + { + LagTrackerWrite(lsn, now); + sendTime = now; + } - LagTrackerWrite(lsn, now); - sendTime = now; + /* try to send a keepalive if required */ + if (!last_write) + { + /* + * We don't need to try sending keepalive unless we get too close to + * walsender timeout. + */ + if (now < TimestampTzPlusMilliseconds(last_reply_timestamp, + wal_sender_timeout / 2)) + return; + + /* Check for input from the client. */ + ProcessRepliesIfAny(); + + /* die if timeout was reached */ + WalSndCheckTimeOut(); + + /* Send keepalive if the time has come */ + WalSndKeepaliveIfNecessary(); + + if (!pq_is_send_pending()) + return; + + /* Try to flush pending output to the client */ + if (pq_flush_if_writable() != 0) + WalSndShutdown(); + } } /* diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h index 1097cc9..2c27ed6 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -26,7 +26,8 @@ typedef LogicalOutputPluginWriterWrite LogicalOutputPluginWriterPrepareWrite; typedef void (*LogicalOutputPluginWriterUpdateProgress) (struct LogicalDecodingContext *lr, XLogRecPtr Ptr, - TransactionId xid + TransactionId xid, + bool last_write ); typedef struct LogicalDecodingContext diff --git a/src/include/replication/output_plugin.h b/src/include/replication/output_plugin.h index a16bebf..3659e5a 100644 --- a/src/include/replication/output_plugin.h +++ b/src/include/replication/output_plugin.h @@ -270,6 +270,6 @@ 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); +extern void OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx, bool last_write); #endif /* OUTPUT_PLUGIN_H */ -- 1.8.3.1 ^ permalink raw reply [nested|flat] 35+ messages in thread
* RE: Logical replication timeout problem @ 2022-03-25 03:20 [email protected] <[email protected]> parent: Amit Kapila <[email protected]> 1 sibling, 0 replies; 35+ messages in thread From: [email protected] @ 2022-03-25 03:20 UTC (permalink / raw) To: 'Amit Kapila' <[email protected]>; [email protected] <[email protected]>; +Cc: Masahiko Sawada <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> Dear Amit, > It seems by mistake you have removed the changes from pgoutput_message > and pgoutput_truncate functions. I have added those back. > Additionally, I made a few other changes: (a) moved the function > UpdateProgress to pgoutput.c as it is not used outside it, (b) change > the new parameter in plugin API from 'send_keep_alive' to 'last_write' > to make it look similar to WalSndPrepareWrite and WalSndWriteData, (c) > made a number of changes in WalSndUpdateProgress API, it is better to > move keep-alive code after lag track code because we do process > replies at that time and there it will compute the lag; (d) > changed/added comments in the code. LGTM, but the patch cannot be applied to current HEAD. Best Regards, Hayato Kuroda FUJITSU LIMITED ^ permalink raw reply [nested|flat] 35+ messages in thread
* RE: Logical replication timeout problem @ 2022-03-25 05:23 [email protected] <[email protected]> parent: Amit Kapila <[email protected]> 1 sibling, 1 reply; 35+ messages in thread From: [email protected] @ 2022-03-25 05:23 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Masahiko Sawada <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Thur, Mar 24, 2022 at 6:32 PM Amit Kapila <[email protected]> wrote: > Thanks for your kindly update. > It seems by mistake you have removed the changes from pgoutput_message > and pgoutput_truncate functions. I have added those back. > Additionally, I made a few other changes: (a) moved the function > UpdateProgress to pgoutput.c as it is not used outside it, (b) change > the new parameter in plugin API from 'send_keep_alive' to 'last_write' > to make it look similar to WalSndPrepareWrite and WalSndWriteData, (c) > made a number of changes in WalSndUpdateProgress API, it is better to > move keep-alive code after lag track code because we do process > replies at that time and there it will compute the lag; (d) > changed/added comments in the code. > > Do let me know what you think of the attached? It looks good to me. Just rebase it because the change in header(75b1521). I tested it and the result looks good to me. Attach the new patch. Regards, Wang wei Attachments: [application/octet-stream] v6-0001-Fix-the-logical-replication-timeout-during-large-.patch (12.6K, ../../OS3PR01MB627542044A4C16511E63D5DA9E1A9@OS3PR01MB6275.jpnprd01.prod.outlook.com/2-v6-0001-Fix-the-logical-replication-timeout-during-large-.patch) download | inline diff: From b93a0fa1514f8f3d9e7cf2e806999ebc358bd486 Mon Sep 17 00:00:00 2001 From: wangw <[email protected]> Date: Fri, 25 Mar 2022 10:42:08 +0800 Subject: [PATCH v6] 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 | 6 +- src/backend/replication/pgoutput/pgoutput.c | 78 +++++++++++++++++++-- src/backend/replication/walsender.c | 46 ++++++++++-- src/include/replication/logical.h | 3 +- src/include/replication/output_plugin.h | 2 +- 5 files changed, 117 insertions(+), 18 deletions(-) diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 934aa13f2d..922b16c7c8 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -680,15 +680,15 @@ 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) +OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx, bool last_write) { if (!ctx->update_progress) return; - ctx->update_progress(ctx, ctx->write_location, ctx->write_xid); + ctx->update_progress(ctx, ctx->write_location, ctx->write_xid, last_write); } /* diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 4cdc698cbb..bf3d6d0ac6 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -94,6 +94,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", @@ -494,7 +495,7 @@ static void pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit(ctx->out, txn, commit_lsn); @@ -525,7 +526,7 @@ static void pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_prepare(ctx->out, txn, prepare_lsn); @@ -539,7 +540,7 @@ static void pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit_prepared(ctx->out, txn, commit_lsn); @@ -555,7 +556,7 @@ pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_rollback_prepared(ctx->out, txn, prepare_end_lsn, @@ -1168,9 +1169,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 @@ -1188,15 +1193,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); @@ -1245,6 +1259,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, logicalrep_write_insert(ctx->out, xid, targetrel, new_slot, data->binary); OutputPluginWrite(ctx, true); + change_sent = true; break; case REORDER_BUFFER_CHANGE_UPDATE: if (change->data.tp.oldtuple) @@ -1312,6 +1327,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, } OutputPluginWrite(ctx, true); + change_sent = true; break; case REORDER_BUFFER_CHANGE_DELETE: if (change->data.tp.oldtuple) @@ -1349,6 +1365,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"); @@ -1357,6 +1374,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Assert(false); } + update_progress(ctx, change_sent); + if (RelationIsValid(ancestor)) { RelationClose(ancestor); @@ -1424,6 +1443,11 @@ 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); @@ -1439,7 +1463,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 @@ -1457,6 +1484,7 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, sz, message); OutputPluginWrite(ctx, true); + update_progress(ctx, true); } static void @@ -1662,7 +1690,7 @@ pgoutput_stream_commit(struct LogicalDecodingContext *ctx, Assert(!in_streaming); Assert(rbtxn_is_streamed(txn)); - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_commit(ctx->out, txn, commit_lsn); @@ -1683,7 +1711,7 @@ pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx, { Assert(rbtxn_is_streamed(txn)); - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn); OutputPluginWrite(ctx, true); @@ -2171,3 +2199,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); + skipped_changes_count = 0; + } +} diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 2d0292a092..bfd6e7ce4e 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -249,7 +249,7 @@ static long WalSndComputeSleeptime(TimestampTz now); 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); +static void WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, 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); @@ -1447,9 +1447,13 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, * LogicalDecodingContext 'update_progress' callback. * * Write the current position to the lag tracker (see XLogSendPhysical). + * + * 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) +WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, + bool last_write) { static TimestampTz sendTime = 0; TimestampTz now = GetCurrentTimestamp(); @@ -1459,12 +1463,40 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId * avoid flooding the lag tracker when we commit frequently. */ #define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 1000 - if (!TimestampDifferenceExceeds(sendTime, now, - WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) - return; + if (TimestampDifferenceExceeds(sendTime, now, + WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) + { + LagTrackerWrite(lsn, now); + sendTime = now; + } - LagTrackerWrite(lsn, now); - sendTime = now; + /* try to send a keepalive if required */ + if (!last_write) + { + /* + * We don't need to try sending keepalive unless we get too close to + * walsender timeout. + */ + if (now < TimestampTzPlusMilliseconds(last_reply_timestamp, + wal_sender_timeout / 2)) + return; + + /* Check for input from the client. */ + ProcessRepliesIfAny(); + + /* die if timeout was reached */ + WalSndCheckTimeOut(); + + /* Send keepalive if the time has come */ + WalSndKeepaliveIfNecessary(); + + if (!pq_is_send_pending()) + return; + + /* Try to flush pending output to the client */ + if (pq_flush_if_writable() != 0) + WalSndShutdown(); + } } /* diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h index 1097cc9799..2c27ed6e50 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -26,7 +26,8 @@ typedef LogicalOutputPluginWriterWrite LogicalOutputPluginWriterPrepareWrite; typedef void (*LogicalOutputPluginWriterUpdateProgress) (struct LogicalDecodingContext *lr, XLogRecPtr Ptr, - TransactionId xid + TransactionId xid, + bool last_write ); typedef struct LogicalDecodingContext diff --git a/src/include/replication/output_plugin.h b/src/include/replication/output_plugin.h index a16bebf76c..3659e5a93a 100644 --- a/src/include/replication/output_plugin.h +++ b/src/include/replication/output_plugin.h @@ -270,6 +270,6 @@ 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); +extern void OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx, bool last_write); #endif /* OUTPUT_PLUGIN_H */ -- 2.18.4 ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2022-03-25 06:19 Masahiko Sawada <[email protected]> parent: [email protected] <[email protected]> 0 siblings, 2 replies; 35+ messages in thread From: Masahiko Sawada @ 2022-03-25 06:19 UTC (permalink / raw) To: [email protected] <[email protected]>; +Cc: Amit Kapila <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Fri, Mar 25, 2022 at 2:23 PM [email protected] <[email protected]> wrote: > > On Thur, Mar 24, 2022 at 6:32 PM Amit Kapila <[email protected]> wrote: > > > Thanks for your kindly update. > > > It seems by mistake you have removed the changes from pgoutput_message > > and pgoutput_truncate functions. I have added those back. > > Additionally, I made a few other changes: (a) moved the function > > UpdateProgress to pgoutput.c as it is not used outside it, (b) change > > the new parameter in plugin API from 'send_keep_alive' to 'last_write' > > to make it look similar to WalSndPrepareWrite and WalSndWriteData, (c) > > made a number of changes in WalSndUpdateProgress API, it is better to > > move keep-alive code after lag track code because we do process > > replies at that time and there it will compute the lag; (d) > > changed/added comments in the code. > > > > Do let me know what you think of the attached? > It looks good to me. Just rebase it because the change in header(75b1521). > I tested it and the result looks good to me. Since commit 75b1521 added decoding of sequence to logical replication, the patch needs to have pgoutput_sequence() call update_progress(). Regards, -- Masahiko Sawada EDB: https://www.enterprisedb.com/ ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2022-03-25 08:32 Amit Kapila <[email protected]> parent: Masahiko Sawada <[email protected]> 1 sibling, 1 reply; 35+ messages in thread From: Amit Kapila @ 2022-03-25 08:32 UTC (permalink / raw) To: Masahiko Sawada <[email protected]>; +Cc: [email protected] <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Fri, Mar 25, 2022 at 11:49 AM Masahiko Sawada <[email protected]> wrote: > > On Fri, Mar 25, 2022 at 2:23 PM [email protected] > <[email protected]> wrote: > > Since commit 75b1521 added decoding of sequence to logical > replication, the patch needs to have pgoutput_sequence() call > update_progress(). > Yeah, I also think this needs to be addressed. But apart from this, I want to know your and other's opinion on the following two points: a. Both this and the patch discussed in the nearby thread [1] add an additional parameter to WalSndUpdateProgress/OutputPluginUpdateProgress and it seems to me that both are required. The additional parameter 'last_write' added by this patch indicates: "If the last write is skipped then try (if we are close to wal_sender_timeout) to send a keepalive message to the receiver to avoid timeouts.". This means it can be used after any 'write' message. OTOH, the parameter 'skipped_xact' added by another patch [1] indicates if we have skipped sending anything for a transaction then sendkeepalive for synchronous replication to avoid any delays in such a transaction. Does this sound reasonable or can you think of a better way to deal with it? b. Do we want to backpatch the patch in this thread? I am reluctant to backpatch because it changes the exposed API which can have an impact and second there exists a workaround (user can increase wal_sender_timeout/wal_receiver_timeout). [1] - https://www.postgresql.org/message-id/OS0PR01MB5716BB24409D4B69206615B1941A9%40OS0PR01MB5716.jpnprd0... -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 35+ messages in thread
* RE: Logical replication timeout problem @ 2022-03-25 10:19 [email protected] <[email protected]> parent: Masahiko Sawada <[email protected]> 1 sibling, 1 reply; 35+ messages in thread From: [email protected] @ 2022-03-25 10:19 UTC (permalink / raw) To: Masahiko Sawada <[email protected]>; +Cc: Amit Kapila <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Fri, Mar 25, 2022 at 2:19 PM Masahiko Sawada <[email protected]> wrote: > On Fri, Mar 25, 2022 at 2:23 PM [email protected] > <[email protected]> wrote: > > > > On Thur, Mar 24, 2022 at 6:32 PM Amit Kapila <[email protected]> > wrote: > > > > > Thanks for your kindly update. > > > > > It seems by mistake you have removed the changes from > pgoutput_message > > > and pgoutput_truncate functions. I have added those back. > > > Additionally, I made a few other changes: (a) moved the function > > > UpdateProgress to pgoutput.c as it is not used outside it, (b) change > > > the new parameter in plugin API from 'send_keep_alive' to 'last_write' > > > to make it look similar to WalSndPrepareWrite and WalSndWriteData, (c) > > > made a number of changes in WalSndUpdateProgress API, it is better to > > > move keep-alive code after lag track code because we do process > > > replies at that time and there it will compute the lag; (d) > > > changed/added comments in the code. > > > > > > Do let me know what you think of the attached? > > It looks good to me. Just rebase it because the change in header(75b1521). > > I tested it and the result looks good to me. > > Since commit 75b1521 added decoding of sequence to logical > replication, the patch needs to have pgoutput_sequence() call > update_progress(). Thanks for your comments. Yes, you are right. Add missing handling of pgoutput_sequence. Attach the new patch. Regards, Wang wei Attachments: [application/octet-stream] v7-0001-Fix-the-logical-replication-timeout-during-large-.patch (13.4K, ../../OS3PR01MB62750C20AD7AF4F13541FC109E1A9@OS3PR01MB6275.jpnprd01.prod.outlook.com/2-v7-0001-Fix-the-logical-replication-timeout-during-large-.patch) download | inline diff: From a73eeb5bba86bc7d4afbdb465ae364f3b9c28727 Mon Sep 17 00:00:00 2001 From: wangw <[email protected]> Date: Fri, 25 Mar 2022 14:34:53 +0800 Subject: [PATCH v7] 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 | 6 +- src/backend/replication/pgoutput/pgoutput.c | 88 +++++++++++++++++++-- src/backend/replication/walsender.c | 46 +++++++++-- src/include/replication/logical.h | 3 +- src/include/replication/output_plugin.h | 2 +- 5 files changed, 127 insertions(+), 18 deletions(-) diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 934aa13f2d..922b16c7c8 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -680,15 +680,15 @@ 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) +OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx, bool last_write) { if (!ctx->update_progress) return; - ctx->update_progress(ctx, ctx->write_location, ctx->write_xid); + ctx->update_progress(ctx, ctx->write_location, ctx->write_xid, last_write); } /* diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 4cdc698cbb..88b523e6ae 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -94,6 +94,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", @@ -494,7 +495,7 @@ static void pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit(ctx->out, txn, commit_lsn); @@ -525,7 +526,7 @@ static void pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_prepare(ctx->out, txn, prepare_lsn); @@ -539,7 +540,7 @@ static void pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit_prepared(ctx->out, txn, commit_lsn); @@ -555,7 +556,7 @@ pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_rollback_prepared(ctx->out, txn, prepare_end_lsn, @@ -1168,9 +1169,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 @@ -1188,15 +1193,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); @@ -1245,6 +1259,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, logicalrep_write_insert(ctx->out, xid, targetrel, new_slot, data->binary); OutputPluginWrite(ctx, true); + change_sent = true; break; case REORDER_BUFFER_CHANGE_UPDATE: if (change->data.tp.oldtuple) @@ -1312,6 +1327,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, } OutputPluginWrite(ctx, true); + change_sent = true; break; case REORDER_BUFFER_CHANGE_DELETE: if (change->data.tp.oldtuple) @@ -1349,6 +1365,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"); @@ -1357,6 +1374,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Assert(false); } + update_progress(ctx, change_sent); + if (RelationIsValid(ancestor)) { RelationClose(ancestor); @@ -1424,6 +1443,11 @@ 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); @@ -1439,7 +1463,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 @@ -1457,6 +1484,7 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, sz, message); OutputPluginWrite(ctx, true); + update_progress(ctx, true); } static void @@ -1470,10 +1498,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 @@ -1490,7 +1524,10 @@ pgoutput_sequence(LogicalDecodingContext *ctx, * We handle just REORDER_BUFFER_CHANGE_SEQUENCE here. */ if (!relentry->pubactions.pubsequence) + { + update_progress(ctx, false); return; + } OutputPluginPrepareWrite(ctx, true); logicalrep_write_sequence(ctx->out, @@ -1502,6 +1539,7 @@ pgoutput_sequence(LogicalDecodingContext *ctx, log_cnt, is_called); OutputPluginWrite(ctx, true); + update_progress(ctx, true); } /* @@ -1662,7 +1700,7 @@ pgoutput_stream_commit(struct LogicalDecodingContext *ctx, Assert(!in_streaming); Assert(rbtxn_is_streamed(txn)); - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_commit(ctx->out, txn, commit_lsn); @@ -1683,7 +1721,7 @@ pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx, { Assert(rbtxn_is_streamed(txn)); - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn); OutputPluginWrite(ctx, true); @@ -2171,3 +2209,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); + skipped_changes_count = 0; + } +} diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 2d0292a092..bfd6e7ce4e 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -249,7 +249,7 @@ static long WalSndComputeSleeptime(TimestampTz now); 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); +static void WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, 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); @@ -1447,9 +1447,13 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, * LogicalDecodingContext 'update_progress' callback. * * Write the current position to the lag tracker (see XLogSendPhysical). + * + * 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) +WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, + bool last_write) { static TimestampTz sendTime = 0; TimestampTz now = GetCurrentTimestamp(); @@ -1459,12 +1463,40 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId * avoid flooding the lag tracker when we commit frequently. */ #define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 1000 - if (!TimestampDifferenceExceeds(sendTime, now, - WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) - return; + if (TimestampDifferenceExceeds(sendTime, now, + WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) + { + LagTrackerWrite(lsn, now); + sendTime = now; + } - LagTrackerWrite(lsn, now); - sendTime = now; + /* try to send a keepalive if required */ + if (!last_write) + { + /* + * We don't need to try sending keepalive unless we get too close to + * walsender timeout. + */ + if (now < TimestampTzPlusMilliseconds(last_reply_timestamp, + wal_sender_timeout / 2)) + return; + + /* Check for input from the client. */ + ProcessRepliesIfAny(); + + /* die if timeout was reached */ + WalSndCheckTimeOut(); + + /* Send keepalive if the time has come */ + WalSndKeepaliveIfNecessary(); + + if (!pq_is_send_pending()) + return; + + /* Try to flush pending output to the client */ + if (pq_flush_if_writable() != 0) + WalSndShutdown(); + } } /* diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h index 1097cc9799..2c27ed6e50 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -26,7 +26,8 @@ typedef LogicalOutputPluginWriterWrite LogicalOutputPluginWriterPrepareWrite; typedef void (*LogicalOutputPluginWriterUpdateProgress) (struct LogicalDecodingContext *lr, XLogRecPtr Ptr, - TransactionId xid + TransactionId xid, + bool last_write ); typedef struct LogicalDecodingContext diff --git a/src/include/replication/output_plugin.h b/src/include/replication/output_plugin.h index a16bebf76c..3659e5a93a 100644 --- a/src/include/replication/output_plugin.h +++ b/src/include/replication/output_plugin.h @@ -270,6 +270,6 @@ 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); +extern void OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx, bool last_write); #endif /* OUTPUT_PLUGIN_H */ -- 2.27.0 ^ permalink raw reply [nested|flat] 35+ messages in thread
* RE: Logical replication timeout problem @ 2022-03-28 01:55 [email protected] <[email protected]> parent: [email protected] <[email protected]> 0 siblings, 1 reply; 35+ messages in thread From: [email protected] @ 2022-03-28 01:55 UTC (permalink / raw) To: [email protected] <[email protected]>; Masahiko Sawada <[email protected]>; +Cc: Amit Kapila <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> Dear Wang-san, Thank you for updating! ...but it also cannot be applied to current HEAD because of the commit 923def9a533. Your patch seems to conflict the adding an argument of logicalrep_write_insert(). It allows specifying columns to publish by skipping some columns in logicalrep_write_tuple() which is called from logicalrep_write_insert() and logicalrep_write_update(). Do we have to consider something special case for that? I thought timeout may occur if users have huge table and publish few columns, but it is corner case. Best Regards, Hayato Kuroda FUJITSU LIMITED ^ permalink raw reply [nested|flat] 35+ messages in thread
* RE: Logical replication timeout problem @ 2022-03-28 06:11 [email protected] <[email protected]> parent: [email protected] <[email protected]> 0 siblings, 2 replies; 35+ messages in thread From: [email protected] @ 2022-03-28 06:11 UTC (permalink / raw) To: [email protected] <[email protected]>; Masahiko Sawada <[email protected]>; +Cc: Amit Kapila <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Mon, Mar 28, 2022 at 9:56 AM Kuroda, Hayato/黒田 隼人 <[email protected]> wrote: > Dear Wang-san, Thanks for your comments. > Thank you for updating! > ...but it also cannot be applied to current HEAD > because of the commit 923def9a533. > > Your patch seems to conflict the adding an argument of > logicalrep_write_insert(). > It allows specifying columns to publish by skipping some columns in > logicalrep_write_tuple() > which is called from logicalrep_write_insert() and logicalrep_write_update(). Thank for your kindly reminder. Rebase the patch. > Do we have to consider something special case for that? > I thought timeout may occur if users have huge table and publish few columns, > but it is corner case. I think maybe we do not need to deal with this use case. The maximum number of table columns allowed by PG is 1600 (macro MaxHeapAttributeNumber), and after loop through all columns in the function logicalrep_write_tuple, the function OutputPluginWrite will be invoked immediately to actually send the data to the subscriber. This refreshes the last time the subscriber received a message. So I think this loop will not cause timeout issues. Regards, Wang wei Attachments: [application/octet-stream] v8-0001-Fix-the-logical-replication-timeout-during-large-.patch (13.4K, ../../OS3PR01MB6275C64F264662E84D2FB7AE9E1D9@OS3PR01MB6275.jpnprd01.prod.outlook.com/2-v8-0001-Fix-the-logical-replication-timeout-during-large-.patch) download | inline diff: From 78b967c2cfa2110ad1a51a582b5cf894ec67c285 Mon Sep 17 00:00:00 2001 From: wangw <[email protected]> Date: Mon, 28 Mar 2022 01:36:55 -0400 Subject: [PATCH v8] 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 | 6 +- src/backend/replication/pgoutput/pgoutput.c | 88 +++++++++++++++++++-- src/backend/replication/walsender.c | 46 +++++++++-- src/include/replication/logical.h | 3 +- src/include/replication/output_plugin.h | 2 +- 5 files changed, 127 insertions(+), 18 deletions(-) diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 934aa13f2d..922b16c7c8 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -680,15 +680,15 @@ 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) +OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx, bool last_write) { if (!ctx->update_progress) return; - ctx->update_progress(ctx, ctx->write_location, ctx->write_xid); + ctx->update_progress(ctx, ctx->write_location, ctx->write_xid, last_write); } /* diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 893833ea83..a86326e75c 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", @@ -511,7 +512,7 @@ static void pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit(ctx->out, txn, commit_lsn); @@ -542,7 +543,7 @@ static void pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_prepare(ctx->out, txn, prepare_lsn); @@ -556,7 +557,7 @@ static void pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit_prepared(ctx->out, txn, commit_lsn); @@ -572,7 +573,7 @@ pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_rollback_prepared(ctx->out, txn, prepare_end_lsn, @@ -1303,9 +1304,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 @@ -1323,15 +1328,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); @@ -1380,6 +1394,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) @@ -1449,6 +1464,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, } OutputPluginWrite(ctx, true); + change_sent = true; break; case REORDER_BUFFER_CHANGE_DELETE: if (change->data.tp.oldtuple) @@ -1486,6 +1502,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"); @@ -1494,6 +1511,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Assert(false); } + update_progress(ctx, change_sent); + if (RelationIsValid(ancestor)) { RelationClose(ancestor); @@ -1561,6 +1580,11 @@ 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); @@ -1576,7 +1600,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 @@ -1594,6 +1621,7 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, sz, message); OutputPluginWrite(ctx, true); + update_progress(ctx, true); } static void @@ -1607,10 +1635,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 @@ -1627,7 +1661,10 @@ pgoutput_sequence(LogicalDecodingContext *ctx, * We handle just REORDER_BUFFER_CHANGE_SEQUENCE here. */ if (!relentry->pubactions.pubsequence) + { + update_progress(ctx, false); return; + } OutputPluginPrepareWrite(ctx, true); logicalrep_write_sequence(ctx->out, @@ -1639,6 +1676,7 @@ pgoutput_sequence(LogicalDecodingContext *ctx, log_cnt, is_called); OutputPluginWrite(ctx, true); + update_progress(ctx, true); } /* @@ -1799,7 +1837,7 @@ pgoutput_stream_commit(struct LogicalDecodingContext *ctx, Assert(!in_streaming); Assert(rbtxn_is_streamed(txn)); - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_commit(ctx->out, txn, commit_lsn); @@ -1820,7 +1858,7 @@ pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx, { Assert(rbtxn_is_streamed(txn)); - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn); OutputPluginWrite(ctx, true); @@ -2318,3 +2356,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); + skipped_changes_count = 0; + } +} diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 2d0292a092..bfd6e7ce4e 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -249,7 +249,7 @@ static long WalSndComputeSleeptime(TimestampTz now); 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); +static void WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, 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); @@ -1447,9 +1447,13 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, * LogicalDecodingContext 'update_progress' callback. * * Write the current position to the lag tracker (see XLogSendPhysical). + * + * 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) +WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, + bool last_write) { static TimestampTz sendTime = 0; TimestampTz now = GetCurrentTimestamp(); @@ -1459,12 +1463,40 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId * avoid flooding the lag tracker when we commit frequently. */ #define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 1000 - if (!TimestampDifferenceExceeds(sendTime, now, - WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) - return; + if (TimestampDifferenceExceeds(sendTime, now, + WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) + { + LagTrackerWrite(lsn, now); + sendTime = now; + } - LagTrackerWrite(lsn, now); - sendTime = now; + /* try to send a keepalive if required */ + if (!last_write) + { + /* + * We don't need to try sending keepalive unless we get too close to + * walsender timeout. + */ + if (now < TimestampTzPlusMilliseconds(last_reply_timestamp, + wal_sender_timeout / 2)) + return; + + /* Check for input from the client. */ + ProcessRepliesIfAny(); + + /* die if timeout was reached */ + WalSndCheckTimeOut(); + + /* Send keepalive if the time has come */ + WalSndKeepaliveIfNecessary(); + + if (!pq_is_send_pending()) + return; + + /* Try to flush pending output to the client */ + if (pq_flush_if_writable() != 0) + WalSndShutdown(); + } } /* diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h index 1097cc9799..2c27ed6e50 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -26,7 +26,8 @@ typedef LogicalOutputPluginWriterWrite LogicalOutputPluginWriterPrepareWrite; typedef void (*LogicalOutputPluginWriterUpdateProgress) (struct LogicalDecodingContext *lr, XLogRecPtr Ptr, - TransactionId xid + TransactionId xid, + bool last_write ); typedef struct LogicalDecodingContext diff --git a/src/include/replication/output_plugin.h b/src/include/replication/output_plugin.h index a16bebf76c..3659e5a93a 100644 --- a/src/include/replication/output_plugin.h +++ b/src/include/replication/output_plugin.h @@ -270,6 +270,6 @@ 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); +extern void OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx, bool last_write); #endif /* OUTPUT_PLUGIN_H */ -- 2.27.0 ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2022-03-28 06:27 Amit Kapila <[email protected]> parent: [email protected] <[email protected]> 1 sibling, 1 reply; 35+ messages in thread From: Amit Kapila @ 2022-03-28 06:27 UTC (permalink / raw) To: [email protected] <[email protected]>; +Cc: [email protected] <[email protected]>; Masahiko Sawada <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Mon, Mar 28, 2022 at 11:41 AM [email protected] <[email protected]> wrote: > > On Mon, Mar 28, 2022 at 9:56 AM Kuroda, Hayato/黒田 隼人 <[email protected]> wrote: > > > Do we have to consider something special case for that? > > I thought timeout may occur if users have huge table and publish few columns, > > but it is corner case. > I think maybe we do not need to deal with this use case. > The maximum number of table columns allowed by PG is 1600 > (macro MaxHeapAttributeNumber), and after loop through all columns in the > function logicalrep_write_tuple, the function OutputPluginWrite will be invoked > immediately to actually send the data to the subscriber. This refreshes the > last time the subscriber received a message. > So I think this loop will not cause timeout issues. > Right, I also don't think it can be a source of timeout. -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 35+ messages in thread
* RE: Logical replication timeout problem @ 2022-03-29 01:29 [email protected] <[email protected]> parent: Amit Kapila <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: [email protected] @ 2022-03-29 01:29 UTC (permalink / raw) To: [email protected] <[email protected]>; 'Amit Kapila' <[email protected]>; +Cc: Masahiko Sawada <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> Dear Amit, Wang, > > I think maybe we do not need to deal with this use case. > > The maximum number of table columns allowed by PG is 1600 > > (macro MaxHeapAttributeNumber), and after loop through all columns in the > > function logicalrep_write_tuple, the function OutputPluginWrite will be invoked > > immediately to actually send the data to the subscriber. This refreshes the > > last time the subscriber received a message. > > So I think this loop will not cause timeout issues. > > > > Right, I also don't think it can be a source of timeout. OK. I have no comments for this version. Best Regards, Hayato Kuroda FUJITSU LIMITED ^ permalink raw reply [nested|flat] 35+ messages in thread
* RE: Logical replication timeout problem @ 2022-03-29 01:44 [email protected] <[email protected]> parent: [email protected] <[email protected]> 1 sibling, 1 reply; 35+ messages in thread From: [email protected] @ 2022-03-29 01:44 UTC (permalink / raw) To: [email protected] <[email protected]>; Masahiko Sawada <[email protected]>; +Cc: Amit Kapila <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Mon, Mar 28, 2022 at 2:11 AM I wrote: > Rebase the patch. After reviewing anohter patch[1], I think this patch should also add a loop in function WalSndUpdateProgress like what did in function WalSndWriteData. So update the patch to be consistent with the existing code and the patch mentioned above. Attach the new patch. [1] - https://www.postgresql.org/message-id/OS0PR01MB5716946347F607F4CFB02FCE941D9%40OS0PR01MB5716.jpnprd0... Regards, Wang wei Attachments: [application/octet-stream] v9-0001-Fix-the-logical-replication-timeout-during-large-.patch (13.9K, ../../OS3PR01MB6275F0E4E27F5522847589ED9E1E9@OS3PR01MB6275.jpnprd01.prod.outlook.com/2-v9-0001-Fix-the-logical-replication-timeout-during-large-.patch) download | inline diff: From a8f054cf31cbff7c4bf2f2bd0dcae219c429421c Mon Sep 17 00:00:00 2001 From: wangw <[email protected]> Date: Tue, 29 Mar 2022 09:08:39 +0800 Subject: [PATCH v9] 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 | 6 +- src/backend/replication/pgoutput/pgoutput.c | 88 +++++++++++++++++++-- src/backend/replication/walsender.c | 43 ++++++++-- src/include/replication/logical.h | 3 +- src/include/replication/output_plugin.h | 2 +- 5 files changed, 124 insertions(+), 18 deletions(-) diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 934aa13f2d..922b16c7c8 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -680,15 +680,15 @@ 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) +OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx, bool last_write) { if (!ctx->update_progress) return; - ctx->update_progress(ctx, ctx->write_location, ctx->write_xid); + ctx->update_progress(ctx, ctx->write_location, ctx->write_xid, last_write); } /* diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 893833ea83..a86326e75c 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", @@ -511,7 +512,7 @@ static void pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit(ctx->out, txn, commit_lsn); @@ -542,7 +543,7 @@ static void pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_prepare(ctx->out, txn, prepare_lsn); @@ -556,7 +557,7 @@ static void pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit_prepared(ctx->out, txn, commit_lsn); @@ -572,7 +573,7 @@ pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time) { - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_rollback_prepared(ctx->out, txn, prepare_end_lsn, @@ -1303,9 +1304,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 @@ -1323,15 +1328,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); @@ -1380,6 +1394,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) @@ -1449,6 +1464,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, } OutputPluginWrite(ctx, true); + change_sent = true; break; case REORDER_BUFFER_CHANGE_DELETE: if (change->data.tp.oldtuple) @@ -1486,6 +1502,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"); @@ -1494,6 +1511,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Assert(false); } + update_progress(ctx, change_sent); + if (RelationIsValid(ancestor)) { RelationClose(ancestor); @@ -1561,6 +1580,11 @@ 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); @@ -1576,7 +1600,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 @@ -1594,6 +1621,7 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, sz, message); OutputPluginWrite(ctx, true); + update_progress(ctx, true); } static void @@ -1607,10 +1635,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 @@ -1627,7 +1661,10 @@ pgoutput_sequence(LogicalDecodingContext *ctx, * We handle just REORDER_BUFFER_CHANGE_SEQUENCE here. */ if (!relentry->pubactions.pubsequence) + { + update_progress(ctx, false); return; + } OutputPluginPrepareWrite(ctx, true); logicalrep_write_sequence(ctx->out, @@ -1639,6 +1676,7 @@ pgoutput_sequence(LogicalDecodingContext *ctx, log_cnt, is_called); OutputPluginWrite(ctx, true); + update_progress(ctx, true); } /* @@ -1799,7 +1837,7 @@ pgoutput_stream_commit(struct LogicalDecodingContext *ctx, Assert(!in_streaming); Assert(rbtxn_is_streamed(txn)); - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_commit(ctx->out, txn, commit_lsn); @@ -1820,7 +1858,7 @@ pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx, { Assert(rbtxn_is_streamed(txn)); - OutputPluginUpdateProgress(ctx); + OutputPluginUpdateProgress(ctx, true); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn); OutputPluginWrite(ctx, true); @@ -2318,3 +2356,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); + skipped_changes_count = 0; + } +} diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 2d0292a092..40083584af 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -242,6 +242,7 @@ static void ProcessStandbyMessage(void); static void ProcessStandbyReplyMessage(void); static void ProcessStandbyHSFeedbackMessage(void); static void ProcessRepliesIfAny(void); +static void ProcessPendingWritesAndTimeOut(void); static void WalSndKeepalive(bool requestReply); static void WalSndKeepaliveIfNecessary(void); static void WalSndCheckTimeOut(void); @@ -249,7 +250,7 @@ static long WalSndComputeSleeptime(TimestampTz now); 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); +static void WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, 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); @@ -1399,6 +1400,16 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, } /* If we have pending write here, go to slow path */ + ProcessPendingWritesAndTimeOut(); +} + +/* + * Wait until there is no pending write. Also process replies from the other + * side and check timeouts during that. + */ +static void +ProcessPendingWritesAndTimeOut(void) +{ for (;;) { long sleeptime; @@ -1447,9 +1458,13 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, * LogicalDecodingContext 'update_progress' callback. * * Write the current position to the lag tracker (see XLogSendPhysical). + * + * 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) +WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, + bool last_write) { static TimestampTz sendTime = 0; TimestampTz now = GetCurrentTimestamp(); @@ -1459,12 +1474,26 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId * avoid flooding the lag tracker when we commit frequently. */ #define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 1000 - if (!TimestampDifferenceExceeds(sendTime, now, - WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) - return; + if (TimestampDifferenceExceeds(sendTime, now, + WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) + { + LagTrackerWrite(lsn, now); + sendTime = now; + } - LagTrackerWrite(lsn, now); - sendTime = now; + /* try to send a keepalive if required */ + if (!last_write) + { + /* + * We don't need to try sending keepalive unless we get too close to + * walsender timeout. + */ + if (now < TimestampTzPlusMilliseconds(last_reply_timestamp, + wal_sender_timeout / 2)) + return; + + ProcessPendingWritesAndTimeOut(); + } } /* diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h index 1097cc9799..2c27ed6e50 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -26,7 +26,8 @@ typedef LogicalOutputPluginWriterWrite LogicalOutputPluginWriterPrepareWrite; typedef void (*LogicalOutputPluginWriterUpdateProgress) (struct LogicalDecodingContext *lr, XLogRecPtr Ptr, - TransactionId xid + TransactionId xid, + bool last_write ); typedef struct LogicalDecodingContext diff --git a/src/include/replication/output_plugin.h b/src/include/replication/output_plugin.h index a16bebf76c..3659e5a93a 100644 --- a/src/include/replication/output_plugin.h +++ b/src/include/replication/output_plugin.h @@ -270,6 +270,6 @@ 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); +extern void OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx, bool last_write); #endif /* OUTPUT_PLUGIN_H */ -- 2.18.4 ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2022-03-29 05:07 Masahiko Sawada <[email protected]> parent: Amit Kapila <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Masahiko Sawada @ 2022-03-29 05:07 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: [email protected] <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Fri, Mar 25, 2022 at 5:33 PM Amit Kapila <[email protected]> wrote: > > On Fri, Mar 25, 2022 at 11:49 AM Masahiko Sawada <[email protected]> wrote: > > > > On Fri, Mar 25, 2022 at 2:23 PM [email protected] > > <[email protected]> wrote: > > > > Since commit 75b1521 added decoding of sequence to logical > > replication, the patch needs to have pgoutput_sequence() call > > update_progress(). > > > > Yeah, I also think this needs to be addressed. But apart from this, I > want to know your and other's opinion on the following two points: > a. Both this and the patch discussed in the nearby thread [1] add an > additional parameter to > WalSndUpdateProgress/OutputPluginUpdateProgress and it seems to me > that both are required. The additional parameter 'last_write' added by > this patch indicates: "If the last write is skipped then try (if we > are close to wal_sender_timeout) to send a keepalive message to the > receiver to avoid timeouts.". This means it can be used after any > 'write' message. OTOH, the parameter 'skipped_xact' added by another > patch [1] indicates if we have skipped sending anything for a > transaction then sendkeepalive for synchronous replication to avoid > any delays in such a transaction. Does this sound reasonable or can > you think of a better way to deal with it? These current approaches look good to me. > b. Do we want to backpatch the patch in this thread? I am reluctant to > backpatch because it changes the exposed API which can have an impact > and second there exists a workaround (user can increase > wal_sender_timeout/wal_receiver_timeout). Yeah, we should avoid API changes between minor versions. I feel it's better to fix it also for back-branches but probably we need another fix for them. The issue reported on this thread seems quite confusable; it looks like a network problem but is not true. Also, the user who faced this issue has to increase wal_sender_timeout due to the decoded data size, which also means to delay detecting network problems. It seems an unrelated trade-off. Regards, -- Masahiko Sawada EDB: https://www.enterprisedb.com/ ^ permalink raw reply [nested|flat] 35+ messages in thread
* RE: Logical replication timeout problem @ 2022-03-30 07:54 [email protected] <[email protected]> parent: [email protected] <[email protected]> 0 siblings, 2 replies; 35+ messages in thread From: [email protected] @ 2022-03-30 07:54 UTC (permalink / raw) To: [email protected] <[email protected]>; [email protected] <[email protected]>; Masahiko Sawada <[email protected]>; +Cc: Amit Kapila <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Tues, Mar 29, 2022 at 9:45 AM I wrote: > Attach the new patch. Rebase the patch because the commit d5a9d86d in current HEAD. Regards, Wang wei Attachments: [application/octet-stream] v10-0001-Fix-the-logical-replication-timeout-during-large.patch (12.9K, ../../OS3PR01MB6275E0C2B4D9E488AD7CBA209E1F9@OS3PR01MB6275.jpnprd01.prod.outlook.com/2-v10-0001-Fix-the-logical-replication-timeout-during-large.patch) download | inline diff: From 844ae008dbd3f9bc5257b421f0b43bf14ca268ff Mon Sep 17 00:00:00 2001 From: wangw <[email protected]> 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 ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2022-03-30 08:59 Amit Kapila <[email protected]> parent: [email protected] <[email protected]> 1 sibling, 1 reply; 35+ messages in thread From: Amit Kapila @ 2022-03-30 08:59 UTC (permalink / raw) To: [email protected] <[email protected]>; +Cc: [email protected] <[email protected]>; Masahiko Sawada <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Wed, Mar 30, 2022 at 1:24 PM [email protected] <[email protected]> wrote: > > On Tues, Mar 29, 2022 at 9:45 AM I wrote: > > Attach the new patch. > > Rebase the patch because the commit d5a9d86d in current HEAD. > Thanks, this looks good to me apart from a minor indentation change which I'll take care of before committing. I am planning to push this day after tomorrow on Friday unless there are any other major comments. -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 35+ messages in thread
* RE: Logical replication timeout problem @ 2022-03-31 02:26 [email protected] <[email protected]> parent: [email protected] <[email protected]> 1 sibling, 0 replies; 35+ messages in thread From: [email protected] @ 2022-03-31 02:26 UTC (permalink / raw) To: [email protected] <[email protected]>; [email protected] <[email protected]>; [email protected] <[email protected]>; Masahiko Sawada <[email protected]>; +Cc: Amit Kapila <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Wed, Mar 30, 2022 3:54 PM [email protected] <[email protected]> wrote: > > Rebase the patch because the commit d5a9d86d in current HEAD. > Thanks for your patch, I tried this patch and confirmed that there is no timeout problem after applying this patch, and I could reproduce this problem on HEAD. Regards, Shi yu ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2022-03-31 12:24 Masahiko Sawada <[email protected]> parent: Amit Kapila <[email protected]> 0 siblings, 2 replies; 35+ messages in thread From: Masahiko Sawada @ 2022-03-31 12:24 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: [email protected] <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Wed, Mar 30, 2022 at 6:00 PM Amit Kapila <[email protected]> wrote: > > On Wed, Mar 30, 2022 at 1:24 PM [email protected] > <[email protected]> wrote: > > > > On Tues, Mar 29, 2022 at 9:45 AM I wrote: > > > Attach the new patch. > > > > Rebase the patch because the commit d5a9d86d in current HEAD. > > > > Thanks, this looks good to me apart from a minor indentation change > which I'll take care of before committing. I am planning to push this > day after tomorrow on Friday unless there are any other major > comments. The patch basically looks good to me. But the only concern to me is that once we get the patch committed, we will have to call update_progress() at all paths in callbacks that process changes. Which seems poor maintainability. On the other hand, possible another solution would be to add a new callback that is called e.g., every 1000 changes so that walsender does its job such as timeout handling while processing the decoded data in reorderbuffer.c. The callback is set only if the walsender does logical decoding, otherwise NULL. With this idea, other plugins will also be able to benefit without changes. But I’m not really sure it’s a good design, and adding a new callback introduces complexity. Regards, -- Masahiko Sawada EDB: https://www.enterprisedb.com/ ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2022-04-01 02:00 Amit Kapila <[email protected]> parent: Masahiko Sawada <[email protected]> 1 sibling, 0 replies; 35+ messages in thread From: Amit Kapila @ 2022-04-01 02:00 UTC (permalink / raw) To: Masahiko Sawada <[email protected]>; +Cc: [email protected] <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Thu, Mar 31, 2022 at 5:55 PM Masahiko Sawada <[email protected]> wrote: > On Wed, Mar 30, 2022 at 6:00 PM Amit Kapila <[email protected]> wrote: > > > > On Wed, Mar 30, 2022 at 1:24 PM [email protected] > > <[email protected]> wrote: > > > > > > On Tues, Mar 29, 2022 at 9:45 AM I wrote: > > > > Attach the new patch. > > > > > > Rebase the patch because the commit d5a9d86d in current HEAD. > > > > > > > Thanks, this looks good to me apart from a minor indentation change > > which I'll take care of before committing. I am planning to push this > > day after tomorrow on Friday unless there are any other major > > comments. > > The patch basically looks good to me. But the only concern to me is > that once we get the patch committed, we will have to call > update_progress() at all paths in callbacks that process changes. > Which seems poor maintainability. > > On the other hand, possible another solution would be to add a new > callback that is called e.g., every 1000 changes so that walsender > does its job such as timeout handling while processing the decoded > data in reorderbuffer.c. The callback is set only if the walsender > does logical decoding, otherwise NULL. With this idea, other plugins > will also be able to benefit without changes. But I’m not really sure > it’s a good design, and adding a new callback introduces complexity. > Yeah, same here. I have also mentioned another way to expose an API from reorderbuffer [1] by introducing a skip API but just not sure if that or this API is generic enough to make it adding worth. Also, note that the current patch makes the progress recording of large transactions somewhat better when most of the changes are skipped. We can further extend it to make it true for other cases as well but that probably can be done separately if required as that is not required for this bug-fix. I intend to commit this patch today but I think it is better to wait for a few more days to see if anybody has any opinion on this matter. I'll push this on Tuesday unless we decide to do something different here. [1] - https://www.postgresql.org/message-id/CAA4eK1%2BfQjndoBOFUn9Wy0hhm3MLyUWEpcT9O7iuCELktfdBiQ%40mail.g... -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2022-04-01 02:03 Euler Taveira <[email protected]> parent: Masahiko Sawada <[email protected]> 1 sibling, 1 reply; 35+ messages in thread From: Euler Taveira @ 2022-04-01 02:03 UTC (permalink / raw) To: Masahiko Sawada <[email protected]>; Amit Kapila <[email protected]>; +Cc: [email protected] <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Thu, Mar 31, 2022, at 9:24 AM, Masahiko Sawada wrote: > The patch basically looks good to me. But the only concern to me is > that once we get the patch committed, we will have to call > update_progress() at all paths in callbacks that process changes. > Which seems poor maintainability. I didn't like the current fix for the same reason. We need a robust feedback system for logical replication. We had this discussion in the "skip empty transactions" thread [1]. > On the other hand, possible another solution would be to add a new > callback that is called e.g., every 1000 changes so that walsender > does its job such as timeout handling while processing the decoded > data in reorderbuffer.c. The callback is set only if the walsender > does logical decoding, otherwise NULL. With this idea, other plugins > will also be able to benefit without changes. But I’m not really sure > it’s a good design, and adding a new callback introduces complexity. No new callback is required. In the current code, each output plugin callback is responsible to call OutputPluginUpdateProgress. It is up to the output plugin author to add calls to this function. The lack of a call in a callback might cause issues like what was described in the initial message. 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? If so, we could have an output plugin parameter to inform which callbacks we would like to call the update progress routine. This would simplify the code, make it less error prone and wouldn't impose a burden on maintainability. [1] https://www.postgresql.org/message-id/20200309183018.tzkzwu635sd366ej%40alap3.anarazel.de -- Euler Taveira EDB https://www.enterprisedb.com/ ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2022-04-01 02:27 Amit Kapila <[email protected]> parent: Euler Taveira <[email protected]> 0 siblings, 1 reply; 35+ messages in thread From: Amit Kapila @ 2022-04-01 02:27 UTC (permalink / raw) To: Euler Taveira <[email protected]>; +Cc: Masahiko Sawada <[email protected]>; [email protected] <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Fri, Apr 1, 2022 at 7:33 AM Euler Taveira <[email protected]> wrote: > > On Thu, Mar 31, 2022, at 9:24 AM, Masahiko Sawada wrote: > > On the other hand, possible another solution would be to add a new > callback that is called e.g., every 1000 changes so that walsender > does its job such as timeout handling while processing the decoded > data in reorderbuffer.c. The callback is set only if the walsender > does logical decoding, otherwise NULL. With this idea, other plugins > will also be able to benefit without changes. But I’m not really sure > it’s a good design, and adding a new callback introduces complexity. > > No new callback is required. > > In the current code, each output plugin callback is responsible to call > OutputPluginUpdateProgress. It is up to the output plugin author to add calls > to this function. The lack of a call in a callback might cause issues like what > was described in the initial message. > 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? > 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? [1] - https://www.postgresql.org/message-id/OS3PR01MB6275DFFDAC7A59FA148931529E209%40OS3PR01MB6275.jpnprd0... -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2022-04-01 02:57 Euler Taveira <[email protected]> parent: Amit Kapila <[email protected]> 0 siblings, 1 reply; 35+ messages in thread From: Euler Taveira @ 2022-04-01 02:57 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Masahiko Sawada <[email protected]>; [email protected] <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> 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. -- Euler Taveira EDB https://www.enterprisedb.com/ ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2022-04-01 04:08 Amit Kapila <[email protected]> parent: Euler Taveira <[email protected]> 0 siblings, 2 replies; 35+ messages in thread From: Amit Kapila @ 2022-04-01 04:08 UTC (permalink / raw) To: Euler Taveira <[email protected]>; +Cc: Masahiko Sawada <[email protected]>; [email protected] <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> 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. -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 35+ messages in thread
* RE: Logical replication timeout problem @ 2022-04-06 05:39 [email protected] <[email protected]> parent: Amit Kapila <[email protected]> 1 sibling, 1 reply; 35+ messages in thread From: [email protected] @ 2022-04-06 05:39 UTC (permalink / raw) To: Amit Kapila <[email protected]>; Euler Taveira <[email protected]>; +Cc: Masahiko Sawada <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> 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 ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2022-04-06 05:58 Amit Kapila <[email protected]> parent: [email protected] <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: Amit Kapila @ 2022-04-06 05:58 UTC (permalink / raw) To: [email protected] <[email protected]>; +Cc: Euler Taveira <[email protected]>; Masahiko Sawada <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Wed, Apr 6, 2022 at 11:09 AM [email protected] <[email protected]> wrote: > > 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: > > > > > > 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. > 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? Also, let's try to evaluate how it impacts lag functionality for large transactions? -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2022-10-18 14:35 Fabrice Chapuis <[email protected]> parent: Amit Kapila <[email protected]> 1 sibling, 1 reply; 35+ messages in thread From: Fabrice Chapuis @ 2022-10-18 14:35 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Euler Taveira <[email protected]>; Masahiko Sawada <[email protected]>; [email protected] <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> Hello Amit, In version 14.4 the timeout problem for logical replication happens again despite the patch provided for this issue in this version. When bulky materialized views are reloaded it broke logical replication. It is possible to solve this problem by using your new "streaming" option. Have you ever had this issue reported to you? Regards Fabrice 2022-10-10 17:19:02 CEST [538424]: [17-1] user=postgres,db=dbxxxa00,client=[local] CONTEXT: SQL statement "REFRESH MATERIALIZED VIEW sxxxa00.table_base" PL/pgSQL function refresh_materialized_view(text) line 5 at EXECUTE 2022-10-10 17:19:02 CEST [538424]: [18-1] user=postgres,db=dbxxxa00,client=[local] STATEMENT: select refresh_materialized_view('sxxxa00.table_base'); 2022-10-10 17:19:02 CEST [538424]: [19-1] user=postgres,db=dbxxxa00,client=[local] LOG: duration: 264815.652 ms statement: select refresh_materialized_view('sxxxa00.table_base'); 2022-10-10 17:19:27 CEST [559156]: [1-1] user=,db=,client= LOG: automatic vacuum of table "dbxxxa00.sxxxa00.table_base": index scans: 0 pages: 0 removed, 296589 remain, 0 skipped due to pins, 0 skipped frozen tuples: 0 removed, 48472622 remain, 0 are dead but not yet removable, oldest xmin: 1501528 index scan not needed: 0 pages from table (0.00% of total) had 0 dead item identifiers removed I/O timings: read: 1.494 ms, write: 0.000 ms avg read rate: 0.028 MB/s, avg write rate: 107.952 MB/s buffer usage: 593301 hits, 77 misses, 294605 dirtied WAL usage: 296644 records, 46119 full page images, 173652718 bytes system usage: CPU: user: 17.26 s, system: 0.29 s, elapsed: 21.32 s 2022-10-10 17:19:28 CEST [559156]: [2-1] user=,db=,client= LOG: automatic analyze of table "dbxxxa00.sxxxa00.table_base" I/O timings: read: 0.043 ms, write: 0.000 ms avg read rate: 0.026 MB/s, avg write rate: 0.026 MB/s buffer usage: 30308 hits, 2 misses, 2 dirtied system usage: CPU: user: 0.54 s, system: 0.00 s, elapsed: 0.59 s 2022-10-10 17:19:34 CEST [3898111]: [6840-1] user=,db=,client= LOG: checkpoint complete: wrote 1194 buffers (0.0%); 0 WAL file(s) added, 0 removed, 0 recycled; write=269.551 s, sync=0.002 s, total=269.560 s; sync files=251, longest=0.00 1 s, average=0.001 s; distance=583790 kB, estimate=583790 kB 2022-10-10 17:20:02 CEST [716163]: [2-1] user=,db=,client= ERROR: terminating logical replication worker due to timeout 2022-10-10 17:20:02 CEST [3897921]: [13-1] user=,db=,client= LOG: background worker "logical replication worker" (PID 716163) exited with exit code 1 2022-10-10 17:20:02 CEST [561346]: [1-1] user=,db=,client= LOG: logical replication apply worker for subscription "subxxx_sxxxa00" has started On Fri, Apr 1, 2022 at 6: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. > > -- > With Regards, > Amit Kapila. > ^ permalink raw reply [nested|flat] 35+ messages in thread
* RE: Logical replication timeout problem @ 2022-10-19 08:15 [email protected] <[email protected]> parent: Fabrice Chapuis <[email protected]> 0 siblings, 2 replies; 35+ messages in thread From: [email protected] @ 2022-10-19 08:15 UTC (permalink / raw) To: Fabrice Chapuis <[email protected]>; Amit Kapila <[email protected]>; +Cc: Euler Taveira <[email protected]>; Masahiko Sawada <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Tue, Oct 18, 2022 at 22:35 PM Fabrice Chapuis <[email protected]> wrote: > Hello Amit, > > In version 14.4 the timeout problem for logical replication happens again despite > the patch provided for this issue in this version. When bulky materialized views > are reloaded it broke logical replication. It is possible to solve this problem by > using your new "streaming" option. > Have you ever had this issue reported to you? > > Regards > > Fabrice > > 2022-10-10 17:19:02 CEST [538424]: [17-1] > user=postgres,db=dbxxxa00,client=[local] CONTEXT: SQL statement "REFRESH > MATERIALIZED VIEW sxxxa00.table_base" > PL/pgSQL function refresh_materialized_view(text) line 5 at EXECUTE > 2022-10-10 17:19:02 CEST [538424]: [18-1] > user=postgres,db=dbxxxa00,client=[local] STATEMENT: select > refresh_materialized_view('sxxxa00.table_base'); > 2022-10-10 17:19:02 CEST [538424]: [19-1] > user=postgres,db=dbxxxa00,client=[local] LOG: duration: 264815.652 > ms statement: select refresh_materialized_view('sxxxa00.table_base'); > 2022-10-10 17:19:27 CEST [559156]: [1-1] user=,db=,client= LOG: automatic > vacuum of table "dbxxxa00.sxxxa00.table_base": index scans: 0 > pages: 0 removed, 296589 remain, 0 skipped due to pins, 0 skipped frozen > tuples: 0 removed, 48472622 remain, 0 are dead but not yet removable, > oldest xmin: 1501528 > index scan not needed: 0 pages from table (0.00% of total) had 0 dead item > identifiers removed > I/O timings: read: 1.494 ms, write: 0.000 ms > avg read rate: 0.028 MB/s, avg write rate: 107.952 MB/s > buffer usage: 593301 hits, 77 misses, 294605 dirtied > WAL usage: 296644 records, 46119 full page images, 173652718 bytes > system usage: CPU: user: 17.26 s, system: 0.29 s, elapsed: 21.32 s > 2022-10-10 17:19:28 CEST [559156]: [2-1] user=,db=,client= LOG: automatic > analyze of table "dbxxxa00.sxxxa00.table_base" > I/O timings: read: 0.043 ms, write: 0.000 ms > avg read rate: 0.026 MB/s, avg write rate: 0.026 MB/s > buffer usage: 30308 hits, 2 misses, 2 dirtied > system usage: CPU: user: 0.54 s, system: 0.00 s, elapsed: 0.59 s > 2022-10-10 17:19:34 CEST [3898111]: [6840-1] user=,db=,client= LOG: checkpoint > complete: wrote 1194 buffers (0.0%); 0 WAL file(s) added, 0 removed, 0 recycled; > write=269.551 s, sync=0.002 s, total=269.560 s; sync files=251, longest=0.00 > 1 s, average=0.001 s; distance=583790 kB, estimate=583790 kB > 2022-10-10 17:20:02 CEST [716163]: [2-1] user=,db=,client= ERROR: terminating > logical replication worker due to timeout > 2022-10-10 17:20:02 CEST [3897921]: [13-1] user=,db=,client= LOG: background > worker "logical replication worker" (PID 716163) exited with exit code 1 > 2022-10-10 17:20:02 CEST [561346]: [1-1] user=,db=,client= LOG: logical > replication apply worker for subscription "subxxx_sxxxa00" has started Thanks for reporting! There is one thing I want to confirm: Is the statement `select refresh_materialized_view('sxxxa00.table_base');` executed on the publisher-side? If so, I think the reason for this timeout problem could be that during DDL (`REFRESH MATERIALIZED VIEW`), lots of temporary data is generated due to rewrite. Since these temporary data will not be processed by the pgoutput plugin, our previous fix for DML had no impact on this case. I think setting "streaming" option to "on" could work around this problem. I tried to write a draft patch (see attachment) on REL_14_4 to fix this. I tried it locally and it seems to work. Could you please confirm whether this problem is fixed after applying this draft patch? If this draft patch works, I will improve it and try to fix this problem. Regards, Wang wei Attachments: [application/octet-stream] 0001-draft-for-REL_14_4.patch (2.4K, ../../OS3PR01MB6275478E5D29E4A563302D3D9E2B9@OS3PR01MB6275.jpnprd01.prod.outlook.com/2-0001-draft-for-REL_14_4.patch) download | inline diff: From f5b260784d9c9f50f01fd5ca6ac71e7fd83e5f42 Mon Sep 17 00:00:00 2001 From: wangw <[email protected]> Date: Wed, 19 Oct 2022 12:06:45 +0800 Subject: [PATCH] draft for REL_14_4 --- .../replication/logical/reorderbuffer.c | 46 +++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index e59d1396b5..5434656859 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -1950,6 +1950,50 @@ ReorderBufferApplyMessage(ReorderBuffer *rb, ReorderBufferTXN *txn, change->data.msg.message); } +/* + * Helper function for ReorderBufferProcessTXN for updating progress. + */ +static inline void +ReorderBufferApplyUpdateProgress(ReorderBuffer *rb, ReorderBufferTXN *txn, + ReorderBufferChange *change) +{ + static int changes_count = 0; + LogicalDecodingContext *ctx = rb->private_data; + + Assert(!ctx->fast_forward); + + /* set output state */ + ctx->accept_writes = false; + ctx->write_xid = txn->xid; + + /* + * Report this change's lsn so replies from clients can give an up-to-date + * answer. This won't ever be enough (and shouldn't be!) to confirm + * receipt of this transaction, but it might allow another transaction's + * commit to be confirmed with one message. + */ + ctx->write_location = change->lsn; + + /* + * 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) + { + OutputPluginUpdateProgress(ctx); + changes_count = 0; + } +} + /* * Function to store the command id and snapshot at the end of the current * stream so that we can reuse the same while sending the next stream. @@ -2393,6 +2437,8 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, elog(ERROR, "tuplecid value in changequeue"); break; } + + ReorderBufferApplyUpdateProgress(rb, txn, change); } /* speculative insertion record must be freed by now */ -- 2.23.0.windows.1 ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2022-10-20 05:46 Fabrice Chapuis <[email protected]> parent: [email protected] <[email protected]> 1 sibling, 1 reply; 35+ messages in thread From: Fabrice Chapuis @ 2022-10-20 05:46 UTC (permalink / raw) To: [email protected]; +Cc: Amit Kapila <[email protected]>; Euler Taveira <[email protected]>; Masahiko Sawada <[email protected]>; [email protected]; Peter Smith <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; Tang, Haiying/唐 海英 <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> Yes the refresh of MV is on the Publisher Side. Thanks for your draft patch, I'll try it I'll back to you as soonas possible One question: why the refresh of the MV is a DDL not a DML? Regards Fabrice On Wed, 19 Oct 2022, 10:15 [email protected] <[email protected]> wrote: > On Tue, Oct 18, 2022 at 22:35 PM Fabrice Chapuis <[email protected]> > wrote: > > Hello Amit, > > > > In version 14.4 the timeout problem for logical replication happens > again despite > > the patch provided for this issue in this version. When bulky > materialized views > > are reloaded it broke logical replication. It is possible to solve this > problem by > > using your new "streaming" option. > > Have you ever had this issue reported to you? > > > > Regards > > > > Fabrice > > > > 2022-10-10 17:19:02 CEST [538424]: [17-1] > > user=postgres,db=dbxxxa00,client=[local] CONTEXT: SQL statement "REFRESH > > MATERIALIZED VIEW sxxxa00.table_base" > > PL/pgSQL function refresh_materialized_view(text) line 5 at > EXECUTE > > 2022-10-10 17:19:02 CEST [538424]: [18-1] > > user=postgres,db=dbxxxa00,client=[local] STATEMENT: select > > refresh_materialized_view('sxxxa00.table_base'); > > 2022-10-10 17:19:02 CEST [538424]: [19-1] > > user=postgres,db=dbxxxa00,client=[local] LOG: duration: 264815.652 > > ms statement: select refresh_materialized_view('sxxxa00.table_base'); > > 2022-10-10 17:19:27 CEST [559156]: [1-1] user=,db=,client= LOG: > automatic > > vacuum of table "dbxxxa00.sxxxa00.table_base": index scans: 0 > > pages: 0 removed, 296589 remain, 0 skipped due to pins, 0 > skipped frozen > > tuples: 0 removed, 48472622 remain, 0 are dead but not yet > removable, > > oldest xmin: 1501528 > > index scan not needed: 0 pages from table (0.00% of total) had 0 > dead item > > identifiers removed > > I/O timings: read: 1.494 ms, write: 0.000 ms > > avg read rate: 0.028 MB/s, avg write rate: 107.952 MB/s > > buffer usage: 593301 hits, 77 misses, 294605 dirtied > > WAL usage: 296644 records, 46119 full page images, 173652718 > bytes > > system usage: CPU: user: 17.26 s, system: 0.29 s, elapsed: 21.32 > s > > 2022-10-10 17:19:28 CEST [559156]: [2-1] user=,db=,client= LOG: > automatic > > analyze of table "dbxxxa00.sxxxa00.table_base" > > I/O timings: read: 0.043 ms, write: 0.000 ms > > avg read rate: 0.026 MB/s, avg write rate: 0.026 MB/s > > buffer usage: 30308 hits, 2 misses, 2 dirtied > > system usage: CPU: user: 0.54 s, system: 0.00 s, elapsed: 0.59 s > > 2022-10-10 17:19:34 CEST [3898111]: [6840-1] user=,db=,client= LOG: > checkpoint > > complete: wrote 1194 buffers (0.0%); 0 WAL file(s) added, 0 removed, 0 > recycled; > > write=269.551 s, sync=0.002 s, total=269.560 s; sync files=251, > longest=0.00 > > 1 s, average=0.001 s; distance=583790 kB, estimate=583790 kB > > 2022-10-10 17:20:02 CEST [716163]: [2-1] user=,db=,client= ERROR: > terminating > > logical replication worker due to timeout > > 2022-10-10 17:20:02 CEST [3897921]: [13-1] user=,db=,client= LOG: > background > > worker "logical replication worker" (PID 716163) exited with exit code 1 > > 2022-10-10 17:20:02 CEST [561346]: [1-1] user=,db=,client= LOG: logical > > replication apply worker for subscription "subxxx_sxxxa00" has started > > Thanks for reporting! > > There is one thing I want to confirm: > Is the statement `select refresh_materialized_view('sxxxa00.table_base');` > executed on the publisher-side? > > If so, I think the reason for this timeout problem could be that during DDL > (`REFRESH MATERIALIZED VIEW`), lots of temporary data is generated due to > rewrite. Since these temporary data will not be processed by the pgoutput > plugin, our previous fix for DML had no impact on this case. > I think setting "streaming" option to "on" could work around this problem. > > I tried to write a draft patch (see attachment) on REL_14_4 to fix this. > I tried it locally and it seems to work. > Could you please confirm whether this problem is fixed after applying this > draft patch? > > If this draft patch works, I will improve it and try to fix this problem. > > Regards, > Wang wei > ^ permalink raw reply [nested|flat] 35+ messages in thread
* RE: Logical replication timeout problem @ 2022-10-20 07:08 [email protected] <[email protected]> parent: Fabrice Chapuis <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: [email protected] @ 2022-10-20 07:08 UTC (permalink / raw) To: Fabrice Chapuis <[email protected]>; +Cc: Amit Kapila <[email protected]>; Euler Taveira <[email protected]>; Masahiko Sawada <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Thurs, Oct 20, 2022 at 13:47 PM Fabrice Chapuis <[email protected]> wrote: > Yes the refresh of MV is on the Publisher Side. > Thanks for your draft patch, I'll try it > I'll back to you as soonas possible Thanks a lot. > One question: why the refresh of the MV is a DDL not a DML? Since in the source, the type of command `REFRESH MATERIALIZED VIEW` is `CMD_UTILITY`, I think this command is DDL (see CmdType in file nodes.h). BTW, after trying to search for DML in the pg-doc, I found the relevant description in the below link: https://www.postgresql.org/docs/devel/logical-replication-publication.html Regards, Wang wei ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2022-11-04 10:13 Fabrice Chapuis <[email protected]> parent: [email protected] <[email protected]> 1 sibling, 1 reply; 35+ messages in thread From: Fabrice Chapuis @ 2022-11-04 10:13 UTC (permalink / raw) To: [email protected] <[email protected]>; +Cc: Amit Kapila <[email protected]>; Euler Taveira <[email protected]>; Masahiko Sawada <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> Hello Wang, I tested the draft patch in my lab for Postgres 14.4, the refresh of the materialized view ran without generating the timeout on the worker. Do you plan to propose this patch at the next commit fest. Regards, Fabrice On Wed, Oct 19, 2022 at 10:15 AM [email protected] < [email protected]> wrote: > On Tue, Oct 18, 2022 at 22:35 PM Fabrice Chapuis <[email protected]> > wrote: > > Hello Amit, > > > > In version 14.4 the timeout problem for logical replication happens > again despite > > the patch provided for this issue in this version. When bulky > materialized views > > are reloaded it broke logical replication. It is possible to solve this > problem by > > using your new "streaming" option. > > Have you ever had this issue reported to you? > > > > Regards > > > > Fabrice > > > > 2022-10-10 17:19:02 CEST [538424]: [17-1] > > user=postgres,db=dbxxxa00,client=[local] CONTEXT: SQL statement "REFRESH > > MATERIALIZED VIEW sxxxa00.table_base" > > PL/pgSQL function refresh_materialized_view(text) line 5 at > EXECUTE > > 2022-10-10 17:19:02 CEST [538424]: [18-1] > > user=postgres,db=dbxxxa00,client=[local] STATEMENT: select > > refresh_materialized_view('sxxxa00.table_base'); > > 2022-10-10 17:19:02 CEST [538424]: [19-1] > > user=postgres,db=dbxxxa00,client=[local] LOG: duration: 264815.652 > > ms statement: select refresh_materialized_view('sxxxa00.table_base'); > > 2022-10-10 17:19:27 CEST [559156]: [1-1] user=,db=,client= LOG: > automatic > > vacuum of table "dbxxxa00.sxxxa00.table_base": index scans: 0 > > pages: 0 removed, 296589 remain, 0 skipped due to pins, 0 > skipped frozen > > tuples: 0 removed, 48472622 remain, 0 are dead but not yet > removable, > > oldest xmin: 1501528 > > index scan not needed: 0 pages from table (0.00% of total) had 0 > dead item > > identifiers removed > > I/O timings: read: 1.494 ms, write: 0.000 ms > > avg read rate: 0.028 MB/s, avg write rate: 107.952 MB/s > > buffer usage: 593301 hits, 77 misses, 294605 dirtied > > WAL usage: 296644 records, 46119 full page images, 173652718 > bytes > > system usage: CPU: user: 17.26 s, system: 0.29 s, elapsed: 21.32 > s > > 2022-10-10 17:19:28 CEST [559156]: [2-1] user=,db=,client= LOG: > automatic > > analyze of table "dbxxxa00.sxxxa00.table_base" > > I/O timings: read: 0.043 ms, write: 0.000 ms > > avg read rate: 0.026 MB/s, avg write rate: 0.026 MB/s > > buffer usage: 30308 hits, 2 misses, 2 dirtied > > system usage: CPU: user: 0.54 s, system: 0.00 s, elapsed: 0.59 s > > 2022-10-10 17:19:34 CEST [3898111]: [6840-1] user=,db=,client= LOG: > checkpoint > > complete: wrote 1194 buffers (0.0%); 0 WAL file(s) added, 0 removed, 0 > recycled; > > write=269.551 s, sync=0.002 s, total=269.560 s; sync files=251, > longest=0.00 > > 1 s, average=0.001 s; distance=583790 kB, estimate=583790 kB > > 2022-10-10 17:20:02 CEST [716163]: [2-1] user=,db=,client= ERROR: > terminating > > logical replication worker due to timeout > > 2022-10-10 17:20:02 CEST [3897921]: [13-1] user=,db=,client= LOG: > background > > worker "logical replication worker" (PID 716163) exited with exit code 1 > > 2022-10-10 17:20:02 CEST [561346]: [1-1] user=,db=,client= LOG: logical > > replication apply worker for subscription "subxxx_sxxxa00" has started > > Thanks for reporting! > > There is one thing I want to confirm: > Is the statement `select refresh_materialized_view('sxxxa00.table_base');` > executed on the publisher-side? > > If so, I think the reason for this timeout problem could be that during DDL > (`REFRESH MATERIALIZED VIEW`), lots of temporary data is generated due to > rewrite. Since these temporary data will not be processed by the pgoutput > plugin, our previous fix for DML had no impact on this case. > I think setting "streaming" option to "on" could work around this problem. > > I tried to write a draft patch (see attachment) on REL_14_4 to fix this. > I tried it locally and it seems to work. > Could you please confirm whether this problem is fixed after applying this > draft patch? > > If this draft patch works, I will improve it and try to fix this problem. > > Regards, > Wang wei > ^ permalink raw reply [nested|flat] 35+ messages in thread
* RE: Logical replication timeout problem @ 2022-11-08 03:04 [email protected] <[email protected]> parent: Fabrice Chapuis <[email protected]> 0 siblings, 1 reply; 35+ messages in thread From: [email protected] @ 2022-11-08 03:04 UTC (permalink / raw) To: Fabrice Chapuis <[email protected]>; +Cc: Amit Kapila <[email protected]>; Euler Taveira <[email protected]>; Masahiko Sawada <[email protected]>; Hayato Kuroda (Fujitsu) <[email protected]>; Peter Smith <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Fri, Nov 4, 2022 at 18:13 PM Fabrice Chapuis <[email protected]> wrote: > Hello Wang, > > I tested the draft patch in my lab for Postgres 14.4, the refresh of the > materialized view ran without generating the timeout on the worker. > Do you plan to propose this patch at the next commit fest. Thanks for your confirmation! I will add this thread to the commit fest soon. The following is the problem analysis and fix approach: I think the problem is when there is a DDL in a transaction that generates lots of temporary data due to rewrite rules, these temporary data will not be processed by the pgoutput - plugin. Therefore, the previous fix (f95d53e) for DML had no impact on this case. To fix this, I think we need to try to send the keepalive messages after each change is processed by walsender, not in the pgoutput-plugin. Attach the patch. Regards, Wang wei Attachments: [application/octet-stream] v1-0001-Fix-the-logical-replication-timeout-during-proces.patch (7.3K, ../../OS3PR01MB62751A8063A9A75A096000D89E3F9@OS3PR01MB6275.jpnprd01.prod.outlook.com/2-v1-0001-Fix-the-logical-replication-timeout-during-proces.patch) download | inline diff: From ff9b213d4a605c8237363802899747de9ea38d9f Mon Sep 17 00:00:00 2001 From: wangw <[email protected]> Date: Mon, 7 Nov 2022 11:26:42 +0800 Subject: [PATCH v1] Fix the logical replication timeout during processing of DDL. The problem is when there is a DDL in a transaction that generates lots of temporary data due to rewrite rules, these temporary data will not be processed by the pgoutput - plugin. Therefore, the previous fix (f95d53e) for DML had no impact on this case. To fix this, we try to send the keepalive messages after each change is processed by walsender, not in the pgoutput - plugin. --- .../replication/logical/reorderbuffer.c | 42 +++++++++++++++ src/backend/replication/pgoutput/pgoutput.c | 54 +++---------------- 2 files changed, 48 insertions(+), 48 deletions(-) diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 22a15a482a..ee97a36842 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -2031,6 +2031,46 @@ ReorderBufferResetTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, } } +/* + * Helper function for ReorderBufferProcessTXN for updating progress. + */ +static inline void +ReorderBufferUpdateProgress(ReorderBuffer *rb, ReorderBufferTXN *txn, + ReorderBufferChange *change) +{ + LogicalDecodingContext *ctx = rb->private_data; + static int changes_count = 0; + + if (!ctx->update_progress) + return; + + Assert(!ctx->fast_forward); + + /* set output state */ + ctx->accept_writes = false; + ctx->write_xid = txn->xid; + ctx->write_location = change->lsn; + ctx->end_xact = false; + + /* + * 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 + + /* + * After continuously processing CHANGES_THRESHOLD changes, we + * try to send a keepalive message if required. + */ + if (++changes_count >= CHANGES_THRESHOLD) + { + ctx->update_progress(ctx, ctx->write_location, ctx->write_xid, false); + changes_count = 0; + } +} + /* * Helper function for ReorderBufferReplay and ReorderBufferStreamTXN. * @@ -2419,6 +2459,8 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, elog(ERROR, "tuplecid value in changequeue"); break; } + + ReorderBufferUpdateProgress(rb, txn, change); } /* speculative insertion record must be freed by now */ diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 2ecaa5b907..6aa12a39e5 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -91,8 +91,6 @@ static void send_relation_and_attrs(Relation relation, TransactionId xid, static void send_repl_origin(LogicalDecodingContext *ctx, RepOriginId origin_id, XLogRecPtr origin_lsn, bool send_origin); -static void update_replication_progress(LogicalDecodingContext *ctx, - bool skipped_xact); /* * Only 3 publication actions are used for row filtering ("insert", "update", @@ -578,7 +576,7 @@ pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, * from this transaction has been sent to the downstream. */ sent_begin_txn = txndata->sent_begin_txn; - update_replication_progress(ctx, !sent_begin_txn); + OutputPluginUpdateProgress(ctx, !sent_begin_txn); pfree(txndata); txn->output_plugin_private = NULL; @@ -617,7 +615,7 @@ static void pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn) { - update_replication_progress(ctx, false); + OutputPluginUpdateProgress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_prepare(ctx->out, txn, prepare_lsn); @@ -631,7 +629,7 @@ static void pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - update_replication_progress(ctx, false); + OutputPluginUpdateProgress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit_prepared(ctx->out, txn, commit_lsn); @@ -647,7 +645,7 @@ pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time) { - update_replication_progress(ctx, false); + OutputPluginUpdateProgress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_rollback_prepared(ctx->out, txn, prepare_end_lsn, @@ -1378,8 +1376,6 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, TupleTableSlot *old_slot = NULL; TupleTableSlot *new_slot = NULL; - update_replication_progress(ctx, false); - if (!is_publishable_relation(relation)) return; @@ -1612,8 +1608,6 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Oid *relids; TransactionId xid = InvalidTransactionId; - update_replication_progress(ctx, false); - /* Remember the xid for the change in streaming mode. See pgoutput_change. */ if (in_streaming) xid = change->txn->xid; @@ -1677,8 +1671,6 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, PGOutputData *data = (PGOutputData *) ctx->output_plugin_private; TransactionId xid = InvalidTransactionId; - update_replication_progress(ctx, false); - if (!data->messages) return; @@ -1874,7 +1866,7 @@ pgoutput_stream_commit(struct LogicalDecodingContext *ctx, Assert(!in_streaming); Assert(rbtxn_is_streamed(txn)); - update_replication_progress(ctx, false); + OutputPluginUpdateProgress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_commit(ctx->out, txn, commit_lsn); @@ -1895,7 +1887,7 @@ pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx, { Assert(rbtxn_is_streamed(txn)); - update_replication_progress(ctx, false); + OutputPluginUpdateProgress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn); OutputPluginWrite(ctx, true); @@ -2380,37 +2372,3 @@ send_repl_origin(LogicalDecodingContext *ctx, RepOriginId origin_id, } } } - -/* - * Try to update progress and send a keepalive message if too many changes were - * processed. - * - * For a large transaction, if we don't send any change to the downstream for a - * long time (exceeds the wal_receiver_timeout of standby) then it can timeout. - * This can happen when all or most of the changes are either not published or - * got filtered out. - */ -static void -update_replication_progress(LogicalDecodingContext *ctx, bool skipped_xact) -{ - static int changes_count = 0; - - /* - * We don't want to try sending a keepalive message after processing each - * change as that can have overhead. Tests revealed that there is no - * noticeable overhead in doing it after continuously processing 100 or so - * changes. - */ -#define CHANGES_THRESHOLD 100 - - /* - * If we are at the end of transaction LSN, update progress tracking. - * Otherwise, after continuously processing CHANGES_THRESHOLD changes, we - * try to send a keepalive message if required. - */ - if (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD) - { - OutputPluginUpdateProgress(ctx, skipped_xact); - changes_count = 0; - } -} -- 2.23.0.windows.1 ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2023-01-06 07:05 Ashutosh Bapat <[email protected]> parent: [email protected] <[email protected]> 0 siblings, 2 replies; 35+ messages in thread From: Ashutosh Bapat @ 2023-01-06 07:05 UTC (permalink / raw) To: [email protected] <[email protected]>; +Cc: Fabrice Chapuis <[email protected]>; Amit Kapila <[email protected]>; Euler Taveira <[email protected]>; Masahiko Sawada <[email protected]>; Hayato Kuroda (Fujitsu) <[email protected]>; Peter Smith <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> Hi Wang, Thanks for working on this. One of our customer faced a similar situation when running BDR with PostgreSQL. I tested your patch and it solves the problem. Please find some review comments below On Tue, Nov 8, 2022 at 8:34 AM [email protected] <[email protected]> wrote: > > > Attach the patch. > +/* + * Helper function for ReorderBufferProcessTXN for updating progress. + */ +static inline void +ReorderBufferUpdateProgress(ReorderBuffer *rb, ReorderBufferTXN *txn, + ReorderBufferChange *change) +{ + LogicalDecodingContext *ctx = rb->private_data; + static int changes_count = 0; It's not easy to know that a variable is static when reading the code which uses it. So it's easy to interpret code wrong. I would probably track it through logical decoding context itself OR through a global variable like other places where we track the last timestamps. But there's more below on this. + + if (!ctx->update_progress) + return; + + Assert(!ctx->fast_forward); + + /* set output state */ + ctx->accept_writes = false; + ctx->write_xid = txn->xid; + ctx->write_location = change->lsn; + ctx->end_xact = false; This patch reverts many of the changes of the previous commit which tried to fix this issue i.e. 55558df2374. end_xact was introduced by the same commit but without much explanation of that in the commit message. Its only user, WalSndUpdateProgress(), is probably making a wrong assumption as well. * 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. IIUC, WAL sender tracks the LSN of the last WAL record read in sentPtr which is sent downstream through a keep alive message. Downstream may acknowledge this LSN. So we do get ack for any LSN, not just commit LSN. So I propose removing end_xact as well. + + /* + * 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 I think a time based threashold makes more sense. What if the timeout was nearing and those 100 changes just took little more time causing a timeout? We already have a time based threashold in WalSndKeepaliveIfNecessary(). And that function is invoked after reading every WAL record in WalSndLoop(). So it does not look like it's an expensive function. If it is expensive we might want to worry about WalSndLoop as well. Does it make more sense to remove this threashold? + + /* + * After continuously processing CHANGES_THRESHOLD changes, we + * try to send a keepalive message if required. + */ + if (++changes_count >= CHANGES_THRESHOLD) + { + ctx->update_progress(ctx, ctx->write_location, ctx->write_xid, false); + changes_count = 0; + } +} + On the other thread, I mentioned that we don't have a TAP test for it. I agree with Amit's opinion there that it's hard to create a test which will timeout everywhere. I think what we need is a way to control the time required for decoding a transaction. A rough idea is to induce a small sleep after decoding every change. The amount of sleep * number of changes will help us estimate and control the amount of time taken to decode a transaction. Then we create a transaction which will take longer than the timeout threashold to decode. But that's a significant code. I don't think PostgreSQL has a facility to induce a delay at a particular place in the code. -- Best Wishes, Ashutosh Bapat ^ permalink raw reply [nested|flat] 35+ messages in thread
* Re: Logical replication timeout problem @ 2023-01-09 05:03 Amit Kapila <[email protected]> parent: Ashutosh Bapat <[email protected]> 1 sibling, 1 reply; 35+ messages in thread From: Amit Kapila @ 2023-01-09 05:03 UTC (permalink / raw) To: Ashutosh Bapat <[email protected]>; +Cc: [email protected] <[email protected]>; Fabrice Chapuis <[email protected]>; Euler Taveira <[email protected]>; Masahiko Sawada <[email protected]>; Hayato Kuroda (Fujitsu) <[email protected]>; Peter Smith <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Fri, Jan 6, 2023 at 12:35 PM Ashutosh Bapat <[email protected]> wrote: > > + > + /* > + * 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 > > I think a time based threashold makes more sense. What if the timeout was > nearing and those 100 changes just took little more time causing a timeout? We > already have a time based threashold in WalSndKeepaliveIfNecessary(). And that > function is invoked after reading every WAL record in WalSndLoop(). So it does > not look like it's an expensive function. If it is expensive we might want to > worry about WalSndLoop as well. Does it make more sense to remove this > threashold? > We have previously tried this for every change [1] and it brings noticeable overhead. In fact, even doing it for every 10 changes also had some overhead which is why we reached this threshold number. I don't think it can lead to timeout due to skipping changes but sure if we see any such report we can further fine-tune this setting or will try to make it time-based but for now I feel it would be safe to use this threshold. > + > + /* > + * After continuously processing CHANGES_THRESHOLD changes, we > + * try to send a keepalive message if required. > + */ > + if (++changes_count >= CHANGES_THRESHOLD) > + { > + ctx->update_progress(ctx, ctx->write_location, ctx->write_xid, false); > + changes_count = 0; > + } > +} > + > > On the other thread, I mentioned that we don't have a TAP test for it. > I agree with > Amit's opinion there that it's hard to create a test which will timeout > everywhere. I think what we need is a way to control the time required for > decoding a transaction. > > A rough idea is to induce a small sleep after decoding every change. The amount > of sleep * number of changes will help us estimate and control the amount of > time taken to decode a transaction. Then we create a transaction which will > take longer than the timeout threashold to decode. But that's a > significant code. I > don't think PostgreSQL has a facility to induce a delay at a particular place > in the code. > Yeah, I don't know how to induce such a delay while decoding changes. One more thing, I think it would be better to expose a new callback API via reorder buffer as suggested previously [2] similar to other reorder buffer APIs instead of directly using reorderbuffer API to invoke plugin API. [1] - https://www.postgresql.org/message-id/OS3PR01MB6275DFFDAC7A59FA148931529E209%40OS3PR01MB6275.jpnprd0... [2] - https://www.postgresql.org/message-id/CAA4eK1%2BfQjndoBOFUn9Wy0hhm3MLyUWEpcT9O7iuCELktfdBiQ%40mail.g... -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 35+ messages in thread
* RE: Logical replication timeout problem @ 2023-01-09 10:38 [email protected] <[email protected]> parent: Ashutosh Bapat <[email protected]> 1 sibling, 0 replies; 35+ messages in thread From: [email protected] @ 2023-01-09 10:38 UTC (permalink / raw) To: Ashutosh Bapat <[email protected]>; +Cc: Fabrice Chapuis <[email protected]>; Amit Kapila <[email protected]>; Euler Taveira <[email protected]>; Masahiko Sawada <[email protected]>; Hayato Kuroda (Fujitsu) <[email protected]>; Peter Smith <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Fri, Jan 6, 2023 at 15:06 PM Ashutosh Bapat <[email protected]> wrote: > Hi Wang, > Thanks for working on this. One of our customer faced a similar > situation when running BDR with PostgreSQL. > > I tested your patch and it solves the problem. > > Please find some review comments below Thanks for your testing and comments. > +/* > + * Helper function for ReorderBufferProcessTXN for updating progress. > + */ > +static inline void > +ReorderBufferUpdateProgress(ReorderBuffer *rb, ReorderBufferTXN *txn, > + ReorderBufferChange *change) > +{ > + LogicalDecodingContext *ctx = rb->private_data; > + static int changes_count = 0; > > It's not easy to know that a variable is static when reading the code which > uses it. So it's easy to interpret code wrong. I would probably track it > through logical decoding context itself OR through a global variable like other > places where we track the last timestamps. But there's more below on this. I'm not sure if we need to add global variables or member variables for a cumulative count that is only used here. How would you feel if I add some comments when declaring this static variable? > + > + if (!ctx->update_progress) > + return; > + > + Assert(!ctx->fast_forward); > + > + /* set output state */ > + ctx->accept_writes = false; > + ctx->write_xid = txn->xid; > + ctx->write_location = change->lsn; > + ctx->end_xact = false; > > This patch reverts many of the changes of the previous commit which tried to > fix this issue i.e. 55558df2374. end_xact was introduced by the same commit but > without much explanation of that in the commit message. Its only user, > WalSndUpdateProgress(), is probably making a wrong assumption as well. > > * 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. > > IIUC, WAL sender tracks the LSN of the last WAL record read in sentPtr which is > sent downstream through a keep alive message. Downstream may > acknowledge this > LSN. So we do get ack for any LSN, not just commit LSN. > > So I propose removing end_xact as well. We didn't track the lag during a transaction because it could make the calculations of lag functionality inaccurate. If we track every lsn, it could fail to record important lsn information because of WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS (see function WalSndUpdateProgress). Please see details in [1] and [2]. Regards, Wang Wei [1] - https://www.postgresql.org/message-id/OS3PR01MB62755D216245199554DDC8DB9EEA9%40OS3PR01MB6275.jpnprd0... [2] - https://www.postgresql.org/message-id/OS3PR01MB627514AE0B3040D8F55A68B99EEA9%40OS3PR01MB6275.jpnprd0... ^ permalink raw reply [nested|flat] 35+ messages in thread
* RE: Logical replication timeout problem @ 2023-01-11 10:41 [email protected] <[email protected]> parent: Amit Kapila <[email protected]> 0 siblings, 0 replies; 35+ messages in thread From: [email protected] @ 2023-01-11 10:41 UTC (permalink / raw) To: Amit Kapila <[email protected]>; Ashutosh Bapat <[email protected]>; +Cc: Fabrice Chapuis <[email protected]>; Euler Taveira <[email protected]>; Masahiko Sawada <[email protected]>; Hayato Kuroda (Fujitsu) <[email protected]>; Peter Smith <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Mon, Jan 9, 2023 at 13:04 PM Amit Kapila <[email protected]> wrote: > Thanks for your comments. > One more thing, I think it would be better to expose a new callback > API via reorder buffer as suggested previously [2] similar to other > reorder buffer APIs instead of directly using reorderbuffer API to > invoke plugin API. Yes, I agree. I think it would be better to add a new callback API on the HEAD. So, I improved the fix approach: Introduce a new optional callback to update the process. This callback function is invoked at the end inside the main loop of the function ReorderBufferProcessTXN() for each change. In this way, I think it seems that similar timeout problems could be avoided. BTW, I did the performance test for this patch. When running the SQL that reproduces the problem (refresh the materialized view in sync logical replication mode), the running time of new function pgoutput_update_progress is less than 0.1% of the total time. I think this result looks OK. Attach the new patch. Regards, Wang Wei Attachments: [application/octet-stream] v2-0001-Fix-the-logical-replication-timeout-during-proces.patch (15.5K, ../../OS3PR01MB6275D9F418DB8EAFB894373F9EFC9@OS3PR01MB6275.jpnprd01.prod.outlook.com/2-v2-0001-Fix-the-logical-replication-timeout-during-proces.patch) download | inline diff: From 001bcf625a0339169cad7599ef4adb2da484b6b3 Mon Sep 17 00:00:00 2001 From: wangw <[email protected]> Date: Mon, 7 Nov 2022 11:26:42 +0800 Subject: [PATCH v2] Fix the logical replication timeout during processing of DDL. The problem is when there is a DDL in a transaction that generates lots of temporary data due to rewrite rules, these temporary data will not be processed by the pgoutput - plugin. Therefore, the previous fix (f95d53e) for DML had no impact on this case. To fix this, we introduced a new optional output plugin callback - 'update_progress_cb'. This callback is called to update the process after each change has been processed during sending data of a transaction (and its subtransactions) to the output plugin. For pgoutput, in this callback we try to update progress and send keep-alive messages when too many changes have been processed. --- doc/src/sgml/logicaldecoding.sgml | 20 +++- src/backend/replication/logical/logical.c | 52 +++++++++++ .../replication/logical/reorderbuffer.c | 2 + src/backend/replication/pgoutput/pgoutput.c | 93 +++++++++---------- src/include/replication/output_plugin.h | 9 ++ src/include/replication/reorderbuffer.h | 11 +++ src/tools/pgindent/typedefs.list | 2 + 7 files changed, 139 insertions(+), 50 deletions(-) diff --git a/doc/src/sgml/logicaldecoding.sgml b/doc/src/sgml/logicaldecoding.sgml index 4cf863a76f..619cba36d3 100644 --- a/doc/src/sgml/logicaldecoding.sgml +++ b/doc/src/sgml/logicaldecoding.sgml @@ -473,6 +473,7 @@ typedef struct OutputPluginCallbacks LogicalDecodeStreamChangeCB stream_change_cb; LogicalDecodeStreamMessageCB stream_message_cb; LogicalDecodeStreamTruncateCB stream_truncate_cb; + LogicalDecodeUpdateProgressCB update_progress_cb; } OutputPluginCallbacks; typedef void (*LogicalOutputPluginInit) (struct OutputPluginCallbacks *cb); @@ -481,8 +482,8 @@ typedef void (*LogicalOutputPluginInit) (struct OutputPluginCallbacks *cb); and <function>commit_cb</function> callbacks are required, while <function>startup_cb</function>, <function>filter_by_origin_cb</function>, <function>truncate_cb</function>, - and <function>shutdown_cb</function> are optional. - If <function>truncate_cb</function> is not set but a + <function>shutdown_cb</function>, and <function>update_progress_cb</function> + are optional. If <function>truncate_cb</function> is not set but a <command>TRUNCATE</command> is to be decoded, the action will be ignored. </para> @@ -1040,6 +1041,21 @@ typedef void (*LogicalDecodeStreamTruncateCB) (struct LogicalDecodingContext *ct </para> </sect3> + <sect3 id="logicaldecoding-output-plugin-update-progress"> + <title>Update Progress Callback</title> + + <para> + The optional <function>update_progress_cb</function> callback is called + after handling every change. This callback is to update the process + during sending data of a transaction (and its subtransactions) to the + output plugin. +<programlisting> +typedef void (*LogicalDecodeUpdateProgressCB) (struct LogicalDecodingContext *ctx, + ReorderBufferTXN *txn); +</programlisting> + </para> + </sect3> + </sect2> <sect2 id="logicaldecoding-output-plugin-output"> diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 52d1fe6269..0e5f02c2ca 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -93,6 +93,11 @@ static void stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *tx static void stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change); +/* update progress callback */ +static void update_progress_cb_wrapper(ReorderBuffer *cache, + ReorderBufferTXN *txn, + ReorderBufferChange *change); + static void LoadOutputPlugin(OutputPluginCallbacks *callbacks, const char *plugin); /* @@ -278,6 +283,11 @@ StartupDecodingContext(List *output_plugin_options, ctx->reorder->commit_prepared = commit_prepared_cb_wrapper; ctx->reorder->rollback_prepared = rollback_prepared_cb_wrapper; + /* + * Callback to support updating progress. + */ + ctx->reorder->update_progress = update_progress_cb_wrapper; + ctx->out = makeStringInfo(); ctx->prepare_write = prepare_write; ctx->write = do_write; @@ -1582,6 +1592,48 @@ stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, error_context_stack = errcallback.previous; } +static void +update_progress_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, + ReorderBufferChange *change) +{ + LogicalDecodingContext *ctx = cache->private_data; + LogicalErrorCallbackState state; + ErrorContextCallback errcallback; + + Assert(!ctx->fast_forward); + + if (!ctx->callbacks.update_progress_cb) + return; + + /* Push callback + info on the error context stack */ + state.ctx = ctx; + state.callback_name = "update_progress"; + state.report_location = change->lsn; + errcallback.callback = output_plugin_error_callback; + errcallback.arg = (void *) &state; + errcallback.previous = error_context_stack; + error_context_stack = &errcallback; + + /* set output state */ + ctx->accept_writes = false; + ctx->write_xid = txn->xid; + + /* + * Report this change's lsn so replies from clients can give an up-to-date + * answer. This won't ever be enough (and shouldn't be!) to confirm + * receipt of this transaction, but it might allow another transaction's + * commit to be confirmed with one message. + */ + ctx->write_location = change->lsn; + + ctx->end_xact = false; + + ctx->callbacks.update_progress_cb(ctx, txn); + + /* Pop the error context stack */ + error_context_stack = errcallback.previous; +} + /* * Set the required catalog xmin horizon for historic snapshots in the current * replication slot. diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 54ee824e6c..2e134bd011 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -2446,6 +2446,8 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, elog(ERROR, "tuplecid value in changequeue"); break; } + + rb->update_progress(rb, txn, change); } /* speculative insertion record must be freed by now */ diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 19c10c028f..865384897e 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -78,6 +78,8 @@ static void pgoutput_stream_commit(struct LogicalDecodingContext *ctx, XLogRecPtr commit_lsn); static void pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn); +static void pgoutput_update_progress(LogicalDecodingContext *ctx, + ReorderBufferTXN *txn); static bool publications_valid; static bool in_streaming; @@ -92,8 +94,6 @@ static void send_relation_and_attrs(Relation relation, TransactionId xid, static void send_repl_origin(LogicalDecodingContext *ctx, RepOriginId origin_id, XLogRecPtr origin_lsn, bool send_origin); -static void update_replication_progress(LogicalDecodingContext *ctx, - bool skipped_xact); /* * Only 3 publication actions are used for row filtering ("insert", "update", @@ -276,6 +276,8 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb) cb->stream_truncate_cb = pgoutput_truncate; /* transaction streaming - two-phase commit */ cb->stream_prepare_cb = pgoutput_stream_prepare_txn; + + cb->update_progress_cb = pgoutput_update_progress; } static void @@ -586,7 +588,7 @@ pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, * from this transaction has been sent to the downstream. */ sent_begin_txn = txndata->sent_begin_txn; - update_replication_progress(ctx, !sent_begin_txn); + OutputPluginUpdateProgress(ctx, !sent_begin_txn); pfree(txndata); txn->output_plugin_private = NULL; @@ -625,7 +627,7 @@ static void pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn) { - update_replication_progress(ctx, false); + OutputPluginUpdateProgress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_prepare(ctx->out, txn, prepare_lsn); @@ -639,7 +641,7 @@ static void pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - update_replication_progress(ctx, false); + OutputPluginUpdateProgress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit_prepared(ctx->out, txn, commit_lsn); @@ -655,7 +657,7 @@ pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time) { - update_replication_progress(ctx, false); + OutputPluginUpdateProgress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_rollback_prepared(ctx->out, txn, prepare_end_lsn, @@ -1386,8 +1388,6 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, TupleTableSlot *old_slot = NULL; TupleTableSlot *new_slot = NULL; - update_replication_progress(ctx, false); - if (!is_publishable_relation(relation)) return; @@ -1622,8 +1622,6 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Oid *relids; TransactionId xid = InvalidTransactionId; - update_replication_progress(ctx, false); - /* Remember the xid for the change in streaming mode. See pgoutput_change. */ if (in_streaming) xid = change->txn->xid; @@ -1687,8 +1685,6 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, PGOutputData *data = (PGOutputData *) ctx->output_plugin_private; TransactionId xid = InvalidTransactionId; - update_replication_progress(ctx, false); - if (!data->messages) return; @@ -1888,7 +1884,7 @@ pgoutput_stream_commit(struct LogicalDecodingContext *ctx, Assert(!in_streaming); Assert(rbtxn_is_streamed(txn)); - update_replication_progress(ctx, false); + OutputPluginUpdateProgress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_commit(ctx->out, txn, commit_lsn); @@ -1909,12 +1905,47 @@ pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx, { Assert(rbtxn_is_streamed(txn)); - update_replication_progress(ctx, false); + OutputPluginUpdateProgress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn); OutputPluginWrite(ctx, true); } +/* + * Update progress callback + * + * Try to update progress and send a keepalive message if too many changes were + * processed. + * + * For a large transaction, if we don't send any change to the downstream for a + * long time (exceeds the wal_receiver_timeout of standby) then it can timeout. + * This can happen when all or most of the changes are either not published or + * got filtered out. + */ +static void +pgoutput_update_progress(LogicalDecodingContext *ctx, ReorderBufferTXN *txn) +{ + static int changes_count = 0; + + /* + * We don't want to try sending a keepalive message after processing each + * change as that can have overhead. Tests revealed that there is no + * noticeable overhead in doing it after continuously processing 100 or so + * changes. + */ +#define CHANGES_THRESHOLD 100 + + /* + * After continuously processing CHANGES_THRESHOLD changes, we + * try to send a keepalive message if required. + */ + if (++changes_count >= CHANGES_THRESHOLD) + { + OutputPluginUpdateProgress(ctx, false); + changes_count = 0; + } +} + /* * Initialize the relation schema sync cache for a decoding session. * @@ -2409,37 +2440,3 @@ send_repl_origin(LogicalDecodingContext *ctx, RepOriginId origin_id, } } } - -/* - * Try to update progress and send a keepalive message if too many changes were - * processed. - * - * For a large transaction, if we don't send any change to the downstream for a - * long time (exceeds the wal_receiver_timeout of standby) then it can timeout. - * This can happen when all or most of the changes are either not published or - * got filtered out. - */ -static void -update_replication_progress(LogicalDecodingContext *ctx, bool skipped_xact) -{ - static int changes_count = 0; - - /* - * We don't want to try sending a keepalive message after processing each - * change as that can have overhead. Tests revealed that there is no - * noticeable overhead in doing it after continuously processing 100 or so - * changes. - */ -#define CHANGES_THRESHOLD 100 - - /* - * If we are at the end of transaction LSN, update progress tracking. - * Otherwise, after continuously processing CHANGES_THRESHOLD changes, we - * try to send a keepalive message if required. - */ - if (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD) - { - OutputPluginUpdateProgress(ctx, skipped_xact); - changes_count = 0; - } -} diff --git a/src/include/replication/output_plugin.h b/src/include/replication/output_plugin.h index 2d89d26586..e877bcac93 100644 --- a/src/include/replication/output_plugin.h +++ b/src/include/replication/output_plugin.h @@ -210,6 +210,12 @@ typedef void (*LogicalDecodeStreamTruncateCB) (struct LogicalDecodingContext *ct Relation relations[], ReorderBufferChange *change); +/* + * Callback for updating progress. + */ +typedef void (*LogicalDecodeUpdateProgressCB) (struct LogicalDecodingContext *ctx, + ReorderBufferTXN *txn); + /* * Output plugin callbacks */ @@ -240,6 +246,9 @@ typedef struct OutputPluginCallbacks LogicalDecodeStreamChangeCB stream_change_cb; LogicalDecodeStreamMessageCB stream_message_cb; LogicalDecodeStreamTruncateCB stream_truncate_cb; + + /* update progress */ + LogicalDecodeUpdateProgressCB update_progress_cb; } OutputPluginCallbacks; /* Functions in replication/logical/logical.c */ diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index f6c4dd75db..5897b27b42 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -525,6 +525,12 @@ typedef void (*ReorderBufferStreamTruncateCB) ( Relation relations[], ReorderBufferChange *change); +/* update progress callback signature */ +typedef void (*ReorderBufferUpdateProgressCB) ( + ReorderBuffer *rb, + ReorderBufferTXN *txn, + ReorderBufferChange *change); + struct ReorderBuffer { /* @@ -588,6 +594,11 @@ struct ReorderBuffer ReorderBufferStreamMessageCB stream_message; ReorderBufferStreamTruncateCB stream_truncate; + /* + * Callbacks to be called when updating progress. + */ + ReorderBufferUpdateProgressCB update_progress; + /* * Pointer that will be passed untouched to the callbacks. */ diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 23bafec5f7..96d940e05c 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1455,6 +1455,7 @@ LogicalDecodeStreamStartCB LogicalDecodeStreamStopCB LogicalDecodeStreamTruncateCB LogicalDecodeTruncateCB +LogicalDecodeUpdateProgressCB LogicalDecodingContext LogicalDecodingMode LogicalErrorCallbackState @@ -2310,6 +2311,7 @@ ReorderBufferToastEnt ReorderBufferTupleBuf ReorderBufferTupleCidEnt ReorderBufferTupleCidKey +ReorderBufferUpdateProgressCB ReorderTuple RepOriginId ReparameterizeForeignPathByChild_function -- 2.23.0.windows.1 ^ permalink raw reply [nested|flat] 35+ messages in thread
end of thread, other threads:[~2023-01-11 10:41 UTC | newest] Thread overview: 35+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2022-03-04 04:18 [PATCH v11 1/5] Correctly update contfol file at the end of archive recovery Kyotaro Horiguchi <[email protected]> 2022-03-22 01:55 RE: Logical replication timeout problem [email protected] <[email protected]> 2022-03-24 10:32 ` Re: Logical replication timeout problem Amit Kapila <[email protected]> 2022-03-25 03:20 ` RE: Logical replication timeout problem [email protected] <[email protected]> 2022-03-25 05:23 ` RE: Logical replication timeout problem [email protected] <[email protected]> 2022-03-25 06:19 ` Re: Logical replication timeout problem Masahiko Sawada <[email protected]> 2022-03-25 08:32 ` Re: Logical replication timeout problem Amit Kapila <[email protected]> 2022-03-29 05:07 ` Re: Logical replication timeout problem Masahiko Sawada <[email protected]> 2022-03-25 10:19 ` RE: Logical replication timeout problem [email protected] <[email protected]> 2022-03-28 01:55 ` RE: Logical replication timeout problem [email protected] <[email protected]> 2022-03-28 06:11 ` RE: Logical replication timeout problem [email protected] <[email protected]> 2022-03-28 06:27 ` Re: Logical replication timeout problem Amit Kapila <[email protected]> 2022-03-29 01:29 ` RE: Logical replication timeout problem [email protected] <[email protected]> 2022-03-29 01:44 ` RE: Logical replication timeout problem [email protected] <[email protected]> 2022-03-30 07:54 ` RE: Logical replication timeout problem [email protected] <[email protected]> 2022-03-30 08:59 ` Re: Logical replication timeout problem Amit Kapila <[email protected]> 2022-03-31 12:24 ` Re: Logical replication timeout problem Masahiko Sawada <[email protected]> 2022-04-01 02:00 ` Re: Logical replication timeout problem Amit Kapila <[email protected]> 2022-04-01 02:03 ` Re: Logical replication timeout problem Euler Taveira <[email protected]> 2022-04-01 02:27 ` Re: Logical replication timeout problem Amit Kapila <[email protected]> 2022-04-01 02:57 ` Re: Logical replication timeout problem Euler Taveira <[email protected]> 2022-04-01 04:08 ` Re: Logical replication timeout problem Amit Kapila <[email protected]> 2022-04-06 05:39 ` RE: Logical replication timeout problem [email protected] <[email protected]> 2022-04-06 05:58 ` Re: Logical replication timeout problem Amit Kapila <[email protected]> 2022-10-18 14:35 ` Re: Logical replication timeout problem Fabrice Chapuis <[email protected]> 2022-10-19 08:15 ` RE: Logical replication timeout problem [email protected] <[email protected]> 2022-10-20 05:46 ` Re: Logical replication timeout problem Fabrice Chapuis <[email protected]> 2022-10-20 07:08 ` RE: Logical replication timeout problem [email protected] <[email protected]> 2022-11-04 10:13 ` Re: Logical replication timeout problem Fabrice Chapuis <[email protected]> 2022-11-08 03:04 ` RE: Logical replication timeout problem [email protected] <[email protected]> 2023-01-06 07:05 ` Re: Logical replication timeout problem Ashutosh Bapat <[email protected]> 2023-01-09 05:03 ` Re: Logical replication timeout problem Amit Kapila <[email protected]> 2023-01-11 10:41 ` RE: Logical replication timeout problem [email protected] <[email protected]> 2023-01-09 10:38 ` RE: Logical replication timeout problem [email protected] <[email protected]> 2022-03-31 02:26 ` RE: Logical replication timeout problem [email protected] <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox