PostgreSQL Source Code  git master
launcher.c
Go to the documentation of this file.
1 /*-------------------------------------------------------------------------
2  * launcher.c
3  * PostgreSQL logical replication worker launcher process
4  *
5  * Copyright (c) 2016-2024, 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"
27 #include "funcapi.h"
28 #include "lib/dshash.h"
29 #include "libpq/pqsignal.h"
30 #include "miscadmin.h"
31 #include "pgstat.h"
32 #include "postmaster/bgworker.h"
34 #include "postmaster/interrupt.h"
35 #include "postmaster/postmaster.h"
38 #include "replication/slot.h"
41 #include "storage/ipc.h"
42 #include "storage/proc.h"
43 #include "storage/procarray.h"
44 #include "storage/procsignal.h"
45 #include "tcop/tcopprot.h"
46 #include "utils/builtins.h"
47 #include "utils/memutils.h"
48 #include "utils/pg_lsn.h"
49 #include "utils/ps_status.h"
50 #include "utils/snapmgr.h"
51 #include "utils/timeout.h"
52 
53 /* max sleep time between cycles (3min) */
54 #define DEFAULT_NAPTIME_PER_CYCLE 180000L
55 
56 /* GUC variables */
60 
62 
63 typedef struct LogicalRepCtxStruct
64 {
65  /* Supervisor process. */
66  pid_t launcher_pid;
67 
68  /* Hash table holding last start times of subscriptions' apply workers. */
71 
72  /* Background workers. */
75 
77 
78 /* an entry in the last-start-times shared hash table */
80 {
81  Oid subid; /* OID of logrep subscription (hash key) */
82  TimestampTz last_start_time; /* last time its apply worker was started */
84 
85 /* parameters for the last-start-times shared hash table */
86 static const dshash_parameters dsh_params = {
87  sizeof(Oid),
92 };
93 
96 
97 static bool on_commit_launcher_wakeup = false;
98 
99 
100 static void ApplyLauncherWakeup(void);
101 static void logicalrep_launcher_onexit(int code, Datum arg);
102 static void logicalrep_worker_onexit(int code, Datum arg);
103 static void logicalrep_worker_detach(void);
104 static void logicalrep_worker_cleanup(LogicalRepWorker *worker);
105 static int logicalrep_pa_worker_count(Oid subid);
106 static void logicalrep_launcher_attach_dshmem(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 *
119 {
120  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  resultcxt = CurrentMemoryContext;
128 
129  /*
130  * Start a transaction so we can access pg_database, and get a snapshot.
131  * We don't have a use for the snapshot itself, but we're interested in
132  * the secondary effect that it sets RecentGlobalXmin. (This is critical
133  * for anything that reads heap pages, because HOT may decide to prune
134  * them even if the process doesn't attempt to modify any tuples.)
135  *
136  * FIXME: This comment is inaccurate / the code buggy. A snapshot that is
137  * not pushed/active does not reliably prevent HOT pruning (->xmin could
138  * e.g. be cleared when cache invalidations are processed).
139  */
141  (void) GetTransactionSnapshot();
142 
143  rel = table_open(SubscriptionRelationId, AccessShareLock);
144  scan = table_beginscan_catalog(rel, 0, NULL);
145 
147  {
149  Subscription *sub;
150  MemoryContext oldcxt;
151 
152  /*
153  * Allocate our results in the caller's context, not the
154  * transaction's. We do this inside the loop, and restore the original
155  * context at the end, so that leaky things like heap_getnext() are
156  * not called in a potentially long-lived context.
157  */
158  oldcxt = MemoryContextSwitchTo(resultcxt);
159 
160  sub = (Subscription *) palloc0(sizeof(Subscription));
161  sub->oid = subform->oid;
162  sub->dbid = subform->subdbid;
163  sub->owner = subform->subowner;
164  sub->enabled = subform->subenabled;
165  sub->name = pstrdup(NameStr(subform->subname));
166  /* We don't fill fields we are not interested in. */
167 
168  res = lappend(res, sub);
169  MemoryContextSwitchTo(oldcxt);
170  }
171 
172  table_endscan(scan);
174 
176 
177  return res;
178 }
179 
180 /*
181  * Wait for a background worker to start up and attach to the shmem context.
182  *
183  * This is only needed for cleaning up the shared memory in case the worker
184  * fails to attach.
185  *
186  * Returns whether the attach was successful.
187  */
188 static bool
190  uint16 generation,
191  BackgroundWorkerHandle *handle)
192 {
193  BgwHandleStatus status;
194  int rc;
195 
196  for (;;)
197  {
198  pid_t pid;
199 
201 
202  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
203 
204  /* Worker either died or has started. Return false if died. */
205  if (!worker->in_use || worker->proc)
206  {
207  LWLockRelease(LogicalRepWorkerLock);
208  return worker->in_use;
209  }
210 
211  LWLockRelease(LogicalRepWorkerLock);
212 
213  /* Check if worker has died before attaching, and clean up after it. */
214  status = GetBackgroundWorkerPid(handle, &pid);
215 
216  if (status == BGWH_STOPPED)
217  {
218  LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
219  /* Ensure that this was indeed the worker we waited for. */
220  if (generation == worker->generation)
222  LWLockRelease(LogicalRepWorkerLock);
223  return false;
224  }
225 
226  /*
227  * We need timeout because we generally don't get notified via latch
228  * about the worker attach. But we don't expect to have to wait long.
229  */
230  rc = WaitLatch(MyLatch,
232  10L, WAIT_EVENT_BGWORKER_STARTUP);
233 
234  if (rc & WL_LATCH_SET)
235  {
238  }
239  }
240 }
241 
242 /*
243  * Walks the workers array and searches for one that matches given
244  * subscription id and relid.
245  *
246  * We are only interested in the leader apply worker or table sync worker.
247  */
249 logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
250 {
251  int i;
252  LogicalRepWorker *res = NULL;
253 
254  Assert(LWLockHeldByMe(LogicalRepWorkerLock));
255 
256  /* Search for attached worker for a given subscription id. */
257  for (i = 0; i < max_logical_replication_workers; i++)
258  {
260 
261  /* Skip parallel apply workers. */
262  if (isParallelApplyWorker(w))
263  continue;
264 
265  if (w->in_use && w->subid == subid && w->relid == relid &&
266  (!only_running || w->proc))
267  {
268  res = w;
269  break;
270  }
271  }
272 
273  return res;
274 }
275 
276 /*
277  * Similar to logicalrep_worker_find(), but returns a list of all workers for
278  * the subscription, instead of just one.
279  */
280 List *
281 logicalrep_workers_find(Oid subid, bool only_running)
282 {
283  int i;
284  List *res = NIL;
285 
286  Assert(LWLockHeldByMe(LogicalRepWorkerLock));
287 
288  /* Search for attached worker for a given subscription id. */
289  for (i = 0; i < max_logical_replication_workers; i++)
290  {
292 
293  if (w->in_use && w->subid == subid && (!only_running || w->proc))
294  res = lappend(res, w);
295  }
296 
297  return res;
298 }
299 
300 /*
301  * Start new logical replication background worker, if possible.
302  *
303  * Returns true on success, false on failure.
304  */
305 bool
307  Oid dbid, Oid subid, const char *subname, Oid userid,
308  Oid relid, dsm_handle subworker_dsm)
309 {
310  BackgroundWorker bgw;
311  BackgroundWorkerHandle *bgw_handle;
312  uint16 generation;
313  int i;
314  int slot = 0;
315  LogicalRepWorker *worker = NULL;
316  int nsyncworkers;
317  int nparallelapplyworkers;
319  bool is_tablesync_worker = (wtype == WORKERTYPE_TABLESYNC);
320  bool is_parallel_apply_worker = (wtype == WORKERTYPE_PARALLEL_APPLY);
321 
322  /*----------
323  * Sanity checks:
324  * - must be valid worker type
325  * - tablesync workers are only ones to have relid
326  * - parallel apply worker is the only kind of subworker
327  */
328  Assert(wtype != WORKERTYPE_UNKNOWN);
329  Assert(is_tablesync_worker == OidIsValid(relid));
330  Assert(is_parallel_apply_worker == (subworker_dsm != DSM_HANDLE_INVALID));
331 
332  ereport(DEBUG1,
333  (errmsg_internal("starting logical replication worker for subscription \"%s\"",
334  subname)));
335 
336  /* Report this after the initial starting message for consistency. */
337  if (max_replication_slots == 0)
338  ereport(ERROR,
339  (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
340  errmsg("cannot start logical replication workers when max_replication_slots = 0")));
341 
342  /*
343  * We need to do the modification of the shared memory under lock so that
344  * we have consistent view.
345  */
346  LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
347 
348 retry:
349  /* Find unused worker slot. */
350  for (i = 0; i < max_logical_replication_workers; i++)
351  {
353 
354  if (!w->in_use)
355  {
356  worker = w;
357  slot = i;
358  break;
359  }
360  }
361 
362  nsyncworkers = logicalrep_sync_worker_count(subid);
363 
365 
366  /*
367  * If we didn't find a free slot, try to do garbage collection. The
368  * reason we do this is because if some worker failed to start up and its
369  * parent has crashed while waiting, the in_use state was never cleared.
370  */
371  if (worker == NULL || nsyncworkers >= max_sync_workers_per_subscription)
372  {
373  bool did_cleanup = false;
374 
375  for (i = 0; i < max_logical_replication_workers; i++)
376  {
378 
379  /*
380  * If the worker was marked in use but didn't manage to attach in
381  * time, clean it up.
382  */
383  if (w->in_use && !w->proc &&
386  {
387  elog(WARNING,
388  "logical replication worker for subscription %u took too long to start; canceled",
389  w->subid);
390 
392  did_cleanup = true;
393  }
394  }
395 
396  if (did_cleanup)
397  goto retry;
398  }
399 
400  /*
401  * We don't allow to invoke more sync workers once we have reached the
402  * sync worker limit per subscription. So, just return silently as we
403  * might get here because of an otherwise harmless race condition.
404  */
405  if (is_tablesync_worker && nsyncworkers >= max_sync_workers_per_subscription)
406  {
407  LWLockRelease(LogicalRepWorkerLock);
408  return false;
409  }
410 
411  nparallelapplyworkers = logicalrep_pa_worker_count(subid);
412 
413  /*
414  * Return false if the number of parallel apply workers reached the limit
415  * per subscription.
416  */
417  if (is_parallel_apply_worker &&
418  nparallelapplyworkers >= max_parallel_apply_workers_per_subscription)
419  {
420  LWLockRelease(LogicalRepWorkerLock);
421  return false;
422  }
423 
424  /*
425  * However if there are no more free worker slots, inform user about it
426  * before exiting.
427  */
428  if (worker == NULL)
429  {
430  LWLockRelease(LogicalRepWorkerLock);
432  (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
433  errmsg("out of logical replication worker slots"),
434  errhint("You might need to increase %s.", "max_logical_replication_workers")));
435  return false;
436  }
437 
438  /* Prepare the worker slot. */
439  worker->type = wtype;
440  worker->launch_time = now;
441  worker->in_use = true;
442  worker->generation++;
443  worker->proc = NULL;
444  worker->dbid = dbid;
445  worker->userid = userid;
446  worker->subid = subid;
447  worker->relid = relid;
448  worker->relstate = SUBREL_STATE_UNKNOWN;
450  worker->stream_fileset = NULL;
451  worker->leader_pid = is_parallel_apply_worker ? MyProcPid : InvalidPid;
452  worker->parallel_apply = is_parallel_apply_worker;
453  worker->last_lsn = InvalidXLogRecPtr;
456  worker->reply_lsn = InvalidXLogRecPtr;
457  TIMESTAMP_NOBEGIN(worker->reply_time);
458 
459  /* Before releasing lock, remember generation for future identification. */
460  generation = worker->generation;
461 
462  LWLockRelease(LogicalRepWorkerLock);
463 
464  /* Register the new dynamic worker. */
465  memset(&bgw, 0, sizeof(bgw));
469  snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres");
470 
471  switch (worker->type)
472  {
473  case WORKERTYPE_APPLY:
474  snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
476  "logical replication apply worker for subscription %u",
477  subid);
478  snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication apply worker");
479  break;
480 
482  snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain");
484  "logical replication parallel apply worker for subscription %u",
485  subid);
486  snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
487 
488  memcpy(bgw.bgw_extra, &subworker_dsm, sizeof(dsm_handle));
489  break;
490 
492  snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TablesyncWorkerMain");
494  "logical replication tablesync worker for subscription %u sync %u",
495  subid,
496  relid);
497  snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication tablesync worker");
498  break;
499 
500  case WORKERTYPE_UNKNOWN:
501  /* Should never happen. */
502  elog(ERROR, "unknown worker type");
503  }
504 
507  bgw.bgw_main_arg = Int32GetDatum(slot);
508 
509  if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
510  {
511  /* Failed to start worker, so clean up the worker slot. */
512  LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
513  Assert(generation == worker->generation);
515  LWLockRelease(LogicalRepWorkerLock);
516 
518  (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
519  errmsg("out of background worker slots"),
520  errhint("You might need to increase %s.", "max_worker_processes")));
521  return false;
522  }
523 
524  /* Now wait until it attaches. */
525  return WaitForReplicationWorkerAttach(worker, generation, bgw_handle);
526 }
527 
528 /*
529  * Internal function to stop the worker and wait until it detaches from the
530  * slot.
531  */
532 static void
534 {
535  uint16 generation;
536 
537  Assert(LWLockHeldByMeInMode(LogicalRepWorkerLock, LW_SHARED));
538 
539  /*
540  * Remember which generation was our worker so we can check if what we see
541  * is still the same one.
542  */
543  generation = worker->generation;
544 
545  /*
546  * If we found a worker but it does not have proc set then it is still
547  * starting up; wait for it to finish starting and then kill it.
548  */
549  while (worker->in_use && !worker->proc)
550  {
551  int rc;
552 
553  LWLockRelease(LogicalRepWorkerLock);
554 
555  /* Wait a bit --- we don't expect to have to wait long. */
556  rc = WaitLatch(MyLatch,
558  10L, WAIT_EVENT_BGWORKER_STARTUP);
559 
560  if (rc & WL_LATCH_SET)
561  {
564  }
565 
566  /* Recheck worker status. */
567  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
568 
569  /*
570  * Check whether the worker slot is no longer used, which would mean
571  * that the worker has exited, or whether the worker generation is
572  * different, meaning that a different worker has taken the slot.
573  */
574  if (!worker->in_use || worker->generation != generation)
575  return;
576 
577  /* Worker has assigned proc, so it has started. */
578  if (worker->proc)
579  break;
580  }
581 
582  /* Now terminate the worker ... */
583  kill(worker->proc->pid, signo);
584 
585  /* ... and wait for it to die. */
586  for (;;)
587  {
588  int rc;
589 
590  /* is it gone? */
591  if (!worker->proc || worker->generation != generation)
592  break;
593 
594  LWLockRelease(LogicalRepWorkerLock);
595 
596  /* Wait a bit --- we don't expect to have to wait long. */
597  rc = WaitLatch(MyLatch,
599  10L, WAIT_EVENT_BGWORKER_SHUTDOWN);
600 
601  if (rc & WL_LATCH_SET)
602  {
605  }
606 
607  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
608  }
609 }
610 
611 /*
612  * Stop the logical replication worker for subid/relid, if any.
613  */
614 void
616 {
617  LogicalRepWorker *worker;
618 
619  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
620 
621  worker = logicalrep_worker_find(subid, relid, false);
622 
623  if (worker)
624  {
625  Assert(!isParallelApplyWorker(worker));
626  logicalrep_worker_stop_internal(worker, SIGTERM);
627  }
628 
629  LWLockRelease(LogicalRepWorkerLock);
630 }
631 
632 /*
633  * Stop the given logical replication parallel apply worker.
634  *
635  * Node that the function sends SIGINT instead of SIGTERM to the parallel apply
636  * worker so that the worker exits cleanly.
637  */
638 void
640 {
641  int slot_no;
642  uint16 generation;
643  LogicalRepWorker *worker;
644 
645  SpinLockAcquire(&winfo->shared->mutex);
646  generation = winfo->shared->logicalrep_worker_generation;
647  slot_no = winfo->shared->logicalrep_worker_slot_no;
648  SpinLockRelease(&winfo->shared->mutex);
649 
650  Assert(slot_no >= 0 && slot_no < max_logical_replication_workers);
651 
652  /*
653  * Detach from the error_mq_handle for the parallel apply worker before
654  * stopping it. This prevents the leader apply worker from trying to
655  * receive the message from the error queue that might already be detached
656  * by the parallel apply worker.
657  */
658  if (winfo->error_mq_handle)
659  {
661  winfo->error_mq_handle = NULL;
662  }
663 
664  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
665 
666  worker = &LogicalRepCtx->workers[slot_no];
667  Assert(isParallelApplyWorker(worker));
668 
669  /*
670  * Only stop the worker if the generation matches and the worker is alive.
671  */
672  if (worker->generation == generation && worker->proc)
673  logicalrep_worker_stop_internal(worker, SIGINT);
674 
675  LWLockRelease(LogicalRepWorkerLock);
676 }
677 
678 /*
679  * Wake up (using latch) any logical replication worker for specified sub/rel.
680  */
681 void
683 {
684  LogicalRepWorker *worker;
685 
686  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
687 
688  worker = logicalrep_worker_find(subid, relid, true);
689 
690  if (worker)
692 
693  LWLockRelease(LogicalRepWorkerLock);
694 }
695 
696 /*
697  * Wake up (using latch) the specified logical replication worker.
698  *
699  * Caller must hold lock, else worker->proc could change under us.
700  */
701 void
703 {
704  Assert(LWLockHeldByMe(LogicalRepWorkerLock));
705 
706  SetLatch(&worker->proc->procLatch);
707 }
708 
709 /*
710  * Attach to a slot.
711  */
712 void
714 {
715  /* Block concurrent access. */
716  LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
717 
718  Assert(slot >= 0 && slot < max_logical_replication_workers);
720 
722  {
723  LWLockRelease(LogicalRepWorkerLock);
724  ereport(ERROR,
725  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
726  errmsg("logical replication worker slot %d is empty, cannot attach",
727  slot)));
728  }
729 
731  {
732  LWLockRelease(LogicalRepWorkerLock);
733  ereport(ERROR,
734  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
735  errmsg("logical replication worker slot %d is already used by "
736  "another worker, cannot attach", slot)));
737  }
738 
741 
742  LWLockRelease(LogicalRepWorkerLock);
743 }
744 
745 /*
746  * Stop the parallel apply workers if any, and detach the leader apply worker
747  * (cleans up the worker info).
748  */
749 static void
751 {
752  /* Stop the parallel apply workers. */
754  {
755  List *workers;
756  ListCell *lc;
757 
758  /*
759  * Detach from the error_mq_handle for all parallel apply workers
760  * before terminating them. This prevents the leader apply worker from
761  * receiving the worker termination message and sending it to logs
762  * when the same is already done by the parallel worker.
763  */
765 
766  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
767 
769  foreach(lc, workers)
770  {
772 
773  if (isParallelApplyWorker(w))
775  }
776 
777  LWLockRelease(LogicalRepWorkerLock);
778  }
779 
780  /* Block concurrent access. */
781  LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
782 
784 
785  LWLockRelease(LogicalRepWorkerLock);
786 }
787 
788 /*
789  * Clean up worker info.
790  */
791 static void
793 {
794  Assert(LWLockHeldByMeInMode(LogicalRepWorkerLock, LW_EXCLUSIVE));
795 
796  worker->type = WORKERTYPE_UNKNOWN;
797  worker->in_use = false;
798  worker->proc = NULL;
799  worker->dbid = InvalidOid;
800  worker->userid = InvalidOid;
801  worker->subid = InvalidOid;
802  worker->relid = InvalidOid;
803  worker->leader_pid = InvalidPid;
804  worker->parallel_apply = false;
805 }
806 
807 /*
808  * Cleanup function for logical replication launcher.
809  *
810  * Called on logical replication launcher exit.
811  */
812 static void
814 {
816 }
817 
818 /*
819  * Cleanup function.
820  *
821  * Called on logical replication worker exit.
822  */
823 static void
825 {
826  /* Disconnect gracefully from the remote side. */
829 
831 
832  /* Cleanup fileset used for streaming transactions. */
833  if (MyLogicalRepWorker->stream_fileset != NULL)
835 
836  /*
837  * Session level locks may be acquired outside of a transaction in
838  * parallel apply mode and will not be released when the worker
839  * terminates, so manually release all locks before the worker exits.
840  *
841  * The locks will be acquired once the worker is initialized.
842  */
845 
847 }
848 
849 /*
850  * Count the number of registered (not necessarily running) sync workers
851  * for a subscription.
852  */
853 int
855 {
856  int i;
857  int res = 0;
858 
859  Assert(LWLockHeldByMe(LogicalRepWorkerLock));
860 
861  /* Search for attached worker for a given subscription id. */
862  for (i = 0; i < max_logical_replication_workers; i++)
863  {
865 
866  if (isTablesyncWorker(w) && w->subid == subid)
867  res++;
868  }
869 
870  return res;
871 }
872 
873 /*
874  * Count the number of registered (but not necessarily running) parallel apply
875  * workers for a subscription.
876  */
877 static int
879 {
880  int i;
881  int res = 0;
882 
883  Assert(LWLockHeldByMe(LogicalRepWorkerLock));
884 
885  /*
886  * Scan all attached parallel apply workers, only counting those which
887  * have the given subscription id.
888  */
889  for (i = 0; i < max_logical_replication_workers; i++)
890  {
892 
893  if (isParallelApplyWorker(w) && w->subid == subid)
894  res++;
895  }
896 
897  return res;
898 }
899 
900 /*
901  * ApplyLauncherShmemSize
902  * Compute space needed for replication launcher shared memory
903  */
904 Size
906 {
907  Size size;
908 
909  /*
910  * Need the fixed struct and the array of LogicalRepWorker.
911  */
912  size = sizeof(LogicalRepCtxStruct);
913  size = MAXALIGN(size);
915  sizeof(LogicalRepWorker)));
916  return size;
917 }
918 
919 /*
920  * ApplyLauncherRegister
921  * Register a background worker running the logical replication launcher.
922  */
923 void
925 {
926  BackgroundWorker bgw;
927 
928  /*
929  * The logical replication launcher is disabled during binary upgrades, to
930  * prevent logical replication workers from running on the source cluster.
931  * That could cause replication origins to move forward after having been
932  * copied to the target cluster, potentially creating conflicts with the
933  * copied data files.
934  */
936  return;
937 
938  memset(&bgw, 0, sizeof(bgw));
942  snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres");
943  snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyLauncherMain");
945  "logical replication launcher");
947  "logical replication launcher");
948  bgw.bgw_restart_time = 5;
949  bgw.bgw_notify_pid = 0;
950  bgw.bgw_main_arg = (Datum) 0;
951 
953 }
954 
955 /*
956  * ApplyLauncherShmemInit
957  * Allocate and initialize replication launcher shared memory
958  */
959 void
961 {
962  bool found;
963 
965  ShmemInitStruct("Logical Replication Launcher Data",
967  &found);
968 
969  if (!found)
970  {
971  int slot;
972 
974 
977 
978  /* Initialize memory and spin locks for each worker slot. */
979  for (slot = 0; slot < max_logical_replication_workers; slot++)
980  {
981  LogicalRepWorker *worker = &LogicalRepCtx->workers[slot];
982 
983  memset(worker, 0, sizeof(LogicalRepWorker));
984  SpinLockInit(&worker->relmutex);
985  }
986  }
987 }
988 
989 /*
990  * Initialize or attach to the dynamic shared hash table that stores the
991  * last-start times, if not already done.
992  * This must be called before accessing the table.
993  */
994 static void
996 {
997  MemoryContext oldcontext;
998 
999  /* Quick exit if we already did this. */
1001  last_start_times != NULL)
1002  return;
1003 
1004  /* Otherwise, use a lock to ensure only one process creates the table. */
1005  LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
1006 
1007  /* Be sure any local memory allocated by DSA routines is persistent. */
1009 
1011  {
1012  /* Initialize dynamic shared hash table for last-start times. */
1017 
1018  /* Store handles in shared memory for other backends to use. */
1021  }
1022  else if (!last_start_times)
1023  {
1024  /* Attach to existing dynamic shared hash table. */
1029  }
1030 
1031  MemoryContextSwitchTo(oldcontext);
1032  LWLockRelease(LogicalRepWorkerLock);
1033 }
1034 
1035 /*
1036  * Set the last-start time for the subscription.
1037  */
1038 static void
1040 {
1042  bool found;
1043 
1045 
1046  entry = dshash_find_or_insert(last_start_times, &subid, &found);
1047  entry->last_start_time = start_time;
1049 }
1050 
1051 /*
1052  * Return the last-start time for the subscription, or 0 if there isn't one.
1053  */
1054 static TimestampTz
1056 {
1058  TimestampTz ret;
1059 
1061 
1062  entry = dshash_find(last_start_times, &subid, false);
1063  if (entry == NULL)
1064  return 0;
1065 
1066  ret = entry->last_start_time;
1068 
1069  return ret;
1070 }
1071 
1072 /*
1073  * Remove the last-start-time entry for the subscription, if one exists.
1074  *
1075  * This has two use-cases: to remove the entry related to a subscription
1076  * that's been deleted or disabled (just to avoid leaking shared memory),
1077  * and to allow immediate restart of an apply worker that has exited
1078  * due to subscription parameter changes.
1079  */
1080 void
1082 {
1084 
1085  (void) dshash_delete_key(last_start_times, &subid);
1086 }
1087 
1088 /*
1089  * Wakeup the launcher on commit if requested.
1090  */
1091 void
1093 {
1094  if (isCommit)
1095  {
1098  }
1099 
1100  on_commit_launcher_wakeup = false;
1101 }
1102 
1103 /*
1104  * Request wakeup of the launcher on commit of the transaction.
1105  *
1106  * This is used to send launcher signal to stop sleeping and process the
1107  * subscriptions when current transaction commits. Should be used when new
1108  * tuple was added to the pg_subscription catalog.
1109 */
1110 void
1112 {
1115 }
1116 
1117 static void
1119 {
1120  if (LogicalRepCtx->launcher_pid != 0)
1122 }
1123 
1124 /*
1125  * Main loop for the apply launcher process.
1126  */
1127 void
1129 {
1130  ereport(DEBUG1,
1131  (errmsg_internal("logical replication launcher started")));
1132 
1134 
1137 
1138  /* Establish signal handlers. */
1140  pqsignal(SIGTERM, die);
1142 
1143  /*
1144  * Establish connection to nailed catalogs (we only ever access
1145  * pg_subscription).
1146  */
1147  BackgroundWorkerInitializeConnection(NULL, NULL, 0);
1148 
1149  /* Enter main loop */
1150  for (;;)
1151  {
1152  int rc;
1153  List *sublist;
1154  ListCell *lc;
1155  MemoryContext subctx;
1156  MemoryContext oldctx;
1157  long wait_time = DEFAULT_NAPTIME_PER_CYCLE;
1158 
1160 
1161  /* Use temporary context to avoid leaking memory across cycles. */
1163  "Logical Replication Launcher sublist",
1165  oldctx = MemoryContextSwitchTo(subctx);
1166 
1167  /* Start any missing workers for enabled subscriptions. */
1168  sublist = get_subscription_list();
1169  foreach(lc, sublist)
1170  {
1171  Subscription *sub = (Subscription *) lfirst(lc);
1172  LogicalRepWorker *w;
1173  TimestampTz last_start;
1174  TimestampTz now;
1175  long elapsed;
1176 
1177  if (!sub->enabled)
1178  continue;
1179 
1180  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
1181  w = logicalrep_worker_find(sub->oid, InvalidOid, false);
1182  LWLockRelease(LogicalRepWorkerLock);
1183 
1184  if (w != NULL)
1185  continue; /* worker is running already */
1186 
1187  /*
1188  * If the worker is eligible to start now, launch it. Otherwise,
1189  * adjust wait_time so that we'll wake up as soon as it can be
1190  * started.
1191  *
1192  * Each subscription's apply worker can only be restarted once per
1193  * wal_retrieve_retry_interval, so that errors do not cause us to
1194  * repeatedly restart the worker as fast as possible. In cases
1195  * where a restart is expected (e.g., subscription parameter
1196  * changes), another process should remove the last-start entry
1197  * for the subscription so that the worker can be restarted
1198  * without waiting for wal_retrieve_retry_interval to elapse.
1199  */
1200  last_start = ApplyLauncherGetWorkerStartTime(sub->oid);
1202  if (last_start == 0 ||
1204  {
1207  sub->dbid, sub->oid, sub->name,
1208  sub->owner, InvalidOid,
1210  }
1211  else
1212  {
1213  wait_time = Min(wait_time,
1214  wal_retrieve_retry_interval - elapsed);
1215  }
1216  }
1217 
1218  /* Switch back to original memory context. */
1219  MemoryContextSwitchTo(oldctx);
1220  /* Clean the temporary memory. */
1221  MemoryContextDelete(subctx);
1222 
1223  /* Wait for more work. */
1224  rc = WaitLatch(MyLatch,
1226  wait_time,
1227  WAIT_EVENT_LOGICAL_LAUNCHER_MAIN);
1228 
1229  if (rc & WL_LATCH_SET)
1230  {
1233  }
1234 
1235  if (ConfigReloadPending)
1236  {
1237  ConfigReloadPending = false;
1239  }
1240  }
1241 
1242  /* Not reachable */
1243 }
1244 
1245 /*
1246  * Is current process the logical replication launcher?
1247  */
1248 bool
1250 {
1252 }
1253 
1254 /*
1255  * Return the pid of the leader apply worker if the given pid is the pid of a
1256  * parallel apply worker, otherwise, return InvalidPid.
1257  */
1258 pid_t
1260 {
1261  int leader_pid = InvalidPid;
1262  int i;
1263 
1264  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
1265 
1266  for (i = 0; i < max_logical_replication_workers; i++)
1267  {
1269 
1270  if (isParallelApplyWorker(w) && w->proc && pid == w->proc->pid)
1271  {
1272  leader_pid = w->leader_pid;
1273  break;
1274  }
1275  }
1276 
1277  LWLockRelease(LogicalRepWorkerLock);
1278 
1279  return leader_pid;
1280 }
1281 
1282 /*
1283  * Returns state of the subscriptions.
1284  */
1285 Datum
1287 {
1288 #define PG_STAT_GET_SUBSCRIPTION_COLS 10
1289  Oid subid = PG_ARGISNULL(0) ? InvalidOid : PG_GETARG_OID(0);
1290  int i;
1291  ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
1292 
1293  InitMaterializedSRF(fcinfo, 0);
1294 
1295  /* Make sure we get consistent view of the workers. */
1296  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
1297 
1298  for (i = 0; i < max_logical_replication_workers; i++)
1299  {
1300  /* for each row */
1302  bool nulls[PG_STAT_GET_SUBSCRIPTION_COLS] = {0};
1303  int worker_pid;
1304  LogicalRepWorker worker;
1305 
1306  memcpy(&worker, &LogicalRepCtx->workers[i],
1307  sizeof(LogicalRepWorker));
1308  if (!worker.proc || !IsBackendPid(worker.proc->pid))
1309  continue;
1310 
1311  if (OidIsValid(subid) && worker.subid != subid)
1312  continue;
1313 
1314  worker_pid = worker.proc->pid;
1315 
1316  values[0] = ObjectIdGetDatum(worker.subid);
1317  if (isTablesyncWorker(&worker))
1318  values[1] = ObjectIdGetDatum(worker.relid);
1319  else
1320  nulls[1] = true;
1321  values[2] = Int32GetDatum(worker_pid);
1322 
1323  if (isParallelApplyWorker(&worker))
1324  values[3] = Int32GetDatum(worker.leader_pid);
1325  else
1326  nulls[3] = true;
1327 
1328  if (XLogRecPtrIsInvalid(worker.last_lsn))
1329  nulls[4] = true;
1330  else
1331  values[4] = LSNGetDatum(worker.last_lsn);
1332  if (worker.last_send_time == 0)
1333  nulls[5] = true;
1334  else
1336  if (worker.last_recv_time == 0)
1337  nulls[6] = true;
1338  else
1340  if (XLogRecPtrIsInvalid(worker.reply_lsn))
1341  nulls[7] = true;
1342  else
1343  values[7] = LSNGetDatum(worker.reply_lsn);
1344  if (worker.reply_time == 0)
1345  nulls[8] = true;
1346  else
1347  values[8] = TimestampTzGetDatum(worker.reply_time);
1348 
1349  switch (worker.type)
1350  {
1351  case WORKERTYPE_APPLY:
1352  values[9] = CStringGetTextDatum("apply");
1353  break;
1355  values[9] = CStringGetTextDatum("parallel apply");
1356  break;
1357  case WORKERTYPE_TABLESYNC:
1358  values[9] = CStringGetTextDatum("table synchronization");
1359  break;
1360  case WORKERTYPE_UNKNOWN:
1361  /* Should never happen. */
1362  elog(ERROR, "unknown worker type");
1363  }
1364 
1365  tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
1366  values, nulls);
1367 
1368  /*
1369  * If only a single subscription was requested, and we found it,
1370  * break.
1371  */
1372  if (OidIsValid(subid))
1373  break;
1374  }
1375 
1376  LWLockRelease(LogicalRepWorkerLock);
1377 
1378  return (Datum) 0;
1379 }
void pa_detach_all_error_mq(void)
bool InitializingApplyWorker
Definition: worker.c:343
WalReceiverConn * LogRepWorkerWalRcvConn
Definition: worker.c:321
long TimestampDifferenceMilliseconds(TimestampTz start_time, TimestampTz stop_time)
Definition: timestamp.c:1767
bool TimestampDifferenceExceeds(TimestampTz start_time, TimestampTz stop_time, int msec)
Definition: timestamp.c:1791
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1655
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1619
void RegisterBackgroundWorker(BackgroundWorker *worker)
Definition: bgworker.c:852
BgwHandleStatus GetBackgroundWorkerPid(BackgroundWorkerHandle *handle, pid_t *pidp)
Definition: bgworker.c:1072
bool RegisterDynamicBackgroundWorker(BackgroundWorker *worker, BackgroundWorkerHandle **handle)
Definition: bgworker.c:960
#define BGW_NEVER_RESTART
Definition: bgworker.h:85
BgwHandleStatus
Definition: bgworker.h:104
@ BGWH_STOPPED
Definition: bgworker.h:107
@ BgWorkerStart_RecoveryFinished
Definition: bgworker.h:81
#define BGWORKER_BACKEND_DATABASE_CONNECTION
Definition: bgworker.h:60
#define BGWORKER_SHMEM_ACCESS
Definition: bgworker.h:53
#define BGW_MAXLEN
Definition: bgworker.h:86
static Datum values[MAXATTR]
Definition: bootstrap.c:156
#define CStringGetTextDatum(s)
Definition: builtins.h:97
#define NameStr(name)
Definition: c.h:735
unsigned short uint16
Definition: c.h:494
#define Min(x, y)
Definition: c.h:993
#define MAXALIGN(LEN)
Definition: c.h:800
#define FLEXIBLE_ARRAY_MEMBER
Definition: c.h:387
#define OidIsValid(objectId)
Definition: c.h:764
size_t Size
Definition: c.h:594
int64 TimestampTz
Definition: timestamp.h:39
#define TIMESTAMP_NOBEGIN(j)
Definition: timestamp.h:159
dsa_area * dsa_attach(dsa_handle handle)
Definition: dsa.c:525
dsa_area * dsa_create(int tranche_id)
Definition: dsa.c:439
void dsa_pin_mapping(dsa_area *area)
Definition: dsa.c:650
dsa_handle dsa_get_handle(dsa_area *area)
Definition: dsa.c:513
void dsa_pin(dsa_area *area)
Definition: dsa.c:990
dsm_handle dsa_handle
Definition: dsa.h:100
#define DSA_HANDLE_INVALID
Definition: dsa.h:103
bool dshash_delete_key(dshash_table *hash_table, const void *key)
Definition: dshash.c:503
void dshash_release_lock(dshash_table *hash_table, void *entry)
Definition: dshash.c:558
void * dshash_find(dshash_table *hash_table, const void *key, bool exclusive)
Definition: dshash.c:390
dshash_table_handle dshash_get_hash_table_handle(dshash_table *hash_table)
Definition: dshash.c:367
dshash_hash dshash_memhash(const void *v, size_t size, void *arg)
Definition: dshash.c:581
void * dshash_find_or_insert(dshash_table *hash_table, const void *key, bool *found)
Definition: dshash.c:433
dshash_table * dshash_attach(dsa_area *area, const dshash_parameters *params, dshash_table_handle handle, void *arg)
Definition: dshash.c:270
int dshash_memcmp(const void *a, const void *b, size_t size, void *arg)
Definition: dshash.c:572
dshash_table * dshash_create(dsa_area *area, const dshash_parameters *params, void *arg)
Definition: dshash.c:206
#define DSHASH_HANDLE_INVALID
Definition: dshash.h:27
dsa_pointer dshash_table_handle
Definition: dshash.h:24
uint32 dsm_handle
Definition: dsm_impl.h:55
#define DSM_HANDLE_INVALID
Definition: dsm_impl.h:58
int errmsg_internal(const char *fmt,...)
Definition: elog.c:1162
int errhint(const char *fmt,...)
Definition: elog.c:1322
int errcode(int sqlerrcode)
Definition: elog.c:860
int errmsg(const char *fmt,...)
Definition: elog.c:1075
#define WARNING
Definition: elog.h:36
#define DEBUG1
Definition: elog.h:30
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:149
void FileSetDeleteAll(FileSet *fileset)
Definition: fileset.c:153
#define PG_GETARG_OID(n)
Definition: fmgr.h:275
#define PG_ARGISNULL(n)
Definition: fmgr.h:209
#define PG_FUNCTION_ARGS
Definition: fmgr.h:193
void InitMaterializedSRF(FunctionCallInfo fcinfo, bits32 flags)
Definition: funcapi.c:76
bool IsBinaryUpgrade
Definition: globals.c:117
int MyProcPid
Definition: globals.c:45
struct Latch * MyLatch
Definition: globals.c:59
@ PGC_SIGHUP
Definition: guc.h:71
void ProcessConfigFile(GucContext context)
HeapTuple heap_getnext(TableScanDesc sscan, ScanDirection direction)
Definition: heapam.c:1086
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
#define GETSTRUCT(TUP)
Definition: htup_details.h:653
volatile sig_atomic_t ConfigReloadPending
Definition: interrupt.c:27
void SignalHandlerForConfigReload(SIGNAL_ARGS)
Definition: interrupt.c:61
void before_shmem_exit(pg_on_exit_callback function, Datum arg)
Definition: ipc.c:337
int i
Definition: isn.c:73
void SetLatch(Latch *latch)
Definition: latch.c:633
void ResetLatch(Latch *latch)
Definition: latch.c:725
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:518
#define WL_TIMEOUT
Definition: latch.h:130
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:132
#define WL_LATCH_SET
Definition: latch.h:127
Datum pg_stat_get_subscription(PG_FUNCTION_ARGS)
Definition: launcher.c:1286
bool logicalrep_worker_launch(LogicalRepWorkerType wtype, Oid dbid, Oid subid, const char *subname, Oid userid, Oid relid, dsm_handle subworker_dsm)
Definition: launcher.c:306
#define DEFAULT_NAPTIME_PER_CYCLE
Definition: launcher.c:54
LogicalRepWorker * logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
Definition: launcher.c:249
void AtEOXact_ApplyLauncher(bool isCommit)
Definition: launcher.c:1092
void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker)
Definition: launcher.c:702
Size ApplyLauncherShmemSize(void)
Definition: launcher.c:905
bool IsLogicalLauncher(void)
Definition: launcher.c:1249
void logicalrep_worker_attach(int slot)
Definition: launcher.c:713
static void ApplyLauncherSetWorkerStartTime(Oid subid, TimestampTz start_time)
Definition: launcher.c:1039
static List * get_subscription_list(void)
Definition: launcher.c:118
static void logicalrep_launcher_onexit(int code, Datum arg)
Definition: launcher.c:813
static dsa_area * last_start_times_dsa
Definition: launcher.c:94
void ApplyLauncherMain(Datum main_arg)
Definition: launcher.c:1128
#define PG_STAT_GET_SUBSCRIPTION_COLS
int max_logical_replication_workers
Definition: launcher.c:57
void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo)
Definition: launcher.c:639
List * logicalrep_workers_find(Oid subid, bool only_running)
Definition: launcher.c:281
static int logicalrep_pa_worker_count(Oid subid)
Definition: launcher.c:878
static bool on_commit_launcher_wakeup
Definition: launcher.c:97
struct LogicalRepCtxStruct LogicalRepCtxStruct
static TimestampTz ApplyLauncherGetWorkerStartTime(Oid subid)
Definition: launcher.c:1055
void logicalrep_worker_wakeup(Oid subid, Oid relid)
Definition: launcher.c:682
void ApplyLauncherShmemInit(void)
Definition: launcher.c:960
static void logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo)
Definition: launcher.c:533
static dshash_table * last_start_times
Definition: launcher.c:95
void logicalrep_worker_stop(Oid subid, Oid relid)
Definition: launcher.c:615
LogicalRepWorker * MyLogicalRepWorker
Definition: launcher.c:61
void ApplyLauncherWakeupAtCommit(void)
Definition: launcher.c:1111
static const dshash_parameters dsh_params
Definition: launcher.c:86
static LogicalRepCtxStruct * LogicalRepCtx
Definition: launcher.c:76
static void logicalrep_worker_onexit(int code, Datum arg)
Definition: launcher.c:824
pid_t GetLeaderApplyWorkerPid(pid_t pid)
Definition: launcher.c:1259
int max_sync_workers_per_subscription
Definition: launcher.c:58
static void logicalrep_worker_detach(void)
Definition: launcher.c:750
static bool WaitForReplicationWorkerAttach(LogicalRepWorker *worker, uint16 generation, BackgroundWorkerHandle *handle)
Definition: launcher.c:189
int logicalrep_sync_worker_count(Oid subid)
Definition: launcher.c:854
void ApplyLauncherForgetWorkerStartTime(Oid subid)
Definition: launcher.c:1081
void ApplyLauncherRegister(void)
Definition: launcher.c:924
struct LauncherLastStartTimesEntry LauncherLastStartTimesEntry
static void ApplyLauncherWakeup(void)
Definition: launcher.c:1118
static void logicalrep_launcher_attach_dshmem(void)
Definition: launcher.c:995
int max_parallel_apply_workers_per_subscription
Definition: launcher.c:59
static void logicalrep_worker_cleanup(LogicalRepWorker *worker)
Definition: launcher.c:792
Assert(fmt[strlen(fmt) - 1] !='\n')
List * lappend(List *list, void *datum)
Definition: list.c:339
void LockReleaseAll(LOCKMETHODID lockmethodid, bool allLocks)
Definition: lock.c:2130
#define DEFAULT_LOCKMETHOD
Definition: lock.h:125
#define AccessShareLock
Definition: lockdefs.h:36
bool LWLockHeldByMe(LWLock *lock)
Definition: lwlock.c:1893
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1168
bool LWLockHeldByMeInMode(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1937
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1781
@ LWTRANCHE_LAUNCHER_HASH
Definition: lwlock.h:209
@ LWTRANCHE_LAUNCHER_DSA
Definition: lwlock.h:208
@ LW_SHARED
Definition: lwlock.h:117
@ LW_EXCLUSIVE
Definition: lwlock.h:116
char * pstrdup(const char *in)
Definition: mcxt.c:1619
MemoryContext TopMemoryContext
Definition: mcxt.c:141
void * palloc0(Size size)
Definition: mcxt.c:1232
MemoryContext CurrentMemoryContext
Definition: mcxt.c:135
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:403
#define AllocSetContextCreate
Definition: memutils.h:128
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:152
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
#define InvalidPid
Definition: miscadmin.h:32
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:124
void * arg
#define MAXPGPATH
static time_t start_time
Definition: pg_ctl.c:94
#define lfirst(lc)
Definition: pg_list.h:172
#define NIL
Definition: pg_list.h:68
static Datum LSNGetDatum(XLogRecPtr X)
Definition: pg_lsn.h:28
NameData subname
FormData_pg_subscription * Form_pg_subscription
#define die(msg)
pqsigfunc pqsignal(int signo, pqsigfunc func)
#define snprintf
Definition: port.h:238
uintptr_t Datum
Definition: postgres.h:64
static Datum ObjectIdGetDatum(Oid X)
Definition: postgres.h:252
static Datum Int32GetDatum(int32 X)
Definition: postgres.h:212
#define InvalidOid
Definition: postgres_ext.h:36
unsigned int Oid
Definition: postgres_ext.h:31
void BackgroundWorkerInitializeConnection(const char *dbname, const char *username, uint32 flags)
Definition: postmaster.c:5559
void BackgroundWorkerUnblockSignals(void)
Definition: postmaster.c:5633
bool IsBackendPid(int pid)
Definition: procarray.c:3203
@ ForwardScanDirection
Definition: sdir.h:28
void shm_mq_detach(shm_mq_handle *mqh)
Definition: shm_mq.c:844
Size add_size(Size s1, Size s2)
Definition: shmem.c:494
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:388
Size mul_size(Size s1, Size s2)
Definition: shmem.c:511
int max_replication_slots
Definition: slot.c:103
Snapshot GetTransactionSnapshot(void)
Definition: snapmgr.c:223
#define SpinLockInit(lock)
Definition: spin.h:60
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
PGPROC * MyProc
Definition: proc.c:67
char bgw_function_name[BGW_MAXLEN]
Definition: bgworker.h:97
Datum bgw_main_arg
Definition: bgworker.h:98
char bgw_name[BGW_MAXLEN]
Definition: bgworker.h:91
int bgw_restart_time
Definition: bgworker.h:95
char bgw_type[BGW_MAXLEN]
Definition: bgworker.h:92
BgWorkerStartTime bgw_start_time
Definition: bgworker.h:94
char bgw_extra[BGW_EXTRALEN]
Definition: bgworker.h:99
pid_t bgw_notify_pid
Definition: bgworker.h:100
char bgw_library_name[MAXPGPATH]
Definition: bgworker.h:96
TimestampTz last_start_time
Definition: launcher.c:82
Definition: pg_list.h:54
dsa_handle last_start_dsa
Definition: launcher.c:69
dshash_table_handle last_start_dsh
Definition: launcher.c:70
LogicalRepWorker workers[FLEXIBLE_ARRAY_MEMBER]
Definition: launcher.c:73
XLogRecPtr relstate_lsn
TimestampTz last_recv_time
LogicalRepWorkerType type
TimestampTz launch_time
TimestampTz reply_time
FileSet * stream_fileset
XLogRecPtr reply_lsn
XLogRecPtr last_lsn
TimestampTz last_send_time
int pid
Definition: proc.h:186
Latch procLatch
Definition: proc.h:170
shm_mq_handle * error_mq_handle
ParallelApplyWorkerShared * shared
TupleDesc setDesc
Definition: execnodes.h:340
Tuplestorestate * setResult
Definition: execnodes.h:339
Definition: dsa.c:368
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:126
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:40
TableScanDesc table_beginscan_catalog(Relation relation, int nkeys, struct ScanKeyData *key)
Definition: tableam.c:112
static void table_endscan(TableScanDesc scan)
Definition: tableam.h:1009
void tuplestore_putvalues(Tuplestorestate *state, TupleDesc tdesc, const Datum *values, const bool *isnull)
Definition: tuplestore.c:750
static Datum TimestampTzGetDatum(TimestampTz X)
Definition: timestamp.h:52
int wal_receiver_timeout
Definition: walreceiver.c:91
#define walrcv_disconnect(conn)
Definition: walreceiver.h:464
#define SIGHUP
Definition: win32_port.h:168
#define kill(pid, sig)
Definition: win32_port.h:485
#define SIGUSR1
Definition: win32_port.h:180
#define isParallelApplyWorker(worker)
LogicalRepWorkerType
@ WORKERTYPE_TABLESYNC
@ WORKERTYPE_UNKNOWN
@ WORKERTYPE_PARALLEL_APPLY
@ WORKERTYPE_APPLY
#define isTablesyncWorker(worker)
static bool am_leader_apply_worker(void)
void StartTransactionCommand(void)
Definition: xact.c:2953
void CommitTransactionCommand(void)
Definition: xact.c:3050
int wal_retrieve_retry_interval
Definition: xlog.c:138
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28