This graph shows which files directly or indirectly include this file:

Functions
void	ApplyLauncherRegister (void)

void	ApplyLauncherMain (Datum main_arg)

Size	ApplyLauncherShmemSize (void)

void	ApplyLauncherShmemInit (void)

void	ApplyLauncherWakeupAtCommit (void)

void	AtEOXact_ApplyLauncher (bool isCommit)

bool	IsLogicalLauncher (void)

Variables
PGDLLIMPORT int	max_logical_replication_workers

PGDLLIMPORT int	max_sync_workers_per_subscription

Function Documentation

◆ ApplyLauncherMain()

void ApplyLauncherMain ( Datum main_arg )

Definition at line 802 of file launcher.c.

 {
     TimestampTz last_start_time = 0;
  
     ereport(DEBUG1,
             (errmsg_internal("logical replication launcher started")));
  
     before_shmem_exit(logicalrep_launcher_onexit, (Datum) 0);
  
     Assert(LogicalRepCtx->launcher_pid == 0);
     LogicalRepCtx->launcher_pid = MyProcPid;
  
     /* Establish signal handlers. */
     pqsignal(SIGHUP, SignalHandlerForConfigReload);
     pqsignal(SIGTERM, die);
     BackgroundWorkerUnblockSignals();
  
     /*
      * Establish connection to nailed catalogs (we only ever access
      * pg_subscription).
      */
     BackgroundWorkerInitializeConnection(NULL, NULL, 0);
  
     /* Enter main loop */
     for (;;)
     {
         int         rc;
         List       *sublist;
         ListCell   *lc;
         MemoryContext subctx;
         MemoryContext oldctx;
         TimestampTz now;
         long        wait_time = DEFAULT_NAPTIME_PER_CYCLE;
  
         CHECK_FOR_INTERRUPTS();
  
         now = GetCurrentTimestamp();
  
         /* Limit the start retry to once a wal_retrieve_retry_interval */
         if (TimestampDifferenceExceeds(last_start_time, now,
                                        wal_retrieve_retry_interval))
         {
             /* Use temporary context for the database list and worker info. */
             subctx = AllocSetContextCreate(TopMemoryContext,
                                            "Logical Replication Launcher sublist",
                                            ALLOCSET_DEFAULT_SIZES);
             oldctx = MemoryContextSwitchTo(subctx);
  
             /* search for subscriptions to start or stop. */
             sublist = get_subscription_list();
  
             /* Start the missing workers for enabled subscriptions. */
             foreach(lc, sublist)
             {
                 Subscription *sub = (Subscription *) lfirst(lc);
                 LogicalRepWorker *w;
  
                 if (!sub->enabled)
                     continue;
  
                 LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
                 w = logicalrep_worker_find(sub->oid, InvalidOid, false);
                 LWLockRelease(LogicalRepWorkerLock);
  
                 if (w == NULL)
                 {
                     last_start_time = now;
                     wait_time = wal_retrieve_retry_interval;
  
                     logicalrep_worker_launch(sub->dbid, sub->oid, sub->name,
                                              sub->owner, InvalidOid);
                 }
             }
  
             /* Switch back to original memory context. */
             MemoryContextSwitchTo(oldctx);
             /* Clean the temporary memory. */
             MemoryContextDelete(subctx);
         }
         else
         {
             /*
              * The wait in previous cycle was interrupted in less than
              * wal_retrieve_retry_interval since last worker was started, this
              * usually means crash of the worker, so we should retry in
              * wal_retrieve_retry_interval again.
              */
             wait_time = wal_retrieve_retry_interval;
         }
  
         /* Wait for more work. */
         rc = WaitLatch(MyLatch,
                        WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
                        wait_time,
                        WAIT_EVENT_LOGICAL_LAUNCHER_MAIN);
  
         if (rc & WL_LATCH_SET)
         {
             ResetLatch(MyLatch);
             CHECK_FOR_INTERRUPTS();
         }
  
         if (ConfigReloadPending)
         {
             ConfigReloadPending = false;
             ProcessConfigFile(PGC_SIGHUP);
         }
     }
  
     /* Not reachable */
 }

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, Assert(), BackgroundWorkerInitializeConnection(), BackgroundWorkerUnblockSignals(), before_shmem_exit(), CHECK_FOR_INTERRUPTS, ConfigReloadPending, Subscription::dbid, DEBUG1, DEFAULT_NAPTIME_PER_CYCLE, die, Subscription::enabled, ereport, errmsg_internal(), get_subscription_list(), GetCurrentTimestamp(), InvalidOid, LogicalRepCtxStruct::launcher_pid, lfirst, logicalrep_launcher_onexit(), logicalrep_worker_find(), logicalrep_worker_launch(), LogicalRepCtx, LW_SHARED, LWLockAcquire(), LWLockRelease(), MemoryContextDelete(), MemoryContextSwitchTo(), MyLatch, MyProcPid, Subscription::name, now(), Subscription::oid, Subscription::owner, PGC_SIGHUP, pqsignal(), ProcessConfigFile(), ResetLatch(), SIGHUP, SignalHandlerForConfigReload(), TimestampDifferenceExceeds(), TopMemoryContext, WAIT_EVENT_LOGICAL_LAUNCHER_MAIN, WaitLatch(), wal_retrieve_retry_interval, WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

◆ ApplyLauncherRegister()

void ApplyLauncherRegister ( void )

Definition at line 707 of file launcher.c.

 {
     BackgroundWorker bgw;
  
     if (max_logical_replication_workers == 0)
         return;
  
     memset(&bgw, 0, sizeof(bgw));
     bgw.bgw_flags = BGWORKER_SHMEM_ACCESS |
         BGWORKER_BACKEND_DATABASE_CONNECTION;
     bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
     snprintf(bgw.bgw_library_name, BGW_MAXLEN, "postgres");
     snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyLauncherMain");
     snprintf(bgw.bgw_name, BGW_MAXLEN,
              "logical replication launcher");
     snprintf(bgw.bgw_type, BGW_MAXLEN,
              "logical replication launcher");
     bgw.bgw_restart_time = 5;
     bgw.bgw_notify_pid = 0;
     bgw.bgw_main_arg = (Datum) 0;
  
     RegisterBackgroundWorker(&bgw);
 }

References BackgroundWorker::bgw_flags, BackgroundWorker::bgw_function_name, BackgroundWorker::bgw_library_name, BackgroundWorker::bgw_main_arg, BGW_MAXLEN, BackgroundWorker::bgw_name, BackgroundWorker::bgw_notify_pid, BackgroundWorker::bgw_restart_time, BackgroundWorker::bgw_start_time, BackgroundWorker::bgw_type, BGWORKER_BACKEND_DATABASE_CONNECTION, BGWORKER_SHMEM_ACCESS, BgWorkerStart_RecoveryFinished, max_logical_replication_workers, RegisterBackgroundWorker(), and snprintf.

Referenced by PostmasterMain().

◆ ApplyLauncherShmemInit()

void ApplyLauncherShmemInit ( void )

Definition at line 736 of file launcher.c.

 {
     bool        found;
  
     LogicalRepCtx = (LogicalRepCtxStruct *)
         ShmemInitStruct("Logical Replication Launcher Data",
                         ApplyLauncherShmemSize(),
                         &found);
  
     if (!found)
     {
         int         slot;
  
         memset(LogicalRepCtx, 0, ApplyLauncherShmemSize());
  
         /* Initialize memory and spin locks for each worker slot. */
         for (slot = 0; slot < max_logical_replication_workers; slot++)
         {
             LogicalRepWorker *worker = &LogicalRepCtx->workers[slot];
  
             memset(worker, 0, sizeof(LogicalRepWorker));
             SpinLockInit(&worker->relmutex);
         }
     }
 }

References ApplyLauncherShmemSize(), LogicalRepCtx, max_logical_replication_workers, LogicalRepWorker::relmutex, ShmemInitStruct(), SpinLockInit, and LogicalRepCtxStruct::workers.

Referenced by CreateSharedMemoryAndSemaphores().

◆ ApplyLauncherShmemSize()

Size ApplyLauncherShmemSize ( void )

Definition at line 688 of file launcher.c.

 {
     Size        size;
  
     /*
      * Need the fixed struct and the array of LogicalRepWorker.
      */
     size = sizeof(LogicalRepCtxStruct);
     size = MAXALIGN(size);
     size = add_size(size, mul_size(max_logical_replication_workers,
                                    sizeof(LogicalRepWorker)));
     return size;
 }

References add_size(), max_logical_replication_workers, MAXALIGN, and mul_size().

Referenced by ApplyLauncherShmemInit(), and CalculateShmemSize().

◆ ApplyLauncherWakeupAtCommit()

void ApplyLauncherWakeupAtCommit ( void )

Definition at line 785 of file launcher.c.

 {
     if (!on_commit_launcher_wakeup)
         on_commit_launcher_wakeup = true;
 }

References on_commit_launcher_wakeup.

Referenced by AlterSubscription(), AlterSubscriptionOwner_internal(), and CreateSubscription().

◆ AtEOXact_ApplyLauncher()

void AtEOXact_ApplyLauncher ( bool isCommit )

Definition at line 766 of file launcher.c.

 {
     if (isCommit)
     {
         if (on_commit_launcher_wakeup)
             ApplyLauncherWakeup();
     }
  
     on_commit_launcher_wakeup = false;
 }

References ApplyLauncherWakeup(), and on_commit_launcher_wakeup.

Referenced by AbortTransaction(), and CommitTransaction().

◆ IsLogicalLauncher()

bool IsLogicalLauncher ( void )

Definition at line 918 of file launcher.c.

 {
     return LogicalRepCtx->launcher_pid == MyProcPid;
 }

References LogicalRepCtxStruct::launcher_pid, LogicalRepCtx, and MyProcPid.

Referenced by ProcessInterrupts().

Variable Documentation

◆ max_logical_replication_workers

PGDLLIMPORT int max_logical_replication_workers

extern

Definition at line 56 of file launcher.c.

Referenced by ApplyLauncherRegister(), ApplyLauncherShmemInit(), ApplyLauncherShmemSize(), logicalrep_sync_worker_count(), logicalrep_worker_attach(), logicalrep_worker_find(), logicalrep_worker_launch(), logicalrep_workers_find(), and pg_stat_get_subscription().

◆ max_sync_workers_per_subscription

PGDLLIMPORT int max_sync_workers_per_subscription