From e8319f864d46cac915f4f0caa901861ecd645654 Mon Sep 17 00:00:00 2001 From: Wang Wei Date: Fri, 10 Mar 2023 13:56:26 +0800 Subject: [PATCH v7 4/6] Rename the function WalSndUpdateProgress() to WalSndUpdateProgressAndKeepalive() --- src/backend/replication/logical/logical.c | 22 +++++++++++----------- src/backend/replication/walsender.c | 18 +++++++++--------- src/include/replication/logical.h | 14 +++++++------- src/tools/pgindent/typedefs.list | 2 +- 4 files changed, 28 insertions(+), 28 deletions(-) diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index c3ec97a0a6..a33b14792b 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -156,7 +156,7 @@ StartupDecodingContext(List *output_plugin_options, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, - LogicalOutputPluginWriterUpdateProgress update_progress) + LogicalOutputPluginWriterUpdateProgressAndKeepalive update_progress_and_keepalive) { ReplicationSlot *slot; MemoryContext context, @@ -292,7 +292,7 @@ StartupDecodingContext(List *output_plugin_options, ctx->out = makeStringInfo(); ctx->prepare_write = prepare_write; ctx->write = do_write; - ctx->update_progress = update_progress; + ctx->update_progress_and_keepalive = update_progress_and_keepalive; ctx->output_plugin_options = output_plugin_options; @@ -316,7 +316,7 @@ StartupDecodingContext(List *output_plugin_options, * marking WAL reserved beforehand. In that scenario, it's up to the * caller to guarantee that WAL remains available. * xl_routine -- XLogReaderRoutine for underlying XLogReader - * prepare_write, do_write, update_progress -- + * prepare_write, do_write, update_progress_and_keepalive -- * callbacks that perform the use-case dependent, actual, work. * * Needs to be called while in a memory context that's at least as long lived @@ -334,7 +334,7 @@ CreateInitDecodingContext(const char *plugin, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, - LogicalOutputPluginWriterUpdateProgress update_progress) + LogicalOutputPluginWriterUpdateProgressAndKeepalive update_progress_and_keepalive) { TransactionId xmin_horizon = InvalidTransactionId; ReplicationSlot *slot; @@ -435,7 +435,7 @@ CreateInitDecodingContext(const char *plugin, ctx = StartupDecodingContext(NIL, restart_lsn, xmin_horizon, need_full_snapshot, false, xl_routine, prepare_write, do_write, - update_progress); + update_progress_and_keepalive); /* call output plugin initialization callback */ old_context = MemoryContextSwitchTo(ctx->context); @@ -475,7 +475,7 @@ CreateInitDecodingContext(const char *plugin, * xl_routine * XLogReaderRoutine used by underlying xlogreader * - * prepare_write, do_write, update_progress + * prepare_write, do_write, update_progress_and_keepalive * callbacks that have to be filled to perform the use-case dependent, * actual work. * @@ -493,7 +493,7 @@ CreateDecodingContext(XLogRecPtr start_lsn, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, - LogicalOutputPluginWriterUpdateProgress update_progress) + LogicalOutputPluginWriterUpdateProgressAndKeepalive update_progress_and_keepalive) { LogicalDecodingContext *ctx; ReplicationSlot *slot; @@ -547,7 +547,7 @@ CreateDecodingContext(XLogRecPtr start_lsn, ctx = StartupDecodingContext(output_plugin_options, start_lsn, InvalidTransactionId, false, fast_forward, xl_routine, prepare_write, - do_write, update_progress); + do_write, update_progress_and_keepalive); /* call output plugin initialization callback */ old_context = MemoryContextSwitchTo(ctx->context); @@ -688,11 +688,11 @@ void OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx, bool skipped_xact) { - if (!ctx->update_progress) + if (!ctx->update_progress_and_keepalive) return; - ctx->update_progress(ctx, ctx->write_location, ctx->write_xid, - skipped_xact); + ctx->update_progress_and_keepalive(ctx, ctx->write_location, ctx->write_xid, + skipped_xact); } /* diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index a22ed48e33..7ac1d7430e 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -250,8 +250,8 @@ 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, - bool skipped_xact); +static void WalSndUpdateProgressAndKeepalive(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, + bool skipped_xact); static XLogRecPtr WalSndWaitForWal(XLogRecPtr loc); static void LagTrackerWrite(XLogRecPtr lsn, TimestampTz local_flush_time); static TimeOffset LagTrackerRead(int head, XLogRecPtr lsn, TimestampTz now); @@ -1126,7 +1126,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd) .segment_open = WalSndSegmentOpen, .segment_close = wal_segment_close), WalSndPrepareWrite, WalSndWriteData, - WalSndUpdateProgress); + WalSndUpdateProgressAndKeepalive); /* * Signal that we don't need the timeout mechanism. We're just @@ -1285,7 +1285,7 @@ StartLogicalReplication(StartReplicationCmd *cmd) .segment_open = WalSndSegmentOpen, .segment_close = wal_segment_close), WalSndPrepareWrite, WalSndWriteData, - WalSndUpdateProgress); + WalSndUpdateProgressAndKeepalive); xlogreader = logical_decoding_ctx->reader; WalSndSetState(WALSNDSTATE_CATCHUP); @@ -1458,15 +1458,15 @@ WalSndSendPending(void) } /* - * LogicalDecodingContext 'update_progress' callback. + * LogicalDecodingContext 'update_progress_and_keepalive' callback. * * Write the current position to the lag tracker (see XLogSendPhysical). * * When skipping empty transactions, send a keepalive message if necessary. */ static void -WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, - bool skipped_xact) +WalSndUpdateProgressAndKeepalive(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, + bool skipped_xact) { static TimestampTz sendTime = 0; TimestampTz now = GetCurrentTimestamp(); @@ -3054,8 +3054,8 @@ XLogSendLogical(void) { /* * Note the lack of any call to LagTrackerWrite() which is handled by - * WalSndUpdateProgress which is called by output plugin through - * logical decoding write api. + * WalSndUpdateProgressAndKeepalive which is called by output plugin + * through logical decoding write api. */ LogicalDecodingProcessRecord(logical_decoding_ctx, logical_decoding_ctx->reader); diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h index 5f49554ea0..c39a619b98 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -24,10 +24,10 @@ typedef void (*LogicalOutputPluginWriterWrite) (struct LogicalDecodingContext *l typedef LogicalOutputPluginWriterWrite LogicalOutputPluginWriterPrepareWrite; -typedef void (*LogicalOutputPluginWriterUpdateProgress) (struct LogicalDecodingContext *lr, - XLogRecPtr Ptr, - TransactionId xid, - bool skipped_xact +typedef void (*LogicalOutputPluginWriterUpdateProgressAndKeepalive) (struct LogicalDecodingContext *lr, + XLogRecPtr Ptr, + TransactionId xid, + bool skipped_xact ); typedef struct LogicalDecodingContext @@ -63,7 +63,7 @@ typedef struct LogicalDecodingContext */ LogicalOutputPluginWriterPrepareWrite prepare_write; LogicalOutputPluginWriterWrite write; - LogicalOutputPluginWriterUpdateProgress update_progress; + LogicalOutputPluginWriterUpdateProgressAndKeepalive update_progress_and_keepalive; /* * Output buffer. @@ -121,14 +121,14 @@ extern LogicalDecodingContext *CreateInitDecodingContext(const char *plugin, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, - LogicalOutputPluginWriterUpdateProgress update_progress); + LogicalOutputPluginWriterUpdateProgressAndKeepalive update_progress_and_keepalive); extern LogicalDecodingContext *CreateDecodingContext(XLogRecPtr start_lsn, List *output_plugin_options, bool fast_forward, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, - LogicalOutputPluginWriterUpdateProgress update_progress); + LogicalOutputPluginWriterUpdateProgressAndKeepalive update_progress_and_keepalive); extern void DecodingContextFindStartpoint(LogicalDecodingContext *ctx); extern bool DecodingContextReady(LogicalDecodingContext *ctx); extern void FreeDecodingContext(LogicalDecodingContext *ctx); diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 86a9303bf5..53fd0fe621 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1469,7 +1469,7 @@ LogicalDecodingContext LogicalErrorCallbackState LogicalOutputPluginInit LogicalOutputPluginWriterPrepareWrite -LogicalOutputPluginWriterUpdateProgress +LogicalOutputPluginWriterUpdateProgressAndKeepalive LogicalOutputPluginWriterWrite LogicalRepBeginData LogicalRepCommitData -- 2.39.1.windows.1