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