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)

bool	XactManipulatesLogicalReplicationWorkers (void)

void	AtEOXact_ApplyLauncher (bool isCommit)

bool	IsLogicalLauncher (void)

Variables
int	max_logical_replication_workers

int	max_sync_workers_per_subscription

Function Documentation

void ApplyLauncherMain ( Datum main_arg )

Definition at line 873 of file launcher.c.

References ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE, ALLOCSET_DEFAULT_MINSIZE, AllocSetContextCreate(), Assert, BackgroundWorkerInitializeConnection(), BackgroundWorkerUnblockSignals(), before_shmem_exit(), CHECK_FOR_INTERRUPTS, Subscription::dbid, DEBUG1, DEFAULT_NAPTIME_PER_CYCLE, die(), Subscription::enabled, ereport, errmsg(), get_subscription_list(), GetCurrentTimestamp(), got_SIGHUP, InvalidOid, LogicalRepCtxStruct::launcher_pid, lfirst, logicalrep_launcher_onexit(), logicalrep_launcher_sighup(), logicalrep_worker_find(), logicalrep_worker_launch(), LW_SHARED, LWLockAcquire(), LWLockRelease(), MemoryContextDelete(), MemoryContextSwitchTo(), MyLatch, MyProcPid, Subscription::name, now(), NULL, Subscription::oid, Subscription::owner, PGC_SIGHUP, pqsignal(), proc_exit(), ProcessConfigFile(), ResetLatch(), SIGHUP, TimestampDifferenceExceeds(), TopMemoryContext, WAIT_EVENT_LOGICAL_LAUNCHER_MAIN, WaitLatch(), wal_retrieve_retry_interval, WL_LATCH_SET, WL_POSTMASTER_DEATH, and WL_TIMEOUT.

 {
     TimestampTz last_start_time = 0;
 
     ereport(DEBUG1,
             (errmsg("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, logicalrep_launcher_sighup);
     pqsignal(SIGTERM, die);
     BackgroundWorkerUnblockSignals();
 
     /*
      * Establish connection to nailed catalogs (we only ever access
      * pg_subscription).
      */
     BackgroundWorkerInitializeConnection(NULL, NULL);
 
     /* 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_MINSIZE,
                                            ALLOCSET_DEFAULT_INITSIZE,
                                            ALLOCSET_DEFAULT_MAXSIZE);
             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_POSTMASTER_DEATH,
                        wait_time,
                        WAIT_EVENT_LOGICAL_LAUNCHER_MAIN);
 
         /* emergency bailout if postmaster has died */
         if (rc & WL_POSTMASTER_DEATH)
             proc_exit(1);
 
         if (rc & WL_LATCH_SET)
         {
             ResetLatch(MyLatch);
             CHECK_FOR_INTERRUPTS();
         }
 
         if (got_SIGHUP)
         {
             got_SIGHUP = false;
             ProcessConfigFile(PGC_SIGHUP);
         }
     }
 
     /* Not reachable */
 }

void ApplyLauncherRegister ( void )

Definition at line 756 of file launcher.c.

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, BGWORKER_BACKEND_DATABASE_CONNECTION, BGWORKER_SHMEM_ACCESS, BgWorkerStart_RecoveryFinished, max_logical_replication_workers, RegisterBackgroundWorker(), and snprintf().

Referenced by PostmasterMain().

 {
     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");
     bgw.bgw_restart_time = 5;
     bgw.bgw_notify_pid = 0;
     bgw.bgw_main_arg = (Datum) 0;
 
     RegisterBackgroundWorker(&bgw);
 }

void ApplyLauncherShmemInit ( void )

Definition at line 783 of file launcher.c.

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

Referenced by CreateSharedMemoryAndSemaphores().

 {
     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);
         }
     }
 }

Size ApplyLauncherShmemSize ( void )

Definition at line 737 of file launcher.c.

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

Referenced by ApplyLauncherShmemInit(), and CreateSharedMemoryAndSemaphores().

 {
     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;
 }

void ApplyLauncherWakeupAtCommit ( void )

Definition at line 856 of file launcher.c.

References on_commit_launcher_wakeup.

Referenced by AlterSubscription(), and CreateSubscription().

 {
     if (!on_commit_launcher_wakeup)
         on_commit_launcher_wakeup = true;
 }

void AtEOXact_ApplyLauncher ( bool isCommit )

Definition at line 823 of file launcher.c.

References ApplyLauncherWakeup(), lfirst, logicalrep_worker_stop(), NIL, on_commit_launcher_wakeup, LogicalRepWorkerId::relid, and LogicalRepWorkerId::subid.

Referenced by AbortTransaction(), and CommitTransaction().

 {
     if (isCommit)
     {
         ListCell   *lc;
 
         foreach(lc, on_commit_stop_workers)
         {
             LogicalRepWorkerId *wid = lfirst(lc);
 
             logicalrep_worker_stop(wid->subid, wid->relid);
         }
 
         if (on_commit_launcher_wakeup)
             ApplyLauncherWakeup();
     }
 
     /*
      * No need to pfree on_commit_stop_workers.  It was allocated in
      * transaction memory context, which is going to be cleaned soon.
      */
     on_commit_stop_workers = NIL;
     on_commit_launcher_wakeup = false;
 }

bool IsLogicalLauncher ( void )

Definition at line 995 of file launcher.c.

References LogicalRepCtxStruct::launcher_pid, and MyProcPid.

Referenced by ProcessInterrupts().

 {
     return LogicalRepCtx->launcher_pid == MyProcPid;
 }

bool XactManipulatesLogicalReplicationWorkers ( void )

Definition at line 814 of file launcher.c.

References NIL.

Referenced by PrepareTransaction().

 {
     return (on_commit_stop_workers != NIL);
 }

Variable Documentation

int max_logical_replication_workers

Definition at line 60 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().

int max_sync_workers_per_subscription

Definition at line 61 of file launcher.c.

Referenced by logicalrep_worker_launch(), and process_syncing_tables_for_apply().

Functions

Variables

Function Documentation

Variable Documentation