Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : * launcher.c
3 : : * PostgreSQL logical replication worker launcher process
4 : : *
5 : : * Copyright (c) 2016-2026, PostgreSQL Global Development Group
6 : : *
7 : : * IDENTIFICATION
8 : : * src/backend/replication/logical/launcher.c
9 : : *
10 : : * NOTES
11 : : * This module contains the logical replication worker launcher which
12 : : * uses the background worker infrastructure to start the logical
13 : : * replication workers for every enabled subscription.
14 : : *
15 : : *-------------------------------------------------------------------------
16 : : */
17 : :
18 : : #include "postgres.h"
19 : :
20 : : #include "access/heapam.h"
21 : : #include "access/htup.h"
22 : : #include "access/htup_details.h"
23 : : #include "access/tableam.h"
24 : : #include "access/xact.h"
25 : : #include "catalog/pg_subscription.h"
26 : : #include "catalog/pg_subscription_rel.h"
27 : : #include "funcapi.h"
28 : : #include "lib/dshash.h"
29 : : #include "miscadmin.h"
30 : : #include "pgstat.h"
31 : : #include "postmaster/bgworker.h"
32 : : #include "postmaster/interrupt.h"
33 : : #include "replication/logicallauncher.h"
34 : : #include "replication/origin.h"
35 : : #include "replication/slot.h"
36 : : #include "replication/walreceiver.h"
37 : : #include "replication/worker_internal.h"
38 : : #include "storage/ipc.h"
39 : : #include "storage/proc.h"
40 : : #include "storage/procarray.h"
41 : : #include "storage/subsystems.h"
42 : : #include "tcop/tcopprot.h"
43 : : #include "utils/builtins.h"
44 : : #include "utils/memutils.h"
45 : : #include "utils/pg_lsn.h"
46 : : #include "utils/snapmgr.h"
47 : : #include "utils/syscache.h"
48 : : #include "utils/wait_event.h"
49 : :
50 : : /* max sleep time between cycles (3min) */
51 : : #define DEFAULT_NAPTIME_PER_CYCLE 180000L
52 : :
53 : : /* GUC variables */
54 : : int max_logical_replication_workers = 4;
55 : : int max_sync_workers_per_subscription = 2;
56 : : int max_parallel_apply_workers_per_subscription = 2;
57 : :
58 : : LogicalRepWorker *MyLogicalRepWorker = NULL;
59 : :
60 : : typedef struct LogicalRepCtxStruct
61 : : {
62 : : /* Supervisor process. */
63 : : pid_t launcher_pid;
64 : :
65 : : /* Hash table holding last start times of subscriptions' apply workers. */
66 : : dsa_handle last_start_dsa;
67 : : dshash_table_handle last_start_dsh;
68 : :
69 : : /* Background workers. */
70 : : LogicalRepWorker workers[FLEXIBLE_ARRAY_MEMBER];
71 : : } LogicalRepCtxStruct;
72 : :
73 : : static LogicalRepCtxStruct *LogicalRepCtx;
74 : :
75 : : static void ApplyLauncherShmemRequest(void *arg);
76 : : static void ApplyLauncherShmemInit(void *arg);
77 : :
78 : : const ShmemCallbacks ApplyLauncherShmemCallbacks = {
79 : : .request_fn = ApplyLauncherShmemRequest,
80 : : .init_fn = ApplyLauncherShmemInit,
81 : : };
82 : :
83 : : /* an entry in the last-start-times shared hash table */
84 : : typedef struct LauncherLastStartTimesEntry
85 : : {
86 : : Oid subid; /* OID of logrep subscription (hash key) */
87 : : TimestampTz last_start_time; /* last time its apply worker was started */
88 : : } LauncherLastStartTimesEntry;
89 : :
90 : : /* parameters for the last-start-times shared hash table */
91 : : static const dshash_parameters dsh_params = {
92 : : sizeof(Oid),
93 : : sizeof(LauncherLastStartTimesEntry),
94 : : dshash_memcmp,
95 : : dshash_memhash,
96 : : dshash_memcpy,
97 : : LWTRANCHE_LAUNCHER_HASH
98 : : };
99 : :
100 : : static dsa_area *last_start_times_dsa = NULL;
101 : : static dshash_table *last_start_times = NULL;
102 : :
103 : : static bool on_commit_launcher_wakeup = false;
104 : :
105 : :
106 : : static void logicalrep_launcher_onexit(int code, Datum arg);
107 : : static void logicalrep_worker_onexit(int code, Datum arg);
108 : : static void logicalrep_worker_detach(void);
109 : : static void logicalrep_worker_cleanup(LogicalRepWorker *worker);
110 : : static int logicalrep_pa_worker_count(Oid subid);
111 : : static void logicalrep_launcher_attach_dshmem(void);
112 : : static void ApplyLauncherSetWorkerStartTime(Oid subid, TimestampTz start_time);
113 : : static TimestampTz ApplyLauncherGetWorkerStartTime(Oid subid);
114 : : static void compute_min_nonremovable_xid(LogicalRepWorker *worker, TransactionId *xmin);
115 : : static bool acquire_conflict_slot_if_exists(void);
116 : : static void update_conflict_slot_xmin(TransactionId new_xmin);
117 : : static void init_conflict_slot_xmin(void);
118 : :
119 : :
120 : : /*
121 : : * Load the list of subscriptions.
122 : : *
123 : : * Only the fields interesting for worker start/stop functions are filled for
124 : : * each subscription.
125 : : */
126 : : static List *
127 : 3335 : get_subscription_list(void)
128 : : {
129 : 3335 : List *res = NIL;
130 : : Relation rel;
131 : : TableScanDesc scan;
132 : : HeapTuple tup;
133 : : MemoryContext resultcxt;
134 : :
135 : : /* This is the context that we will allocate our output data in */
136 : 3335 : resultcxt = CurrentMemoryContext;
137 : :
138 : : /*
139 : : * Start a transaction so we can access pg_subscription.
140 : : */
141 : 3335 : StartTransactionCommand();
142 : :
143 : 3335 : rel = table_open(SubscriptionRelationId, AccessShareLock);
144 : 3335 : scan = table_beginscan_catalog(rel, 0, NULL);
145 : :
146 [ + + ]: 4399 : while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
147 : : {
148 : 1064 : Form_pg_subscription subform = (Form_pg_subscription) GETSTRUCT(tup);
149 : : Subscription *sub;
150 : : MemoryContext oldcxt;
151 : :
152 : : /*
153 : : * Allocate our results in the caller's context, not the
154 : : * transaction's. We do this inside the loop, and restore the original
155 : : * context at the end, so that leaky things like heap_getnext() are
156 : : * not called in a potentially long-lived context.
157 : : */
158 : 1064 : oldcxt = MemoryContextSwitchTo(resultcxt);
159 : :
160 : 1064 : sub = palloc0_object(Subscription);
161 : 1064 : sub->oid = subform->oid;
162 : 1064 : sub->dbid = subform->subdbid;
163 : 1064 : sub->owner = subform->subowner;
164 : 1064 : sub->enabled = subform->subenabled;
165 : 1064 : sub->name = pstrdup(NameStr(subform->subname));
166 : 1064 : sub->retaindeadtuples = subform->subretaindeadtuples;
167 : 1064 : sub->retentionactive = subform->subretentionactive;
168 : : /* We don't fill fields we are not interested in. */
169 : :
170 : 1064 : res = lappend(res, sub);
171 : 1064 : MemoryContextSwitchTo(oldcxt);
172 : : }
173 : :
174 : 3334 : table_endscan(scan);
175 : 3334 : table_close(rel, AccessShareLock);
176 : :
177 : 3334 : CommitTransactionCommand();
178 : :
179 : 3334 : return res;
180 : : }
181 : :
182 : : /*
183 : : * Wait for a background worker to start up and attach to the shmem context.
184 : : *
185 : : * This is only needed for cleaning up the shared memory in case the worker
186 : : * fails to attach.
187 : : *
188 : : * Returns whether the attach was successful.
189 : : */
190 : : static bool
191 : 474 : WaitForReplicationWorkerAttach(LogicalRepWorker *worker,
192 : : uint16 generation,
193 : : BackgroundWorkerHandle *handle)
194 : : {
195 : 474 : bool result = false;
196 : 474 : bool dropped_latch = false;
197 : :
198 : : for (;;)
199 : 1139 : {
200 : : BgwHandleStatus status;
201 : : pid_t pid;
202 : : int rc;
203 : :
204 [ - + ]: 1613 : CHECK_FOR_INTERRUPTS();
205 : :
206 : 1613 : LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
207 : :
208 : : /* Worker either died or has started. Return false if died. */
209 [ + + + + ]: 1613 : if (!worker->in_use || worker->proc)
210 : : {
211 : 470 : result = worker->in_use;
212 : 470 : LWLockRelease(LogicalRepWorkerLock);
213 : 470 : break;
214 : : }
215 : :
216 : 1143 : LWLockRelease(LogicalRepWorkerLock);
217 : :
218 : : /* Check if worker has died before attaching, and clean up after it. */
219 : 1143 : status = GetBackgroundWorkerPid(handle, &pid);
220 : :
221 [ - + ]: 1143 : if (status == BGWH_STOPPED)
222 : : {
223 : 0 : LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
224 : : /* Ensure that this was indeed the worker we waited for. */
225 [ # # ]: 0 : if (generation == worker->generation)
226 : 0 : logicalrep_worker_cleanup(worker);
227 : 0 : LWLockRelease(LogicalRepWorkerLock);
228 : 0 : break; /* result is already false */
229 : : }
230 : :
231 : : /*
232 : : * We need timeout because we generally don't get notified via latch
233 : : * about the worker attach. But we don't expect to have to wait long.
234 : : */
235 : 1143 : rc = WaitLatch(MyLatch,
236 : : WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
237 : : 10L, WAIT_EVENT_BGWORKER_STARTUP);
238 : :
239 [ + + ]: 1143 : if (rc & WL_LATCH_SET)
240 : : {
241 : 540 : ResetLatch(MyLatch);
242 [ + + ]: 540 : CHECK_FOR_INTERRUPTS();
243 : 536 : dropped_latch = true;
244 : : }
245 : : }
246 : :
247 : : /*
248 : : * If we had to clear a latch event in order to wait, be sure to restore
249 : : * it before exiting. Otherwise caller may miss events.
250 : : */
251 [ + + ]: 470 : if (dropped_latch)
252 : 468 : SetLatch(MyLatch);
253 : :
254 : 470 : return result;
255 : : }
256 : :
257 : : /*
258 : : * Walks the workers array and searches for one that matches given worker type,
259 : : * subscription id, and relation id.
260 : : *
261 : : * For both apply workers and sequencesync workers, the relid should be set to
262 : : * InvalidOid, as these workers handle changes across all tables and sequences
263 : : * respectively, rather than targeting a specific relation. For tablesync
264 : : * workers, the relid should be set to the OID of the relation being
265 : : * synchronized.
266 : : */
267 : : LogicalRepWorker *
268 : 3446 : logicalrep_worker_find(LogicalRepWorkerType wtype, Oid subid, Oid relid,
269 : : bool only_running)
270 : : {
271 : : int i;
272 : 3446 : LogicalRepWorker *res = NULL;
273 : :
274 : : /* relid must be valid only for table sync workers */
275 : : Assert((wtype == WORKERTYPE_TABLESYNC) == OidIsValid(relid));
276 : : Assert(LWLockHeldByMe(LogicalRepWorkerLock));
277 : :
278 : : /* Search for an attached worker that matches the specified criteria. */
279 [ + + ]: 10461 : for (i = 0; i < max_logical_replication_workers; i++)
280 : : {
281 : 9128 : LogicalRepWorker *w = &LogicalRepCtx->workers[i];
282 : :
283 : : /* Skip parallel apply workers. */
284 [ + + - + ]: 9128 : if (isParallelApplyWorker(w))
285 : 0 : continue;
286 : :
287 [ + + + + : 9128 : if (w->in_use && w->subid == subid && w->relid == relid &&
+ + ]
288 [ + + + + : 2161 : w->type == wtype && (!only_running || w->proc))
+ - ]
289 : : {
290 : 2113 : res = w;
291 : 2113 : break;
292 : : }
293 : : }
294 : :
295 : 3446 : return res;
296 : : }
297 : :
298 : : /*
299 : : * Similar to logicalrep_worker_find(), but returns a list of all workers for
300 : : * the subscription, instead of just one.
301 : : */
302 : : List *
303 : 902 : logicalrep_workers_find(Oid subid, bool only_running, bool acquire_lock)
304 : : {
305 : : int i;
306 : 902 : List *res = NIL;
307 : :
308 [ + + ]: 902 : if (acquire_lock)
309 : 172 : LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
310 : :
311 : : Assert(LWLockHeldByMe(LogicalRepWorkerLock));
312 : :
313 : : /* Search for attached worker for a given subscription id. */
314 [ + + ]: 4640 : for (i = 0; i < max_logical_replication_workers; i++)
315 : : {
316 : 3738 : LogicalRepWorker *w = &LogicalRepCtx->workers[i];
317 : :
318 [ + + + + : 3738 : if (w->in_use && w->subid == subid && (!only_running || w->proc))
+ + + + ]
319 : 550 : res = lappend(res, w);
320 : : }
321 : :
322 [ + + ]: 902 : if (acquire_lock)
323 : 172 : LWLockRelease(LogicalRepWorkerLock);
324 : :
325 : 902 : return res;
326 : : }
327 : :
328 : : /*
329 : : * Start new logical replication background worker, if possible.
330 : : *
331 : : * Returns true on success, false on failure.
332 : : */
333 : : bool
334 : 482 : logicalrep_worker_launch(LogicalRepWorkerType wtype,
335 : : Oid dbid, Oid subid, const char *subname, Oid userid,
336 : : Oid relid, dsm_handle subworker_dsm,
337 : : bool retain_dead_tuples)
338 : : {
339 : : BackgroundWorker bgw;
340 : : BackgroundWorkerHandle *bgw_handle;
341 : : uint16 generation;
342 : : int i;
343 : 482 : int slot = 0;
344 : 482 : LogicalRepWorker *worker = NULL;
345 : : int nsyncworkers;
346 : : int nparallelapplyworkers;
347 : : TimestampTz now;
348 : 482 : bool is_tablesync_worker = (wtype == WORKERTYPE_TABLESYNC);
349 : 482 : bool is_sequencesync_worker = (wtype == WORKERTYPE_SEQUENCESYNC);
350 : 482 : bool is_parallel_apply_worker = (wtype == WORKERTYPE_PARALLEL_APPLY);
351 : :
352 : : /*----------
353 : : * Sanity checks:
354 : : * - must be valid worker type
355 : : * - tablesync workers are only ones to have relid
356 : : * - parallel apply worker is the only kind of subworker
357 : : * - The replication slot used in conflict detection is created when
358 : : * retain_dead_tuples is enabled
359 : : */
360 : : Assert(wtype != WORKERTYPE_UNKNOWN);
361 : : Assert(is_tablesync_worker == OidIsValid(relid));
362 : : Assert(is_parallel_apply_worker == (subworker_dsm != DSM_HANDLE_INVALID));
363 : : Assert(!retain_dead_tuples || MyReplicationSlot);
364 : :
365 [ + + ]: 482 : ereport(DEBUG1,
366 : : (errmsg_internal("starting logical replication worker for subscription \"%s\"",
367 : : subname)));
368 : :
369 : : /* Report this after the initial starting message for consistency. */
370 [ - + ]: 482 : if (max_active_replication_origins == 0)
371 [ # # ]: 0 : ereport(ERROR,
372 : : (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
373 : : errmsg("cannot start logical replication workers when \"max_active_replication_origins\" is 0")));
374 : :
375 : : /*
376 : : * We need to do the modification of the shared memory under lock so that
377 : : * we have consistent view.
378 : : */
379 : 482 : LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
380 : :
381 : 482 : retry:
382 : : /* Find unused worker slot. */
383 [ + - ]: 848 : for (i = 0; i < max_logical_replication_workers; i++)
384 : : {
385 : 848 : LogicalRepWorker *w = &LogicalRepCtx->workers[i];
386 : :
387 [ + + ]: 848 : if (!w->in_use)
388 : : {
389 : 482 : worker = w;
390 : 482 : slot = i;
391 : 482 : break;
392 : : }
393 : : }
394 : :
395 : 482 : nsyncworkers = logicalrep_sync_worker_count(subid);
396 : :
397 : 482 : now = GetCurrentTimestamp();
398 : :
399 : : /*
400 : : * If we didn't find a free slot, try to do garbage collection. The
401 : : * reason we do this is because if some worker failed to start up and its
402 : : * parent has crashed while waiting, the in_use state was never cleared.
403 : : */
404 [ + - - + ]: 482 : if (worker == NULL || nsyncworkers >= max_sync_workers_per_subscription)
405 : : {
406 : 0 : bool did_cleanup = false;
407 : :
408 [ # # ]: 0 : for (i = 0; i < max_logical_replication_workers; i++)
409 : : {
410 : 0 : LogicalRepWorker *w = &LogicalRepCtx->workers[i];
411 : :
412 : : /*
413 : : * If the worker was marked in use but didn't manage to attach in
414 : : * time, clean it up.
415 : : */
416 [ # # # # : 0 : if (w->in_use && !w->proc &&
# # ]
417 : 0 : TimestampDifferenceExceeds(w->launch_time, now,
418 : : wal_receiver_timeout))
419 : : {
420 [ # # ]: 0 : elog(WARNING,
421 : : "logical replication worker for subscription %u took too long to start; canceled",
422 : : w->subid);
423 : :
424 : 0 : logicalrep_worker_cleanup(w);
425 : 0 : did_cleanup = true;
426 : : }
427 : : }
428 : :
429 [ # # ]: 0 : if (did_cleanup)
430 : 0 : goto retry;
431 : : }
432 : :
433 : : /*
434 : : * We don't allow to invoke more sync workers once we have reached the
435 : : * sync worker limit per subscription. So, just return silently as we
436 : : * might get here because of an otherwise harmless race condition.
437 : : */
438 [ + + + + ]: 482 : if ((is_tablesync_worker || is_sequencesync_worker) &&
439 [ - + ]: 234 : nsyncworkers >= max_sync_workers_per_subscription)
440 : : {
441 : 0 : LWLockRelease(LogicalRepWorkerLock);
442 : 0 : return false;
443 : : }
444 : :
445 : 482 : nparallelapplyworkers = logicalrep_pa_worker_count(subid);
446 : :
447 : : /*
448 : : * Return false if the number of parallel apply workers reached the limit
449 : : * per subscription.
450 : : */
451 [ + + ]: 482 : if (is_parallel_apply_worker &&
452 [ - + ]: 12 : nparallelapplyworkers >= max_parallel_apply_workers_per_subscription)
453 : : {
454 : 0 : LWLockRelease(LogicalRepWorkerLock);
455 : 0 : return false;
456 : : }
457 : :
458 : : /*
459 : : * However if there are no more free worker slots, inform user about it
460 : : * before exiting.
461 : : */
462 [ - + ]: 482 : if (worker == NULL)
463 : : {
464 : 0 : LWLockRelease(LogicalRepWorkerLock);
465 [ # # ]: 0 : ereport(WARNING,
466 : : (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
467 : : errmsg("out of logical replication worker slots"),
468 : : errhint("You might need to increase \"%s\".", "max_logical_replication_workers")));
469 : 0 : return false;
470 : : }
471 : :
472 : : /* Prepare the worker slot. */
473 : 482 : worker->type = wtype;
474 : 482 : worker->launch_time = now;
475 : 482 : worker->in_use = true;
476 : 482 : worker->generation++;
477 : 482 : worker->proc = NULL;
478 : 482 : worker->dbid = dbid;
479 : 482 : worker->userid = userid;
480 : 482 : worker->subid = subid;
481 : 482 : worker->relid = relid;
482 : 482 : worker->relstate = SUBREL_STATE_UNKNOWN;
483 : 482 : worker->relstate_lsn = InvalidXLogRecPtr;
484 : 482 : worker->stream_fileset = NULL;
485 [ + + ]: 482 : worker->leader_pid = is_parallel_apply_worker ? MyProcPid : InvalidPid;
486 : 482 : worker->parallel_apply = is_parallel_apply_worker;
487 : 482 : worker->oldest_nonremovable_xid = retain_dead_tuples
488 : 2 : ? MyReplicationSlot->data.xmin
489 [ + + ]: 482 : : InvalidTransactionId;
490 : 482 : worker->last_lsn = InvalidXLogRecPtr;
491 : 482 : TIMESTAMP_NOBEGIN(worker->last_send_time);
492 : 482 : TIMESTAMP_NOBEGIN(worker->last_recv_time);
493 : 482 : worker->reply_lsn = InvalidXLogRecPtr;
494 : 482 : TIMESTAMP_NOBEGIN(worker->reply_time);
495 : 482 : worker->last_seqsync_start_time = 0;
496 : :
497 : : /* Before releasing lock, remember generation for future identification. */
498 : 482 : generation = worker->generation;
499 : :
500 : 482 : LWLockRelease(LogicalRepWorkerLock);
501 : :
502 : : /* Register the new dynamic worker. */
503 : 482 : memset(&bgw, 0, sizeof(bgw));
504 : 482 : bgw.bgw_flags = BGWORKER_SHMEM_ACCESS |
505 : : BGWORKER_BACKEND_DATABASE_CONNECTION;
506 : 482 : bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
507 : 482 : snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres");
508 : :
509 [ + + + + : 482 : switch (worker->type)
- - ]
510 : : {
511 : 236 : case WORKERTYPE_APPLY:
512 : 236 : snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
513 : 236 : snprintf(bgw.bgw_name, BGW_MAXLEN,
514 : : "logical replication apply worker for subscription %u",
515 : : subid);
516 : 236 : snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication apply worker");
517 : 236 : break;
518 : :
519 : 12 : case WORKERTYPE_PARALLEL_APPLY:
520 : 12 : snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain");
521 : 12 : snprintf(bgw.bgw_name, BGW_MAXLEN,
522 : : "logical replication parallel apply worker for subscription %u",
523 : : subid);
524 : 12 : snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
525 : :
526 : 12 : memcpy(bgw.bgw_extra, &subworker_dsm, sizeof(dsm_handle));
527 : 12 : break;
528 : :
529 : 14 : case WORKERTYPE_SEQUENCESYNC:
530 : 14 : snprintf(bgw.bgw_function_name, BGW_MAXLEN, "SequenceSyncWorkerMain");
531 : 14 : snprintf(bgw.bgw_name, BGW_MAXLEN,
532 : : "logical replication sequencesync worker for subscription %u",
533 : : subid);
534 : 14 : snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication sequencesync worker");
535 : 14 : break;
536 : :
537 : 220 : case WORKERTYPE_TABLESYNC:
538 : 220 : snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TableSyncWorkerMain");
539 : 220 : snprintf(bgw.bgw_name, BGW_MAXLEN,
540 : : "logical replication tablesync worker for subscription %u sync %u",
541 : : subid,
542 : : relid);
543 : 220 : snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication tablesync worker");
544 : 220 : break;
545 : :
546 : 0 : case WORKERTYPE_UNKNOWN:
547 : : /* Should never happen. */
548 [ # # ]: 0 : elog(ERROR, "unknown worker type");
549 : : }
550 : :
551 : 482 : bgw.bgw_restart_time = BGW_NEVER_RESTART;
552 : 482 : bgw.bgw_notify_pid = MyProcPid;
553 : 482 : bgw.bgw_main_arg = Int32GetDatum(slot);
554 : :
555 [ + + ]: 482 : if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
556 : : {
557 : : /* Failed to start worker, so clean up the worker slot. */
558 : 8 : LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
559 : : Assert(generation == worker->generation);
560 : 8 : logicalrep_worker_cleanup(worker);
561 : 8 : LWLockRelease(LogicalRepWorkerLock);
562 : :
563 [ + - ]: 8 : ereport(WARNING,
564 : : (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
565 : : errmsg("out of background worker slots"),
566 : : errhint("You might need to increase \"%s\".", "max_worker_processes")));
567 : 8 : return false;
568 : : }
569 : :
570 : : /* Now wait until it attaches. */
571 : 474 : return WaitForReplicationWorkerAttach(worker, generation, bgw_handle);
572 : : }
573 : :
574 : : /*
575 : : * Internal function to stop the worker and wait until it detaches from the
576 : : * slot.
577 : : */
578 : : static void
579 : 88 : logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo)
580 : : {
581 : : uint16 generation;
582 : :
583 : : Assert(LWLockHeldByMeInMode(LogicalRepWorkerLock, LW_SHARED));
584 : :
585 : : /*
586 : : * Remember which generation was our worker so we can check if what we see
587 : : * is still the same one.
588 : : */
589 : 88 : generation = worker->generation;
590 : :
591 : : /*
592 : : * If we found a worker but it does not have proc set then it is still
593 : : * starting up; wait for it to finish starting and then kill it.
594 : : */
595 [ + - + + ]: 88 : while (worker->in_use && !worker->proc)
596 : : {
597 : : int rc;
598 : :
599 : 2 : LWLockRelease(LogicalRepWorkerLock);
600 : :
601 : : /* Wait a bit --- we don't expect to have to wait long. */
602 : 2 : rc = WaitLatch(MyLatch,
603 : : WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
604 : : 10L, WAIT_EVENT_BGWORKER_STARTUP);
605 : :
606 [ - + ]: 2 : if (rc & WL_LATCH_SET)
607 : : {
608 : 0 : ResetLatch(MyLatch);
609 [ # # ]: 0 : CHECK_FOR_INTERRUPTS();
610 : : }
611 : :
612 : : /* Recheck worker status. */
613 : 2 : LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
614 : :
615 : : /*
616 : : * Check whether the worker slot is no longer used, which would mean
617 : : * that the worker has exited, or whether the worker generation is
618 : : * different, meaning that a different worker has taken the slot.
619 : : */
620 [ + - - + ]: 2 : if (!worker->in_use || worker->generation != generation)
621 : 0 : return;
622 : :
623 : : /* Worker has assigned proc, so it has started. */
624 [ + - ]: 2 : if (worker->proc)
625 : 2 : break;
626 : : }
627 : :
628 : : /* Now terminate the worker ... */
629 : 88 : kill(worker->proc->pid, signo);
630 : :
631 : : /* ... and wait for it to die. */
632 : : for (;;)
633 : 114 : {
634 : : int rc;
635 : :
636 : : /* is it gone? */
637 [ + + + + ]: 202 : if (!worker->proc || worker->generation != generation)
638 : : break;
639 : :
640 : 114 : LWLockRelease(LogicalRepWorkerLock);
641 : :
642 : : /* Wait a bit --- we don't expect to have to wait long. */
643 : 114 : rc = WaitLatch(MyLatch,
644 : : WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
645 : : 10L, WAIT_EVENT_BGWORKER_SHUTDOWN);
646 : :
647 [ + + ]: 114 : if (rc & WL_LATCH_SET)
648 : : {
649 : 29 : ResetLatch(MyLatch);
650 [ + + ]: 29 : CHECK_FOR_INTERRUPTS();
651 : : }
652 : :
653 : 114 : LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
654 : : }
655 : : }
656 : :
657 : : /*
658 : : * Stop the logical replication worker that matches the specified worker type,
659 : : * subscription id, and relation id.
660 : : */
661 : : void
662 : 101 : logicalrep_worker_stop(LogicalRepWorkerType wtype, Oid subid, Oid relid)
663 : : {
664 : : LogicalRepWorker *worker;
665 : :
666 : : /* relid must be valid only for table sync workers */
667 : : Assert((wtype == WORKERTYPE_TABLESYNC) == OidIsValid(relid));
668 : :
669 : 101 : LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
670 : :
671 : 101 : worker = logicalrep_worker_find(wtype, subid, relid, false);
672 : :
673 [ + + ]: 101 : if (worker)
674 : : {
675 : : Assert(!isParallelApplyWorker(worker));
676 : 79 : logicalrep_worker_stop_internal(worker, SIGTERM);
677 : : }
678 : :
679 : 101 : LWLockRelease(LogicalRepWorkerLock);
680 : 101 : }
681 : :
682 : : /*
683 : : * Stop the given logical replication parallel apply worker.
684 : : *
685 : : * Node that the function sends SIGUSR2 instead of SIGTERM to the parallel apply
686 : : * worker so that the worker exits cleanly.
687 : : */
688 : : void
689 : 5 : logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo)
690 : : {
691 : : int slot_no;
692 : : uint16 generation;
693 : : LogicalRepWorker *worker;
694 : :
695 : 5 : SpinLockAcquire(&winfo->shared->mutex);
696 : 5 : generation = winfo->shared->logicalrep_worker_generation;
697 : 5 : slot_no = winfo->shared->logicalrep_worker_slot_no;
698 : 5 : SpinLockRelease(&winfo->shared->mutex);
699 : :
700 : : Assert(slot_no >= 0 && slot_no < max_logical_replication_workers);
701 : :
702 : : /*
703 : : * Detach from the error_mq_handle for the parallel apply worker before
704 : : * stopping it. This prevents the leader apply worker from trying to
705 : : * receive the message from the error queue that might already be detached
706 : : * by the parallel apply worker.
707 : : */
708 [ + - ]: 5 : if (winfo->error_mq_handle)
709 : : {
710 : 5 : shm_mq_detach(winfo->error_mq_handle);
711 : 5 : winfo->error_mq_handle = NULL;
712 : : }
713 : :
714 : 5 : LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
715 : :
716 : 5 : worker = &LogicalRepCtx->workers[slot_no];
717 : : Assert(isParallelApplyWorker(worker));
718 : :
719 : : /*
720 : : * Only stop the worker if the generation matches and the worker is alive.
721 : : */
722 [ + - + - ]: 5 : if (worker->generation == generation && worker->proc)
723 : 5 : logicalrep_worker_stop_internal(worker, SIGUSR2);
724 : :
725 : 5 : LWLockRelease(LogicalRepWorkerLock);
726 : 5 : }
727 : :
728 : : /*
729 : : * Wake up (using latch) any logical replication worker that matches the
730 : : * specified worker type, subscription id, and relation id.
731 : : */
732 : : void
733 : 229 : logicalrep_worker_wakeup(LogicalRepWorkerType wtype, Oid subid, Oid relid)
734 : : {
735 : : LogicalRepWorker *worker;
736 : :
737 : : /* relid must be valid only for table sync workers */
738 : : Assert((wtype == WORKERTYPE_TABLESYNC) == OidIsValid(relid));
739 : :
740 : 229 : LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
741 : :
742 : 229 : worker = logicalrep_worker_find(wtype, subid, relid, true);
743 : :
744 [ + - ]: 229 : if (worker)
745 : 229 : logicalrep_worker_wakeup_ptr(worker);
746 : :
747 : 229 : LWLockRelease(LogicalRepWorkerLock);
748 : 229 : }
749 : :
750 : : /*
751 : : * Wake up (using latch) the specified logical replication worker.
752 : : *
753 : : * Caller must hold lock, else worker->proc could change under us.
754 : : */
755 : : void
756 : 705 : logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker)
757 : : {
758 : : Assert(LWLockHeldByMe(LogicalRepWorkerLock));
759 : :
760 : 705 : SetLatch(&worker->proc->procLatch);
761 : 705 : }
762 : :
763 : : /*
764 : : * Attach to a slot.
765 : : */
766 : : void
767 : 628 : logicalrep_worker_attach(int slot)
768 : : {
769 : : /* Block concurrent access. */
770 : 628 : LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
771 : :
772 : : Assert(slot >= 0 && slot < max_logical_replication_workers);
773 : 628 : MyLogicalRepWorker = &LogicalRepCtx->workers[slot];
774 : :
775 [ - + ]: 628 : if (!MyLogicalRepWorker->in_use)
776 : : {
777 : 0 : LWLockRelease(LogicalRepWorkerLock);
778 [ # # ]: 0 : ereport(ERROR,
779 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
780 : : errmsg("logical replication worker slot %d is empty, cannot attach",
781 : : slot)));
782 : : }
783 : :
784 [ - + ]: 628 : if (MyLogicalRepWorker->proc)
785 : : {
786 : 0 : LWLockRelease(LogicalRepWorkerLock);
787 [ # # ]: 0 : ereport(ERROR,
788 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
789 : : errmsg("logical replication worker slot %d is already used by "
790 : : "another worker, cannot attach", slot)));
791 : : }
792 : :
793 : 628 : MyLogicalRepWorker->proc = MyProc;
794 : 628 : before_shmem_exit(logicalrep_worker_onexit, (Datum) 0);
795 : :
796 : 628 : LWLockRelease(LogicalRepWorkerLock);
797 : 628 : }
798 : :
799 : : /*
800 : : * Stop the parallel apply workers if any, and detach the leader apply worker
801 : : * (cleans up the worker info).
802 : : */
803 : : static void
804 : 628 : logicalrep_worker_detach(void)
805 : : {
806 : : /* Stop the parallel apply workers. */
807 [ + + ]: 628 : if (am_leader_apply_worker())
808 : : {
809 : : List *workers;
810 : : ListCell *lc;
811 : :
812 : : /*
813 : : * Detach from the error_mq_handle for all parallel apply workers
814 : : * before terminating them. This prevents the leader apply worker from
815 : : * receiving the worker termination message and sending it to logs
816 : : * when the same is already done by the parallel worker.
817 : : */
818 : 388 : pa_detach_all_error_mq();
819 : :
820 : 388 : LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
821 : :
822 : 388 : workers = logicalrep_workers_find(MyLogicalRepWorker->subid, true, false);
823 [ + - + + : 780 : foreach(lc, workers)
+ + ]
824 : : {
825 : 392 : LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
826 : :
827 [ + - + + ]: 392 : if (isParallelApplyWorker(w))
828 : 4 : logicalrep_worker_stop_internal(w, SIGTERM);
829 : : }
830 : :
831 : 388 : LWLockRelease(LogicalRepWorkerLock);
832 : :
833 : 388 : list_free(workers);
834 : : }
835 : :
836 : : /* Block concurrent access. */
837 : 628 : LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
838 : :
839 : 628 : logicalrep_worker_cleanup(MyLogicalRepWorker);
840 : :
841 : 628 : LWLockRelease(LogicalRepWorkerLock);
842 : 628 : }
843 : :
844 : : /*
845 : : * Clean up worker info.
846 : : */
847 : : static void
848 : 636 : logicalrep_worker_cleanup(LogicalRepWorker *worker)
849 : : {
850 : : Assert(LWLockHeldByMeInMode(LogicalRepWorkerLock, LW_EXCLUSIVE));
851 : :
852 : 636 : worker->type = WORKERTYPE_UNKNOWN;
853 : 636 : worker->in_use = false;
854 : 636 : worker->proc = NULL;
855 : 636 : worker->dbid = InvalidOid;
856 : 636 : worker->userid = InvalidOid;
857 : 636 : worker->subid = InvalidOid;
858 : 636 : worker->relid = InvalidOid;
859 : 636 : worker->leader_pid = InvalidPid;
860 : 636 : worker->parallel_apply = false;
861 : 636 : }
862 : :
863 : : /*
864 : : * Cleanup function for logical replication launcher.
865 : : *
866 : : * Called on logical replication launcher exit.
867 : : */
868 : : static void
869 : 520 : logicalrep_launcher_onexit(int code, Datum arg)
870 : : {
871 : 520 : LogicalRepCtx->launcher_pid = 0;
872 : 520 : }
873 : :
874 : : /*
875 : : * Reset the last_seqsync_start_time of the sequencesync worker in the
876 : : * subscription's apply worker.
877 : : *
878 : : * Note that this value is not stored in the sequencesync worker, because that
879 : : * has finished already and is about to exit.
880 : : */
881 : : void
882 : 7 : logicalrep_reset_seqsync_start_time(void)
883 : : {
884 : : LogicalRepWorker *worker;
885 : :
886 : : /*
887 : : * The apply worker can't access last_seqsync_start_time concurrently, so
888 : : * it is okay to use SHARED lock here. See ProcessSequencesForSync().
889 : : */
890 : 7 : LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
891 : :
892 : 7 : worker = logicalrep_worker_find(WORKERTYPE_APPLY,
893 : 7 : MyLogicalRepWorker->subid, InvalidOid,
894 : : true);
895 [ + - ]: 7 : if (worker)
896 : 7 : worker->last_seqsync_start_time = 0;
897 : :
898 : 7 : LWLockRelease(LogicalRepWorkerLock);
899 : 7 : }
900 : :
901 : : /*
902 : : * Cleanup function.
903 : : *
904 : : * Called on logical replication worker exit.
905 : : */
906 : : static void
907 : 628 : logicalrep_worker_onexit(int code, Datum arg)
908 : : {
909 : : /* Disconnect gracefully from the remote side. */
910 [ + + ]: 628 : if (LogRepWorkerWalRcvConn)
911 : 500 : walrcv_disconnect(LogRepWorkerWalRcvConn);
912 : :
913 : 628 : logicalrep_worker_detach();
914 : :
915 : : /* Cleanup fileset used for streaming transactions. */
916 [ + + ]: 628 : if (MyLogicalRepWorker->stream_fileset != NULL)
917 : 14 : FileSetDeleteAll(MyLogicalRepWorker->stream_fileset);
918 : :
919 : : /*
920 : : * Session level locks may be acquired outside of a transaction in
921 : : * parallel apply mode and will not be released when the worker
922 : : * terminates, so manually release all locks before the worker exits.
923 : : *
924 : : * The locks will be acquired once the worker is initialized.
925 : : */
926 [ + + ]: 628 : if (!InitializingApplyWorker)
927 : 555 : LockReleaseAll(DEFAULT_LOCKMETHOD, true);
928 : :
929 : 628 : ApplyLauncherWakeup();
930 : 628 : }
931 : :
932 : : /*
933 : : * Count the number of registered (not necessarily running) sync workers
934 : : * for a subscription.
935 : : */
936 : : int
937 : 1404 : logicalrep_sync_worker_count(Oid subid)
938 : : {
939 : : int i;
940 : 1404 : int res = 0;
941 : :
942 : : Assert(LWLockHeldByMe(LogicalRepWorkerLock));
943 : :
944 : : /* Search for attached worker for a given subscription id. */
945 [ + + ]: 7230 : for (i = 0; i < max_logical_replication_workers; i++)
946 : : {
947 : 5826 : LogicalRepWorker *w = &LogicalRepCtx->workers[i];
948 : :
949 [ + + + - : 5826 : if (w->subid == subid && (isTableSyncWorker(w) || isSequenceSyncWorker(w)))
+ + + - -
+ ]
950 : 1387 : res++;
951 : : }
952 : :
953 : 1404 : return res;
954 : : }
955 : :
956 : : /*
957 : : * Count the number of registered (but not necessarily running) parallel apply
958 : : * workers for a subscription.
959 : : */
960 : : static int
961 : 482 : logicalrep_pa_worker_count(Oid subid)
962 : : {
963 : : int i;
964 : 482 : int res = 0;
965 : :
966 : : Assert(LWLockHeldByMe(LogicalRepWorkerLock));
967 : :
968 : : /*
969 : : * Scan all attached parallel apply workers, only counting those which
970 : : * have the given subscription id.
971 : : */
972 [ + + ]: 2536 : for (i = 0; i < max_logical_replication_workers; i++)
973 : : {
974 : 2054 : LogicalRepWorker *w = &LogicalRepCtx->workers[i];
975 : :
976 [ + + + + : 2054 : if (isParallelApplyWorker(w) && w->subid == subid)
+ - ]
977 : 2 : res++;
978 : : }
979 : :
980 : 482 : return res;
981 : : }
982 : :
983 : : /*
984 : : * ApplyLauncherShmemRequest
985 : : * Register shared memory space needed for replication launcher
986 : : */
987 : : static void
988 : 1253 : ApplyLauncherShmemRequest(void *arg)
989 : : {
990 : : Size size;
991 : :
992 : : /*
993 : : * Need the fixed struct and the array of LogicalRepWorker.
994 : : */
995 : 1253 : size = sizeof(LogicalRepCtxStruct);
996 : 1253 : size = MAXALIGN(size);
997 : 1253 : size = add_size(size, mul_size(max_logical_replication_workers,
998 : : sizeof(LogicalRepWorker)));
999 : 1253 : ShmemRequestStruct(.name = "Logical Replication Launcher Data",
1000 : : .size = size,
1001 : : .ptr = (void **) &LogicalRepCtx,
1002 : : );
1003 : 1253 : }
1004 : :
1005 : : /*
1006 : : * ApplyLauncherRegister
1007 : : * Register a background worker running the logical replication launcher.
1008 : : */
1009 : : void
1010 : 1013 : ApplyLauncherRegister(void)
1011 : : {
1012 : : BackgroundWorker bgw;
1013 : :
1014 : : /*
1015 : : * The logical replication launcher is disabled during binary upgrades, to
1016 : : * prevent logical replication workers from running on the source cluster.
1017 : : * That could cause replication origins to move forward after having been
1018 : : * copied to the target cluster, potentially creating conflicts with the
1019 : : * copied data files.
1020 : : */
1021 [ + + + + ]: 1013 : if (max_logical_replication_workers == 0 || IsBinaryUpgrade)
1022 : 61 : return;
1023 : :
1024 : 952 : memset(&bgw, 0, sizeof(bgw));
1025 : 952 : bgw.bgw_flags = BGWORKER_SHMEM_ACCESS |
1026 : : BGWORKER_BACKEND_DATABASE_CONNECTION;
1027 : 952 : bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
1028 : 952 : snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres");
1029 : 952 : snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyLauncherMain");
1030 : 952 : snprintf(bgw.bgw_name, BGW_MAXLEN,
1031 : : "logical replication launcher");
1032 : 952 : snprintf(bgw.bgw_type, BGW_MAXLEN,
1033 : : "logical replication launcher");
1034 : 952 : bgw.bgw_restart_time = 5;
1035 : 952 : bgw.bgw_notify_pid = 0;
1036 : 952 : bgw.bgw_main_arg = (Datum) 0;
1037 : :
1038 : 952 : RegisterBackgroundWorker(&bgw);
1039 : : }
1040 : :
1041 : : /*
1042 : : * ApplyLauncherShmemInit
1043 : : * Initialize replication launcher shared memory
1044 : : */
1045 : : static void
1046 : 1250 : ApplyLauncherShmemInit(void *arg)
1047 : : {
1048 : : int slot;
1049 : :
1050 : 1250 : LogicalRepCtx->last_start_dsa = DSA_HANDLE_INVALID;
1051 : 1250 : LogicalRepCtx->last_start_dsh = DSHASH_HANDLE_INVALID;
1052 : :
1053 : : /* Initialize memory and spin locks for each worker slot. */
1054 [ + + ]: 6213 : for (slot = 0; slot < max_logical_replication_workers; slot++)
1055 : : {
1056 : 4963 : LogicalRepWorker *worker = &LogicalRepCtx->workers[slot];
1057 : :
1058 : 4963 : memset(worker, 0, sizeof(LogicalRepWorker));
1059 : 4963 : SpinLockInit(&worker->relmutex);
1060 : : }
1061 : 1250 : }
1062 : :
1063 : : /*
1064 : : * Initialize or attach to the dynamic shared hash table that stores the
1065 : : * last-start times, if not already done.
1066 : : * This must be called before accessing the table.
1067 : : */
1068 : : static void
1069 : 915 : logicalrep_launcher_attach_dshmem(void)
1070 : : {
1071 : : MemoryContext oldcontext;
1072 : :
1073 : : /* Quick exit if we already did this. */
1074 [ + + ]: 915 : if (LogicalRepCtx->last_start_dsh != DSHASH_HANDLE_INVALID &&
1075 [ + + ]: 855 : last_start_times != NULL)
1076 : 645 : return;
1077 : :
1078 : : /* Otherwise, use a lock to ensure only one process creates the table. */
1079 : 270 : LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
1080 : :
1081 : : /* Be sure any local memory allocated by DSA routines is persistent. */
1082 : 270 : oldcontext = MemoryContextSwitchTo(TopMemoryContext);
1083 : :
1084 [ + + ]: 270 : if (LogicalRepCtx->last_start_dsh == DSHASH_HANDLE_INVALID)
1085 : : {
1086 : : /* Initialize dynamic shared hash table for last-start times. */
1087 : 60 : last_start_times_dsa = dsa_create(LWTRANCHE_LAUNCHER_DSA);
1088 : 60 : dsa_pin(last_start_times_dsa);
1089 : 60 : dsa_pin_mapping(last_start_times_dsa);
1090 : 60 : last_start_times = dshash_create(last_start_times_dsa, &dsh_params, NULL);
1091 : :
1092 : : /* Store handles in shared memory for other backends to use. */
1093 : 60 : LogicalRepCtx->last_start_dsa = dsa_get_handle(last_start_times_dsa);
1094 : 60 : LogicalRepCtx->last_start_dsh = dshash_get_hash_table_handle(last_start_times);
1095 : : }
1096 [ + - ]: 210 : else if (!last_start_times)
1097 : : {
1098 : : /* Attach to existing dynamic shared hash table. */
1099 : 210 : last_start_times_dsa = dsa_attach(LogicalRepCtx->last_start_dsa);
1100 : 210 : dsa_pin_mapping(last_start_times_dsa);
1101 : 210 : last_start_times = dshash_attach(last_start_times_dsa, &dsh_params,
1102 : 210 : LogicalRepCtx->last_start_dsh, NULL);
1103 : : }
1104 : :
1105 : 270 : MemoryContextSwitchTo(oldcontext);
1106 : 270 : LWLockRelease(LogicalRepWorkerLock);
1107 : : }
1108 : :
1109 : : /*
1110 : : * Set the last-start time for the subscription.
1111 : : */
1112 : : static void
1113 : 236 : ApplyLauncherSetWorkerStartTime(Oid subid, TimestampTz start_time)
1114 : : {
1115 : : LauncherLastStartTimesEntry *entry;
1116 : : bool found;
1117 : :
1118 : 236 : logicalrep_launcher_attach_dshmem();
1119 : :
1120 : 236 : entry = dshash_find_or_insert(last_start_times, &subid, &found);
1121 : 236 : entry->last_start_time = start_time;
1122 : 236 : dshash_release_lock(last_start_times, entry);
1123 : 236 : }
1124 : :
1125 : : /*
1126 : : * Return the last-start time for the subscription, or 0 if there isn't one.
1127 : : */
1128 : : static TimestampTz
1129 : 385 : ApplyLauncherGetWorkerStartTime(Oid subid)
1130 : : {
1131 : : LauncherLastStartTimesEntry *entry;
1132 : : TimestampTz ret;
1133 : :
1134 : 385 : logicalrep_launcher_attach_dshmem();
1135 : :
1136 : 385 : entry = dshash_find(last_start_times, &subid, false);
1137 [ + + ]: 385 : if (entry == NULL)
1138 : 136 : return 0;
1139 : :
1140 : 249 : ret = entry->last_start_time;
1141 : 249 : dshash_release_lock(last_start_times, entry);
1142 : :
1143 : 249 : return ret;
1144 : : }
1145 : :
1146 : : /*
1147 : : * Remove the last-start-time entry for the subscription, if one exists.
1148 : : *
1149 : : * This has two use-cases: to remove the entry related to a subscription
1150 : : * that's been deleted or disabled (just to avoid leaking shared memory),
1151 : : * and to allow immediate restart of an apply worker that has exited
1152 : : * due to subscription parameter changes.
1153 : : */
1154 : : void
1155 : 294 : ApplyLauncherForgetWorkerStartTime(Oid subid)
1156 : : {
1157 : 294 : logicalrep_launcher_attach_dshmem();
1158 : :
1159 : 294 : (void) dshash_delete_key(last_start_times, &subid);
1160 : 294 : }
1161 : :
1162 : : /*
1163 : : * Wakeup the launcher on commit if requested.
1164 : : */
1165 : : void
1166 : 650749 : AtEOXact_ApplyLauncher(bool isCommit)
1167 : : {
1168 [ + + ]: 650749 : if (isCommit)
1169 : : {
1170 [ + + ]: 614497 : if (on_commit_launcher_wakeup)
1171 : 162 : ApplyLauncherWakeup();
1172 : : }
1173 : :
1174 : 650749 : on_commit_launcher_wakeup = false;
1175 : 650749 : }
1176 : :
1177 : : /*
1178 : : * Request wakeup of the launcher on commit of the transaction.
1179 : : *
1180 : : * This is used to send launcher signal to stop sleeping and process the
1181 : : * subscriptions when current transaction commits. Should be used when new
1182 : : * tuple was added to the pg_subscription catalog.
1183 : : */
1184 : : void
1185 : 163 : ApplyLauncherWakeupAtCommit(void)
1186 : : {
1187 [ + + ]: 163 : if (!on_commit_launcher_wakeup)
1188 : 162 : on_commit_launcher_wakeup = true;
1189 : 163 : }
1190 : :
1191 : : /*
1192 : : * Wakeup the launcher immediately.
1193 : : */
1194 : : void
1195 : 842 : ApplyLauncherWakeup(void)
1196 : : {
1197 [ + + ]: 842 : if (LogicalRepCtx->launcher_pid != 0)
1198 : 823 : kill(LogicalRepCtx->launcher_pid, SIGUSR1);
1199 : 842 : }
1200 : :
1201 : : /*
1202 : : * Main loop for the apply launcher process.
1203 : : */
1204 : : void
1205 : 520 : ApplyLauncherMain(Datum main_arg)
1206 : : {
1207 [ + + ]: 520 : ereport(DEBUG1,
1208 : : (errmsg_internal("logical replication launcher started")));
1209 : :
1210 : 520 : before_shmem_exit(logicalrep_launcher_onexit, (Datum) 0);
1211 : :
1212 : : Assert(LogicalRepCtx->launcher_pid == 0);
1213 : 520 : LogicalRepCtx->launcher_pid = MyProcPid;
1214 : :
1215 : : /* Establish signal handlers. */
1216 : 520 : pqsignal(SIGHUP, SignalHandlerForConfigReload);
1217 : 520 : BackgroundWorkerUnblockSignals();
1218 : :
1219 : : /*
1220 : : * Establish connection to nailed catalogs (we only ever access
1221 : : * pg_subscription).
1222 : : */
1223 : 520 : BackgroundWorkerInitializeConnection(NULL, NULL, 0);
1224 : :
1225 : : /*
1226 : : * Acquire the conflict detection slot at startup to ensure it can be
1227 : : * dropped if no longer needed after a restart.
1228 : : */
1229 : 520 : acquire_conflict_slot_if_exists();
1230 : :
1231 : : /* Enter main loop */
1232 : : for (;;)
1233 : 2815 : {
1234 : : int rc;
1235 : : List *sublist;
1236 : : ListCell *lc;
1237 : : MemoryContext subctx;
1238 : : MemoryContext oldctx;
1239 : 3335 : long wait_time = DEFAULT_NAPTIME_PER_CYCLE;
1240 : 3335 : bool can_update_xmin = true;
1241 : 3335 : bool retain_dead_tuples = false;
1242 : 3335 : TransactionId xmin = InvalidTransactionId;
1243 : :
1244 [ - + ]: 3335 : CHECK_FOR_INTERRUPTS();
1245 : :
1246 : : /* Use temporary context to avoid leaking memory across cycles. */
1247 : 3335 : subctx = AllocSetContextCreate(TopMemoryContext,
1248 : : "Logical Replication Launcher sublist",
1249 : : ALLOCSET_DEFAULT_SIZES);
1250 : 3335 : oldctx = MemoryContextSwitchTo(subctx);
1251 : :
1252 : : /*
1253 : : * Start any missing workers for enabled subscriptions.
1254 : : *
1255 : : * Also, during the iteration through all subscriptions, we compute
1256 : : * the minimum XID required to protect deleted tuples for conflict
1257 : : * detection if one of the subscription enables retain_dead_tuples
1258 : : * option.
1259 : : */
1260 : 3335 : sublist = get_subscription_list();
1261 [ + + + + : 4393 : foreach(lc, sublist)
+ + ]
1262 : : {
1263 : 1063 : Subscription *sub = (Subscription *) lfirst(lc);
1264 : : LogicalRepWorker *w;
1265 : : TimestampTz last_start;
1266 : : TimestampTz now;
1267 : : long elapsed;
1268 : :
1269 [ + + ]: 1063 : if (sub->retaindeadtuples)
1270 : : {
1271 : 98 : retain_dead_tuples = true;
1272 : :
1273 : : /*
1274 : : * Create a replication slot to retain information necessary
1275 : : * for conflict detection such as dead tuples, commit
1276 : : * timestamps, and origins.
1277 : : *
1278 : : * The slot is created before starting the apply worker to
1279 : : * prevent it from unnecessarily maintaining its
1280 : : * oldest_nonremovable_xid.
1281 : : *
1282 : : * The slot is created even for a disabled subscription to
1283 : : * ensure that conflict-related information is available when
1284 : : * applying remote changes that occurred before the
1285 : : * subscription was enabled.
1286 : : */
1287 : 98 : CreateConflictDetectionSlot();
1288 : :
1289 [ + - ]: 98 : if (sub->retentionactive)
1290 : : {
1291 : : /*
1292 : : * Can't advance xmin of the slot unless all the
1293 : : * subscriptions actively retaining dead tuples are
1294 : : * enabled. This is required to ensure that we don't
1295 : : * advance the xmin of CONFLICT_DETECTION_SLOT if one of
1296 : : * the subscriptions is not enabled. Otherwise, we won't
1297 : : * be able to detect conflicts reliably for such a
1298 : : * subscription even though it has set the
1299 : : * retain_dead_tuples option.
1300 : : */
1301 : 98 : can_update_xmin &= sub->enabled;
1302 : :
1303 : : /*
1304 : : * Initialize the slot once the subscription activates
1305 : : * retention.
1306 : : */
1307 [ - + ]: 98 : if (!TransactionIdIsValid(MyReplicationSlot->data.xmin))
1308 : 0 : init_conflict_slot_xmin();
1309 : : }
1310 : : }
1311 : :
1312 [ + + ]: 1063 : if (!sub->enabled)
1313 : 61 : continue;
1314 : :
1315 : 1002 : LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
1316 : 1002 : w = logicalrep_worker_find(WORKERTYPE_APPLY, sub->oid, InvalidOid,
1317 : : false);
1318 : :
1319 [ + + ]: 1002 : if (w != NULL)
1320 : : {
1321 : : /*
1322 : : * Compute the minimum xmin required to protect dead tuples
1323 : : * required for conflict detection among all running apply
1324 : : * workers. This computation is performed while holding
1325 : : * LogicalRepWorkerLock to prevent accessing invalid worker
1326 : : * data, in scenarios where a worker might exit and reset its
1327 : : * state concurrently.
1328 : : */
1329 [ + + ]: 617 : if (sub->retaindeadtuples &&
1330 [ + - + - ]: 94 : sub->retentionactive &&
1331 : : can_update_xmin)
1332 : 94 : compute_min_nonremovable_xid(w, &xmin);
1333 : :
1334 : 617 : LWLockRelease(LogicalRepWorkerLock);
1335 : :
1336 : : /* worker is running already */
1337 : 617 : continue;
1338 : : }
1339 : :
1340 : 385 : LWLockRelease(LogicalRepWorkerLock);
1341 : :
1342 : : /*
1343 : : * Can't advance xmin of the slot unless all the workers
1344 : : * corresponding to subscriptions actively retaining dead tuples
1345 : : * are running, disabling the further computation of the minimum
1346 : : * nonremovable xid.
1347 : : */
1348 [ + + + - ]: 385 : if (sub->retaindeadtuples && sub->retentionactive)
1349 : 2 : can_update_xmin = false;
1350 : :
1351 : : /*
1352 : : * If the worker is eligible to start now, launch it. Otherwise,
1353 : : * adjust wait_time so that we'll wake up as soon as it can be
1354 : : * started.
1355 : : *
1356 : : * Each subscription's apply worker can only be restarted once per
1357 : : * wal_retrieve_retry_interval, so that errors do not cause us to
1358 : : * repeatedly restart the worker as fast as possible. In cases
1359 : : * where a restart is expected (e.g., subscription parameter
1360 : : * changes), another process should remove the last-start entry
1361 : : * for the subscription so that the worker can be restarted
1362 : : * without waiting for wal_retrieve_retry_interval to elapse.
1363 : : */
1364 : 385 : last_start = ApplyLauncherGetWorkerStartTime(sub->oid);
1365 : 385 : now = GetCurrentTimestamp();
1366 [ + + ]: 385 : if (last_start == 0 ||
1367 [ + + ]: 249 : (elapsed = TimestampDifferenceMilliseconds(last_start, now)) >= wal_retrieve_retry_interval)
1368 : : {
1369 : 236 : ApplyLauncherSetWorkerStartTime(sub->oid, now);
1370 [ + + ]: 241 : if (!logicalrep_worker_launch(WORKERTYPE_APPLY,
1371 : 236 : sub->dbid, sub->oid, sub->name,
1372 : : sub->owner, InvalidOid,
1373 : : DSM_HANDLE_INVALID,
1374 [ + + ]: 238 : sub->retaindeadtuples &&
1375 [ + - ]: 238 : sub->retentionactive))
1376 : : {
1377 : : /*
1378 : : * We get here either if we failed to launch a worker
1379 : : * (perhaps for resource-exhaustion reasons) or if we
1380 : : * launched one but it immediately quit. Either way, it
1381 : : * seems appropriate to try again after
1382 : : * wal_retrieve_retry_interval.
1383 : : */
1384 : 9 : wait_time = Min(wait_time,
1385 : : wal_retrieve_retry_interval);
1386 : : }
1387 : : }
1388 : : else
1389 : : {
1390 : 149 : wait_time = Min(wait_time,
1391 : : wal_retrieve_retry_interval - elapsed);
1392 : : }
1393 : : }
1394 : :
1395 : : /*
1396 : : * Drop the CONFLICT_DETECTION_SLOT slot if there is no subscription
1397 : : * that requires us to retain dead tuples. Otherwise, if required,
1398 : : * advance the slot's xmin to protect dead tuples required for the
1399 : : * conflict detection.
1400 : : *
1401 : : * Additionally, if all apply workers for subscriptions with
1402 : : * retain_dead_tuples enabled have requested to stop retention, the
1403 : : * slot's xmin will be set to InvalidTransactionId allowing the
1404 : : * removal of dead tuples.
1405 : : */
1406 [ + + ]: 3330 : if (MyReplicationSlot)
1407 : : {
1408 [ + + ]: 99 : if (!retain_dead_tuples)
1409 : : /* XXX unclear why we don't request logical decoding disable */
1410 : 1 : ReplicationSlotDropAcquired(false);
1411 [ + + ]: 98 : else if (can_update_xmin)
1412 : 94 : update_conflict_slot_xmin(xmin);
1413 : : }
1414 : :
1415 : : /* Switch back to original memory context. */
1416 : 3330 : MemoryContextSwitchTo(oldctx);
1417 : : /* Clean the temporary memory. */
1418 : 3330 : MemoryContextDelete(subctx);
1419 : :
1420 : : /* Wait for more work. */
1421 : 3330 : rc = WaitLatch(MyLatch,
1422 : : WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
1423 : : wait_time,
1424 : : WAIT_EVENT_LOGICAL_LAUNCHER_MAIN);
1425 : :
1426 [ + + ]: 3327 : if (rc & WL_LATCH_SET)
1427 : : {
1428 : 3293 : ResetLatch(MyLatch);
1429 [ + + ]: 3293 : CHECK_FOR_INTERRUPTS();
1430 : : }
1431 : :
1432 [ + + ]: 2815 : if (ConfigReloadPending)
1433 : : {
1434 : 51 : ConfigReloadPending = false;
1435 : 51 : ProcessConfigFile(PGC_SIGHUP);
1436 : : }
1437 : : }
1438 : :
1439 : : /* Not reachable */
1440 : : }
1441 : :
1442 : : /*
1443 : : * Determine the minimum non-removable transaction ID across all apply workers
1444 : : * for subscriptions that have retain_dead_tuples enabled. Store the result
1445 : : * in *xmin.
1446 : : */
1447 : : static void
1448 : 94 : compute_min_nonremovable_xid(LogicalRepWorker *worker, TransactionId *xmin)
1449 : : {
1450 : : TransactionId nonremovable_xid;
1451 : :
1452 : : Assert(worker != NULL);
1453 : :
1454 : : /*
1455 : : * The replication slot for conflict detection must be created before the
1456 : : * worker starts.
1457 : : */
1458 : : Assert(MyReplicationSlot);
1459 : :
1460 : 94 : SpinLockAcquire(&worker->relmutex);
1461 : 94 : nonremovable_xid = worker->oldest_nonremovable_xid;
1462 : 94 : SpinLockRelease(&worker->relmutex);
1463 : :
1464 : : /*
1465 : : * Return if the apply worker has stopped retention concurrently.
1466 : : *
1467 : : * Although this function is invoked only when retentionactive is true,
1468 : : * the apply worker might stop retention after the launcher fetches the
1469 : : * retentionactive flag.
1470 : : */
1471 [ - + ]: 94 : if (!TransactionIdIsValid(nonremovable_xid))
1472 : 0 : return;
1473 : :
1474 [ - + - - ]: 94 : if (!TransactionIdIsValid(*xmin) ||
1475 : 0 : TransactionIdPrecedes(nonremovable_xid, *xmin))
1476 : 94 : *xmin = nonremovable_xid;
1477 : : }
1478 : :
1479 : : /*
1480 : : * Acquire the replication slot used to retain information for conflict
1481 : : * detection, if it exists.
1482 : : *
1483 : : * Return true if successfully acquired, otherwise return false.
1484 : : */
1485 : : static bool
1486 : 520 : acquire_conflict_slot_if_exists(void)
1487 : : {
1488 [ + + ]: 520 : if (!SearchNamedReplicationSlot(CONFLICT_DETECTION_SLOT, true))
1489 : 519 : return false;
1490 : :
1491 : 1 : ReplicationSlotAcquire(CONFLICT_DETECTION_SLOT, true, false);
1492 : 1 : return true;
1493 : : }
1494 : :
1495 : : /*
1496 : : * Update the xmin the replication slot used to retain information required
1497 : : * for conflict detection.
1498 : : */
1499 : : static void
1500 : 94 : update_conflict_slot_xmin(TransactionId new_xmin)
1501 : : {
1502 : : Assert(MyReplicationSlot);
1503 : : Assert(!TransactionIdIsValid(new_xmin) ||
1504 : : TransactionIdPrecedesOrEquals(MyReplicationSlot->data.xmin, new_xmin));
1505 : :
1506 : : /* Return if the xmin value of the slot cannot be updated */
1507 [ + + ]: 94 : if (TransactionIdEquals(MyReplicationSlot->data.xmin, new_xmin))
1508 : 72 : return;
1509 : :
1510 : 22 : SpinLockAcquire(&MyReplicationSlot->mutex);
1511 : 22 : MyReplicationSlot->effective_xmin = new_xmin;
1512 : 22 : MyReplicationSlot->data.xmin = new_xmin;
1513 : 22 : SpinLockRelease(&MyReplicationSlot->mutex);
1514 : :
1515 [ - + ]: 22 : elog(DEBUG1, "updated xmin: %u", MyReplicationSlot->data.xmin);
1516 : :
1517 : 22 : ReplicationSlotMarkDirty();
1518 : 22 : ReplicationSlotsComputeRequiredXmin(false);
1519 : :
1520 : : /*
1521 : : * Like PhysicalConfirmReceivedLocation(), do not save slot information
1522 : : * each time. This is acceptable because all concurrent transactions on
1523 : : * the publisher that require the data preceding the slot's xmin should
1524 : : * have already been applied and flushed on the subscriber before the xmin
1525 : : * is advanced. So, even if the slot's xmin regresses after a restart, it
1526 : : * will be advanced again in the next cycle. Therefore, no data required
1527 : : * for conflict detection will be prematurely removed.
1528 : : */
1529 : 22 : return;
1530 : : }
1531 : :
1532 : : /*
1533 : : * Initialize the xmin for the conflict detection slot.
1534 : : */
1535 : : static void
1536 : 4 : init_conflict_slot_xmin(void)
1537 : : {
1538 : : TransactionId xmin_horizon;
1539 : :
1540 : : /* Replication slot must exist but shouldn't be initialized. */
1541 : : Assert(MyReplicationSlot &&
1542 : : !TransactionIdIsValid(MyReplicationSlot->data.xmin));
1543 : :
1544 : 4 : LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
1545 : 4 : LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
1546 : :
1547 : 4 : xmin_horizon = GetOldestSafeDecodingTransactionId(false);
1548 : :
1549 : 4 : SpinLockAcquire(&MyReplicationSlot->mutex);
1550 : 4 : MyReplicationSlot->effective_xmin = xmin_horizon;
1551 : 4 : MyReplicationSlot->data.xmin = xmin_horizon;
1552 : 4 : SpinLockRelease(&MyReplicationSlot->mutex);
1553 : :
1554 : 4 : ReplicationSlotsComputeRequiredXmin(true);
1555 : :
1556 : 4 : LWLockRelease(ProcArrayLock);
1557 : 4 : LWLockRelease(ReplicationSlotControlLock);
1558 : :
1559 : : /* Write this slot to disk */
1560 : 4 : ReplicationSlotMarkDirty();
1561 : 4 : ReplicationSlotSave();
1562 : 4 : }
1563 : :
1564 : : /*
1565 : : * Create and acquire the replication slot used to retain information for
1566 : : * conflict detection, if not yet.
1567 : : */
1568 : : void
1569 : 99 : CreateConflictDetectionSlot(void)
1570 : : {
1571 : : /* Exit early, if the replication slot is already created and acquired */
1572 [ + + ]: 99 : if (MyReplicationSlot)
1573 : 95 : return;
1574 : :
1575 [ + - ]: 4 : ereport(LOG,
1576 : : errmsg("creating replication conflict detection slot"));
1577 : :
1578 : 4 : ReplicationSlotCreate(CONFLICT_DETECTION_SLOT, false, RS_PERSISTENT, false,
1579 : : false, false, false);
1580 : :
1581 : 4 : init_conflict_slot_xmin();
1582 : : }
1583 : :
1584 : : /*
1585 : : * Is current process the logical replication launcher?
1586 : : */
1587 : : bool
1588 : 2818 : IsLogicalLauncher(void)
1589 : : {
1590 : 2818 : return LogicalRepCtx->launcher_pid == MyProcPid;
1591 : : }
1592 : :
1593 : : /*
1594 : : * Return the pid of the leader apply worker if the given pid is the pid of a
1595 : : * parallel apply worker, otherwise, return InvalidPid.
1596 : : */
1597 : : pid_t
1598 : 777 : GetLeaderApplyWorkerPid(pid_t pid)
1599 : : {
1600 : 777 : int leader_pid = InvalidPid;
1601 : : int i;
1602 : :
1603 : 777 : LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
1604 : :
1605 [ + + ]: 3885 : for (i = 0; i < max_logical_replication_workers; i++)
1606 : : {
1607 : 3108 : LogicalRepWorker *w = &LogicalRepCtx->workers[i];
1608 : :
1609 [ + + - + : 3108 : if (isParallelApplyWorker(w) && w->proc && pid == w->proc->pid)
- - - - ]
1610 : : {
1611 : 0 : leader_pid = w->leader_pid;
1612 : 0 : break;
1613 : : }
1614 : : }
1615 : :
1616 : 777 : LWLockRelease(LogicalRepWorkerLock);
1617 : :
1618 : 777 : return leader_pid;
1619 : : }
1620 : :
1621 : : /*
1622 : : * Returns state of the subscriptions.
1623 : : */
1624 : : Datum
1625 : 1 : pg_stat_get_subscription(PG_FUNCTION_ARGS)
1626 : : {
1627 : : #define PG_STAT_GET_SUBSCRIPTION_COLS 10
1628 [ - + ]: 1 : Oid subid = PG_ARGISNULL(0) ? InvalidOid : PG_GETARG_OID(0);
1629 : : int i;
1630 : 1 : ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
1631 : :
1632 : 1 : InitMaterializedSRF(fcinfo, 0);
1633 : :
1634 : : /* Make sure we get consistent view of the workers. */
1635 : 1 : LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
1636 : :
1637 [ + + ]: 5 : for (i = 0; i < max_logical_replication_workers; i++)
1638 : : {
1639 : : /* for each row */
1640 : 4 : Datum values[PG_STAT_GET_SUBSCRIPTION_COLS] = {0};
1641 : 4 : bool nulls[PG_STAT_GET_SUBSCRIPTION_COLS] = {0};
1642 : : int worker_pid;
1643 : : LogicalRepWorker worker;
1644 : :
1645 : 4 : memcpy(&worker, &LogicalRepCtx->workers[i],
1646 : : sizeof(LogicalRepWorker));
1647 [ + + - + ]: 4 : if (!worker.proc || !IsBackendPid(worker.proc->pid))
1648 : 2 : continue;
1649 : :
1650 [ - + - - ]: 2 : if (OidIsValid(subid) && worker.subid != subid)
1651 : 0 : continue;
1652 : :
1653 : 2 : worker_pid = worker.proc->pid;
1654 : :
1655 : 2 : values[0] = ObjectIdGetDatum(worker.subid);
1656 [ + - - + ]: 2 : if (isTableSyncWorker(&worker))
1657 : 0 : values[1] = ObjectIdGetDatum(worker.relid);
1658 : : else
1659 : 2 : nulls[1] = true;
1660 : 2 : values[2] = Int32GetDatum(worker_pid);
1661 : :
1662 [ + - - + ]: 2 : if (isParallelApplyWorker(&worker))
1663 : 0 : values[3] = Int32GetDatum(worker.leader_pid);
1664 : : else
1665 : 2 : nulls[3] = true;
1666 : :
1667 [ - + ]: 2 : if (!XLogRecPtrIsValid(worker.last_lsn))
1668 : 0 : nulls[4] = true;
1669 : : else
1670 : 2 : values[4] = LSNGetDatum(worker.last_lsn);
1671 [ - + ]: 2 : if (worker.last_send_time == 0)
1672 : 0 : nulls[5] = true;
1673 : : else
1674 : 2 : values[5] = TimestampTzGetDatum(worker.last_send_time);
1675 [ - + ]: 2 : if (worker.last_recv_time == 0)
1676 : 0 : nulls[6] = true;
1677 : : else
1678 : 2 : values[6] = TimestampTzGetDatum(worker.last_recv_time);
1679 [ - + ]: 2 : if (!XLogRecPtrIsValid(worker.reply_lsn))
1680 : 0 : nulls[7] = true;
1681 : : else
1682 : 2 : values[7] = LSNGetDatum(worker.reply_lsn);
1683 [ - + ]: 2 : if (worker.reply_time == 0)
1684 : 0 : nulls[8] = true;
1685 : : else
1686 : 2 : values[8] = TimestampTzGetDatum(worker.reply_time);
1687 : :
1688 [ + - - - : 2 : switch (worker.type)
- - ]
1689 : : {
1690 : 2 : case WORKERTYPE_APPLY:
1691 : 2 : values[9] = CStringGetTextDatum("apply");
1692 : 2 : break;
1693 : 0 : case WORKERTYPE_PARALLEL_APPLY:
1694 : 0 : values[9] = CStringGetTextDatum("parallel apply");
1695 : 0 : break;
1696 : 0 : case WORKERTYPE_SEQUENCESYNC:
1697 : 0 : values[9] = CStringGetTextDatum("sequence synchronization");
1698 : 0 : break;
1699 : 0 : case WORKERTYPE_TABLESYNC:
1700 : 0 : values[9] = CStringGetTextDatum("table synchronization");
1701 : 0 : break;
1702 : 0 : case WORKERTYPE_UNKNOWN:
1703 : : /* Should never happen. */
1704 [ # # ]: 0 : elog(ERROR, "unknown worker type");
1705 : : }
1706 : :
1707 : 2 : tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
1708 : : values, nulls);
1709 : :
1710 : : /*
1711 : : * If only a single subscription was requested, and we found it,
1712 : : * break.
1713 : : */
1714 [ - + ]: 2 : if (OidIsValid(subid))
1715 : 0 : break;
1716 : : }
1717 : :
1718 : 1 : LWLockRelease(LogicalRepWorkerLock);
1719 : :
1720 : 1 : return (Datum) 0;
1721 : : }
|
