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