From 8afba828a3c1cf3e23a85f9c9f034bbc90977237 Mon Sep 17 00:00:00 2001 From: Peter Smith Date: Fri, 28 Jul 2023 14:45:00 +1000 Subject: [PATCH v1] Add LogicalRepWorkerType enum --- src/backend/replication/logical/launcher.c | 4 ++++ src/include/replication/worker_internal.h | 20 ++++++++++++++++---- 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index 542af7d..702732f 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -427,6 +427,10 @@ retry: } /* Prepare the worker slot. */ + worker->type = + OidIsValid(relid) ? LR_WORKER_TABLESYNC : + is_parallel_apply_worker ? LR_WORKER_APPLY_PARALLEL : + LR_WORKER_APPLY; worker->launch_time = now; worker->in_use = true; worker->generation++; diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 343e781..97447dc 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -27,8 +27,19 @@ #include "storage/spin.h" +/* Different kinds of workers */ +typedef enum LogicalRepWorkerType +{ + LR_WORKER_TABLESYNC, + LR_WORKER_APPLY, + LR_WORKER_APPLY_PARALLEL +} LogicalRepWorkerType; + typedef struct LogicalRepWorker { + /* What type of worker is this? */ + LogicalRepWorkerType type; + /* Time at which this worker was launched. */ TimestampTz launch_time; @@ -305,19 +316,20 @@ extern void pa_decr_and_wait_stream_block(void); extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn); -#define isParallelApplyWorker(worker) ((worker)->leader_pid != InvalidPid) +#define isLeaderApplyWorker(worker) ((worker)->type == LR_WORKER_APPLY) +#define isParallelApplyWorker(worker) ((worker)->type == LR_WORKER_APPLY_PARALLEL) +#define isTablesyncWorker(worker) ((worker)->type == LR_WORKER_TABLESYNC) static inline bool am_tablesync_worker(void) { - return OidIsValid(MyLogicalRepWorker->relid); + return isTablesyncWorker(MyLogicalRepWorker); } static inline bool am_leader_apply_worker(void) { - return (!am_tablesync_worker() && - !isParallelApplyWorker(MyLogicalRepWorker)); + return isLeaderApplyWorker(MyLogicalRepWorker); } static inline bool -- 1.8.3.1