applyparallelworker.c File Reference

#include "postgres.h"
#include "libpq/pqformat.h"
#include "libpq/pqmq.h"
#include "pgstat.h"
#include "postmaster/interrupt.h"
#include "replication/logicallauncher.h"
#include "replication/logicalworker.h"
#include "replication/origin.h"
#include "replication/worker_internal.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "tcop/tcopprot.h"
#include "utils/inval.h"
#include "utils/memutils.h"
#include "utils/syscache.h"

Include dependency graph for applyparallelworker.c:

Go to the source code of this file.

Data Structures
struct	ParallelApplyWorkerEntry

Macros
#define	PG_LOGICAL_APPLY_SHM_MAGIC   0x787ca067
#define	PARALLEL_APPLY_KEY_SHARED   1
#define	PARALLEL_APPLY_KEY_MQ   2
#define	PARALLEL_APPLY_KEY_ERROR_QUEUE   3
#define	DSM_QUEUE_SIZE   (16 * 1024 * 1024)
#define	DSM_ERROR_QUEUE_SIZE   (16 * 1024)
#define	SIZE_STATS_MESSAGE   (2 * sizeof(XLogRecPtr) + sizeof(TimestampTz))
#define	PARALLEL_APPLY_LOCK_STREAM   0
#define	PARALLEL_APPLY_LOCK_XACT   1
#define	SHM_SEND_RETRY_INTERVAL_MS   1000
#define	SHM_SEND_TIMEOUT_MS   (10000 - SHM_SEND_RETRY_INTERVAL_MS)

Typedefs
typedef struct ParallelApplyWorkerEntry	ParallelApplyWorkerEntry

Functions
static void	pa_free_worker_info (ParallelApplyWorkerInfo *winfo)
static ParallelTransState	pa_get_xact_state (ParallelApplyWorkerShared *wshared)
static PartialFileSetState	pa_get_fileset_state (void)
static bool	pa_can_start (void)
static bool	pa_setup_dsm (ParallelApplyWorkerInfo *winfo)
static ParallelApplyWorkerInfo *	pa_launch_parallel_worker (void)
void	pa_allocate_worker (TransactionId xid)
ParallelApplyWorkerInfo *	pa_find_worker (TransactionId xid)
static void	pa_free_worker (ParallelApplyWorkerInfo *winfo)
void	pa_detach_all_error_mq (void)
static bool	pa_has_spooled_message_pending ()
static bool	pa_process_spooled_messages_if_required (void)
static void	ProcessParallelApplyInterrupts (void)
static void	LogicalParallelApplyLoop (shm_mq_handle *mqh)
static void	pa_shutdown (int code, Datum arg)
void	ParallelApplyWorkerMain (Datum main_arg)
void	HandleParallelApplyMessageInterrupt (void)
static void	HandleParallelApplyMessage (StringInfo msg)
void	HandleParallelApplyMessages (void)
bool	pa_send_data (ParallelApplyWorkerInfo *winfo, Size nbytes, const void *data)
void	pa_switch_to_partial_serialize (ParallelApplyWorkerInfo *winfo, bool stream_locked)
static void	pa_wait_for_xact_state (ParallelApplyWorkerInfo *winfo, ParallelTransState xact_state)
static void	pa_wait_for_xact_finish (ParallelApplyWorkerInfo *winfo)
void	pa_set_xact_state (ParallelApplyWorkerShared *wshared, ParallelTransState xact_state)
void	pa_set_stream_apply_worker (ParallelApplyWorkerInfo *winfo)
static void	pa_savepoint_name (Oid suboid, TransactionId xid, char *spname, Size szsp)
void	pa_start_subtrans (TransactionId current_xid, TransactionId top_xid)
void	pa_reset_subtrans (void)
void	pa_stream_abort (LogicalRepStreamAbortData *abort_data)
void	pa_set_fileset_state (ParallelApplyWorkerShared *wshared, PartialFileSetState fileset_state)
void	pa_lock_stream (TransactionId xid, LOCKMODE lockmode)
void	pa_unlock_stream (TransactionId xid, LOCKMODE lockmode)
void	pa_lock_transaction (TransactionId xid, LOCKMODE lockmode)
void	pa_unlock_transaction (TransactionId xid, LOCKMODE lockmode)
void	pa_decr_and_wait_stream_block (void)
void	pa_xact_finish (ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)

Variables
static HTAB *	ParallelApplyTxnHash = NULL
static List *	ParallelApplyWorkerPool = NIL
ParallelApplyWorkerShared *	MyParallelShared = NULL
volatile sig_atomic_t	ParallelApplyMessagePending = false
static ParallelApplyWorkerInfo *	stream_apply_worker = NULL
static List *	subxactlist = NIL

Macro Definition Documentation

DSM_ERROR_QUEUE_SIZE

#define DSM_ERROR_QUEUE_SIZE   (16 * 1024)

Definition at line 195 of file applyparallelworker.c.

DSM_QUEUE_SIZE

#define DSM_QUEUE_SIZE   (16 * 1024 * 1024)

Definition at line 187 of file applyparallelworker.c.

PARALLEL_APPLY_KEY_ERROR_QUEUE

#define PARALLEL_APPLY_KEY_ERROR_QUEUE   3

Definition at line 184 of file applyparallelworker.c.

PARALLEL_APPLY_KEY_MQ

#define PARALLEL_APPLY_KEY_MQ   2

Definition at line 183 of file applyparallelworker.c.

PARALLEL_APPLY_KEY_SHARED

#define PARALLEL_APPLY_KEY_SHARED   1

Definition at line 182 of file applyparallelworker.c.

PARALLEL_APPLY_LOCK_STREAM

#define PARALLEL_APPLY_LOCK_STREAM   0

Definition at line 209 of file applyparallelworker.c.

PARALLEL_APPLY_LOCK_XACT

#define PARALLEL_APPLY_LOCK_XACT   1

Definition at line 210 of file applyparallelworker.c.

PG_LOGICAL_APPLY_SHM_MAGIC

#define PG_LOGICAL_APPLY_SHM_MAGIC   0x787ca067

Definition at line 175 of file applyparallelworker.c.

SHM_SEND_RETRY_INTERVAL_MS

#define SHM_SEND_RETRY_INTERVAL_MS   1000

SHM_SEND_TIMEOUT_MS

#define SHM_SEND_TIMEOUT_MS   (10000 - SHM_SEND_RETRY_INTERVAL_MS)

SIZE_STATS_MESSAGE

#define SIZE_STATS_MESSAGE   (2 * sizeof(XLogRecPtr) + sizeof(TimestampTz))

Definition at line 203 of file applyparallelworker.c.

Typedef Documentation

ParallelApplyWorkerEntry

typedef struct ParallelApplyWorkerEntry ParallelApplyWorkerEntry

Function Documentation

HandleParallelApplyMessage()

static void HandleParallelApplyMessage ( StringInfo  msg )

static

Definition at line 1001 of file applyparallelworker.c.

{
    char        msgtype;
 
    msgtype = pq_getmsgbyte(msg);
 
    switch (msgtype)
    {
        case 'E':               /* ErrorResponse */
            {
                ErrorData   edata;
 
                /* Parse ErrorResponse. */
                pq_parse_errornotice(msg, &edata);
 
                /*
                 * If desired, add a context line to show that this is a
                 * message propagated from a parallel apply worker. Otherwise,
                 * it can sometimes be confusing to understand what actually
                 * happened.
                 */
                if (edata.context)
                    edata.context = psprintf("%s\n%s", edata.context,
                                             _("logical replication parallel apply worker"));
                else
                    edata.context = pstrdup(_("logical replication parallel apply worker"));
 
                /*
                 * Context beyond that should use the error context callbacks
                 * that were in effect in LogicalRepApplyLoop().
                 */
                error_context_stack = apply_error_context_stack;
 
                /*
                 * The actual error must have been reported by the parallel
                 * apply worker.
                 */
                ereport(ERROR,
                        (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                         errmsg("logical replication parallel apply worker exited due to error"),
                         errcontext("%s", edata.context)));
            }
 
            /*
             * Don't need to do anything about NoticeResponse and
             * NotifyResponse as the logical replication worker doesn't need
             * to send messages to the client.
             */
        case 'N':
        case 'A':
            break;
 
        default:
            elog(ERROR, "unrecognized message type received from logical replication parallel apply worker: %c (message length %d bytes)",
                 msgtype, msg->len);
    }
}

References _, apply_error_context_stack, ErrorData::context, elog, ereport, errcode(), errcontext, errmsg(), ERROR, error_context_stack, StringInfoData::len, pq_getmsgbyte(), pq_parse_errornotice(), psprintf(), and pstrdup().

Referenced by HandleParallelApplyMessages().

HandleParallelApplyMessageInterrupt()

void HandleParallelApplyMessageInterrupt

(

void

)

Definition at line 989 of file applyparallelworker.c.

{
    InterruptPending = true;
    ParallelApplyMessagePending = true;
    SetLatch(MyLatch);
}

References InterruptPending, MyLatch, ParallelApplyMessagePending, and SetLatch().

Referenced by procsignal_sigusr1_handler().

HandleParallelApplyMessages()

void HandleParallelApplyMessages

(

void

)

Definition at line 1063 of file applyparallelworker.c.

{
    ListCell   *lc;
    MemoryContext oldcontext;
 
    static MemoryContext hpam_context = NULL;
 
    /*
     * This is invoked from ProcessInterrupts(), and since some of the
     * functions it calls contain CHECK_FOR_INTERRUPTS(), there is a potential
     * for recursive calls if more signals are received while this runs. It's
     * unclear that recursive entry would be safe, and it doesn't seem useful
     * even if it is safe, so let's block interrupts until done.
     */
    HOLD_INTERRUPTS();
 
    /*
     * Moreover, CurrentMemoryContext might be pointing almost anywhere. We
     * don't want to risk leaking data into long-lived contexts, so let's do
     * our work here in a private context that we can reset on each use.
     */
    if (!hpam_context)          /* first time through? */
        hpam_context = AllocSetContextCreate(TopMemoryContext,
                                             "HandleParallelApplyMessages",
                                             ALLOCSET_DEFAULT_SIZES);
    else
        MemoryContextReset(hpam_context);
 
    oldcontext = MemoryContextSwitchTo(hpam_context);
 
    ParallelApplyMessagePending = false;
 
    foreach(lc, ParallelApplyWorkerPool)
    {
        shm_mq_result res;
        Size        nbytes;
        void       *data;
        ParallelApplyWorkerInfo *winfo = (ParallelApplyWorkerInfo *) lfirst(lc);
 
        /*
         * The leader will detach from the error queue and set it to NULL
         * before preparing to stop all parallel apply workers, so we don't
         * need to handle error messages anymore. See
         * logicalrep_worker_detach.
         */
        if (!winfo->error_mq_handle)
            continue;
 
        res = shm_mq_receive(winfo->error_mq_handle, &nbytes, &data, true);
 
        if (res == SHM_MQ_WOULD_BLOCK)
            continue;
        else if (res == SHM_MQ_SUCCESS)
        {
            StringInfoData msg;
 
            initStringInfo(&msg);
            appendBinaryStringInfo(&msg, data, nbytes);
            HandleParallelApplyMessage(&msg);
            pfree(msg.data);
        }
        else
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("lost connection to the logical replication parallel apply worker")));
    }
 
    MemoryContextSwitchTo(oldcontext);
 
    /* Might as well clear the context on our way out */
    MemoryContextReset(hpam_context);
 
    RESUME_INTERRUPTS();
}

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, appendBinaryStringInfo(), StringInfoData::data, data, ereport, errcode(), errmsg(), ERROR, ParallelApplyWorkerInfo::error_mq_handle, HandleParallelApplyMessage(), HOLD_INTERRUPTS, initStringInfo(), lfirst, MemoryContextReset(), MemoryContextSwitchTo(), ParallelApplyMessagePending, ParallelApplyWorkerPool, pfree(), res, RESUME_INTERRUPTS, shm_mq_receive(), SHM_MQ_SUCCESS, SHM_MQ_WOULD_BLOCK, and TopMemoryContext.

Referenced by ProcessInterrupts().

LogicalParallelApplyLoop()

static void LogicalParallelApplyLoop ( shm_mq_handle *  mqh )

static

Definition at line 734 of file applyparallelworker.c.

{
    shm_mq_result shmq_res;
    ErrorContextCallback errcallback;
    MemoryContext oldcxt = CurrentMemoryContext;
 
    /*
     * Init the ApplyMessageContext which we clean up after each replication
     * protocol message.
     */
    ApplyMessageContext = AllocSetContextCreate(ApplyContext,
                                                "ApplyMessageContext",
                                                ALLOCSET_DEFAULT_SIZES);
 
    /*
     * Push apply error context callback. Fields will be filled while applying
     * a change.
     */
    errcallback.callback = apply_error_callback;
    errcallback.previous = error_context_stack;
    error_context_stack = &errcallback;
 
    for (;;)
    {
        void       *data;
        Size        len;
 
        ProcessParallelApplyInterrupts();
 
        /* Ensure we are reading the data into our memory context. */
        MemoryContextSwitchTo(ApplyMessageContext);
 
        shmq_res = shm_mq_receive(mqh, &len, &data, true);
 
        if (shmq_res == SHM_MQ_SUCCESS)
        {
            StringInfoData s;
            int         c;
 
            if (len == 0)
                elog(ERROR, "invalid message length");
 
            initReadOnlyStringInfo(&s, data, len);
 
            /*
             * The first byte of messages sent from leader apply worker to
             * parallel apply workers can only be 'w'.
             */
            c = pq_getmsgbyte(&s);
            if (c != 'w')
                elog(ERROR, "unexpected message \"%c\"", c);
 
            /*
             * Ignore statistics fields that have been updated by the leader
             * apply worker.
             *
             * XXX We can avoid sending the statistics fields from the leader
             * apply worker but for that, it needs to rebuild the entire
             * message by removing these fields which could be more work than
             * simply ignoring these fields in the parallel apply worker.
             */
            s.cursor += SIZE_STATS_MESSAGE;
 
            apply_dispatch(&s);
        }
        else if (shmq_res == SHM_MQ_WOULD_BLOCK)
        {
            /* Replay the changes from the file, if any. */
            if (!pa_process_spooled_messages_if_required())
            {
                int         rc;
 
                /* Wait for more work. */
                rc = WaitLatch(MyLatch,
                               WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
                               1000L,
                               WAIT_EVENT_LOGICAL_PARALLEL_APPLY_MAIN);
 
                if (rc & WL_LATCH_SET)
                    ResetLatch(MyLatch);
            }
        }
        else
        {
            Assert(shmq_res == SHM_MQ_DETACHED);
 
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("lost connection to the logical replication apply worker")));
        }
 
        MemoryContextReset(ApplyMessageContext);
        MemoryContextSwitchTo(oldcxt);
    }
 
    /* Pop the error context stack. */
    error_context_stack = errcallback.previous;
 
    MemoryContextSwitchTo(oldcxt);
}

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, apply_dispatch(), apply_error_callback(), ApplyContext, ApplyMessageContext, Assert, ErrorContextCallback::callback, CurrentMemoryContext, StringInfoData::cursor, data, elog, ereport, errcode(), errmsg(), ERROR, error_context_stack, initReadOnlyStringInfo(), len, MemoryContextReset(), MemoryContextSwitchTo(), MyLatch, pa_process_spooled_messages_if_required(), pq_getmsgbyte(), ErrorContextCallback::previous, ProcessParallelApplyInterrupts(), ResetLatch(), SHM_MQ_DETACHED, shm_mq_receive(), SHM_MQ_SUCCESS, SHM_MQ_WOULD_BLOCK, SIZE_STATS_MESSAGE, WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

Referenced by ParallelApplyWorkerMain().

pa_allocate_worker()

void pa_allocate_worker

(

TransactionId

xid

)

Definition at line 470 of file applyparallelworker.c.

{
    bool        found;
    ParallelApplyWorkerInfo *winfo = NULL;
    ParallelApplyWorkerEntry *entry;
 
    if (!pa_can_start())
        return;
 
    winfo = pa_launch_parallel_worker();
    if (!winfo)
        return;
 
    /* First time through, initialize parallel apply worker state hashtable. */
    if (!ParallelApplyTxnHash)
    {
        HASHCTL     ctl;
 
        MemSet(&ctl, 0, sizeof(ctl));
        ctl.keysize = sizeof(TransactionId);
        ctl.entrysize = sizeof(ParallelApplyWorkerEntry);
        ctl.hcxt = ApplyContext;
 
        ParallelApplyTxnHash = hash_create("logical replication parallel apply workers hash",
                                           16, &ctl,
                                           HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
    }
 
    /* Create an entry for the requested transaction. */
    entry = hash_search(ParallelApplyTxnHash, &xid, HASH_ENTER, &found);
    if (found)
        elog(ERROR, "hash table corrupted");
 
    /* Update the transaction information in shared memory. */
    SpinLockAcquire(&winfo->shared->mutex);
    winfo->shared->xact_state = PARALLEL_TRANS_UNKNOWN;
    winfo->shared->xid = xid;
    SpinLockRelease(&winfo->shared->mutex);
 
    winfo->in_use = true;
    winfo->serialize_changes = false;
    entry->winfo = winfo;
}

References ApplyContext, ctl, elog, ERROR, HASH_BLOBS, HASH_CONTEXT, hash_create(), HASH_ELEM, HASH_ENTER, hash_search(), ParallelApplyWorkerInfo::in_use, MemSet, ParallelApplyWorkerShared::mutex, pa_can_start(), pa_launch_parallel_worker(), PARALLEL_TRANS_UNKNOWN, ParallelApplyTxnHash, ParallelApplyWorkerInfo::serialize_changes, ParallelApplyWorkerInfo::shared, SpinLockAcquire, SpinLockRelease, ParallelApplyWorkerEntry::winfo, ParallelApplyWorkerShared::xact_state, and ParallelApplyWorkerShared::xid.

Referenced by apply_handle_stream_start().

pa_can_start()

static bool pa_can_start ( void  )

static

Definition at line 265 of file applyparallelworker.c.

{
    /* Only leader apply workers can start parallel apply workers. */
    if (!am_leader_apply_worker())
        return false;
 
    /*
     * It is good to check for any change in the subscription parameter to
     * avoid the case where for a very long time the change doesn't get
     * reflected. This can happen when there is a constant flow of streaming
     * transactions that are handled by parallel apply workers.
     *
     * It is better to do it before the below checks so that the latest values
     * of subscription can be used for the checks.
     */
    maybe_reread_subscription();
 
    /*
     * Don't start a new parallel apply worker if the subscription is not
     * using parallel streaming mode, or if the publisher does not support
     * parallel apply.
     */
    if (!MyLogicalRepWorker->parallel_apply)
        return false;
 
    /*
     * Don't start a new parallel worker if user has set skiplsn as it's
     * possible that they want to skip the streaming transaction. For
     * streaming transactions, we need to serialize the transaction to a file
     * so that we can get the last LSN of the transaction to judge whether to
     * skip before starting to apply the change.
     *
     * One might think that we could allow parallelism if the first lsn of the
     * transaction is greater than skiplsn, but we don't send it with the
     * STREAM START message, and it doesn't seem worth sending the extra eight
     * bytes with the STREAM START to enable parallelism for this case.
     */
    if (!XLogRecPtrIsInvalid(MySubscription->skiplsn))
        return false;
 
    /*
     * For streaming transactions that are being applied using a parallel
     * apply worker, we cannot decide whether to apply the change for a
     * relation that is not in the READY state (see
     * should_apply_changes_for_rel) as we won't know remote_final_lsn by that
     * time. So, we don't start the new parallel apply worker in this case.
     */
    if (!AllTablesyncsReady())
        return false;
 
    return true;
}

References AllTablesyncsReady(), am_leader_apply_worker(), maybe_reread_subscription(), MyLogicalRepWorker, MySubscription, LogicalRepWorker::parallel_apply, Subscription::skiplsn, and XLogRecPtrIsInvalid.

Referenced by pa_allocate_worker().

pa_decr_and_wait_stream_block()

void pa_decr_and_wait_stream_block

(

void

)

Definition at line 1591 of file applyparallelworker.c.

{
    Assert(am_parallel_apply_worker());
 
    /*
     * It is only possible to not have any pending stream chunks when we are
     * applying spooled messages.
     */
    if (pg_atomic_read_u32(&MyParallelShared->pending_stream_count) == 0)
    {
        if (pa_has_spooled_message_pending())
            return;
 
        elog(ERROR, "invalid pending streaming chunk 0");
    }
 
    if (pg_atomic_sub_fetch_u32(&MyParallelShared->pending_stream_count, 1) == 0)
    {
        pa_lock_stream(MyParallelShared->xid, AccessShareLock);
        pa_unlock_stream(MyParallelShared->xid, AccessShareLock);
    }
}

References AccessShareLock, am_parallel_apply_worker(), Assert, elog, ERROR, MyParallelShared, pa_has_spooled_message_pending(), pa_lock_stream(), pa_unlock_stream(), ParallelApplyWorkerShared::pending_stream_count, pg_atomic_read_u32(), pg_atomic_sub_fetch_u32(), and ParallelApplyWorkerShared::xid.

Referenced by apply_handle_stream_abort(), and apply_handle_stream_stop().

pa_detach_all_error_mq()

void pa_detach_all_error_mq

(

void

)

Definition at line 622 of file applyparallelworker.c.

{
    ListCell   *lc;
 
    foreach(lc, ParallelApplyWorkerPool)
    {
        ParallelApplyWorkerInfo *winfo = (ParallelApplyWorkerInfo *) lfirst(lc);
 
        if (winfo->error_mq_handle)
        {
            shm_mq_detach(winfo->error_mq_handle);
            winfo->error_mq_handle = NULL;
        }
    }
}

References ParallelApplyWorkerInfo::error_mq_handle, lfirst, ParallelApplyWorkerPool, and shm_mq_detach().

Referenced by logicalrep_worker_detach().

pa_find_worker()

ParallelApplyWorkerInfo* pa_find_worker

(

TransactionId

xid

)

Definition at line 518 of file applyparallelworker.c.

{
    bool        found;
    ParallelApplyWorkerEntry *entry;
 
    if (!TransactionIdIsValid(xid))
        return NULL;
 
    if (!ParallelApplyTxnHash)
        return NULL;
 
    /* Return the cached parallel apply worker if valid. */
    if (stream_apply_worker)
        return stream_apply_worker;
 
    /* Find an entry for the requested transaction. */
    entry = hash_search(ParallelApplyTxnHash, &xid, HASH_FIND, &found);
    if (found)
    {
        /* The worker must not have exited.  */
        Assert(entry->winfo->in_use);
        return entry->winfo;
    }
 
    return NULL;
}

References Assert, HASH_FIND, hash_search(), ParallelApplyWorkerInfo::in_use, ParallelApplyTxnHash, stream_apply_worker, TransactionIdIsValid, and ParallelApplyWorkerEntry::winfo.

Referenced by get_transaction_apply_action().

pa_free_worker()

static void pa_free_worker ( ParallelApplyWorkerInfo *  winfo )

static

Definition at line 556 of file applyparallelworker.c.

{
    Assert(!am_parallel_apply_worker());
    Assert(winfo->in_use);
    Assert(pa_get_xact_state(winfo->shared) == PARALLEL_TRANS_FINISHED);
 
    if (!hash_search(ParallelApplyTxnHash, &winfo->shared->xid, HASH_REMOVE, NULL))
        elog(ERROR, "hash table corrupted");
 
    /*
     * Stop the worker if there are enough workers in the pool.
     *
     * XXX Additionally, we also stop the worker if the leader apply worker
     * serialize part of the transaction data due to a send timeout. This is
     * because the message could be partially written to the queue and there
     * is no way to clean the queue other than resending the message until it
     * succeeds. Instead of trying to send the data which anyway would have
     * been serialized and then letting the parallel apply worker deal with
     * the spurious message, we stop the worker.
     */
    if (winfo->serialize_changes ||
        list_length(ParallelApplyWorkerPool) >
        (max_parallel_apply_workers_per_subscription / 2))
    {
        logicalrep_pa_worker_stop(winfo);
        pa_free_worker_info(winfo);
 
        return;
    }
 
    winfo->in_use = false;
    winfo->serialize_changes = false;
}

References am_parallel_apply_worker(), Assert, elog, ERROR, HASH_REMOVE, hash_search(), ParallelApplyWorkerInfo::in_use, list_length(), logicalrep_pa_worker_stop(), max_parallel_apply_workers_per_subscription, pa_free_worker_info(), pa_get_xact_state(), PARALLEL_TRANS_FINISHED, ParallelApplyTxnHash, ParallelApplyWorkerPool, ParallelApplyWorkerInfo::serialize_changes, ParallelApplyWorkerInfo::shared, and ParallelApplyWorkerShared::xid.

Referenced by pa_xact_finish().

pa_free_worker_info()

static void pa_free_worker_info ( ParallelApplyWorkerInfo *  winfo )

static

Definition at line 595 of file applyparallelworker.c.

{
    Assert(winfo);
 
    if (winfo->mq_handle)
        shm_mq_detach(winfo->mq_handle);
 
    if (winfo->error_mq_handle)
        shm_mq_detach(winfo->error_mq_handle);
 
    /* Unlink the files with serialized changes. */
    if (winfo->serialize_changes)
        stream_cleanup_files(MyLogicalRepWorker->subid, winfo->shared->xid);
 
    if (winfo->dsm_seg)
        dsm_detach(winfo->dsm_seg);
 
    /* Remove from the worker pool. */
    ParallelApplyWorkerPool = list_delete_ptr(ParallelApplyWorkerPool, winfo);
 
    pfree(winfo);
}

References Assert, dsm_detach(), ParallelApplyWorkerInfo::dsm_seg, ParallelApplyWorkerInfo::error_mq_handle, list_delete_ptr(), ParallelApplyWorkerInfo::mq_handle, MyLogicalRepWorker, ParallelApplyWorkerPool, pfree(), ParallelApplyWorkerInfo::serialize_changes, ParallelApplyWorkerInfo::shared, shm_mq_detach(), stream_cleanup_files(), LogicalRepWorker::subid, and ParallelApplyWorkerShared::xid.

Referenced by pa_free_worker(), and pa_launch_parallel_worker().

pa_get_fileset_state()

static PartialFileSetState pa_get_fileset_state ( void  )

static

Definition at line 1518 of file applyparallelworker.c.

{
    PartialFileSetState fileset_state;
 
    Assert(am_parallel_apply_worker());
 
    SpinLockAcquire(&MyParallelShared->mutex);
    fileset_state = MyParallelShared->fileset_state;
    SpinLockRelease(&MyParallelShared->mutex);
 
    return fileset_state;
}

References am_parallel_apply_worker(), Assert, ParallelApplyWorkerShared::fileset_state, ParallelApplyWorkerShared::mutex, MyParallelShared, SpinLockAcquire, and SpinLockRelease.

Referenced by pa_has_spooled_message_pending(), and pa_process_spooled_messages_if_required().

pa_get_xact_state()

static ParallelTransState pa_get_xact_state ( ParallelApplyWorkerShared *  wshared )

static

Definition at line 1319 of file applyparallelworker.c.

{
    ParallelTransState xact_state;
 
    SpinLockAcquire(&wshared->mutex);
    xact_state = wshared->xact_state;
    SpinLockRelease(&wshared->mutex);
 
    return xact_state;
}

References ParallelApplyWorkerShared::mutex, SpinLockAcquire, SpinLockRelease, and ParallelApplyWorkerShared::xact_state.

Referenced by pa_free_worker(), pa_wait_for_xact_finish(), and pa_wait_for_xact_state().

pa_has_spooled_message_pending()

static bool pa_has_spooled_message_pending ( )

static

Definition at line 642 of file applyparallelworker.c.

{
    PartialFileSetState fileset_state;
 
    fileset_state = pa_get_fileset_state();
 
    return (fileset_state != FS_EMPTY);
}

References FS_EMPTY, and pa_get_fileset_state().

Referenced by pa_decr_and_wait_stream_block().

pa_launch_parallel_worker()

static ParallelApplyWorkerInfo* pa_launch_parallel_worker ( void  )

static

Definition at line 404 of file applyparallelworker.c.

{
    MemoryContext oldcontext;
    bool        launched;
    ParallelApplyWorkerInfo *winfo;
    ListCell   *lc;
 
    /* Try to get an available parallel apply worker from the worker pool. */
    foreach(lc, ParallelApplyWorkerPool)
    {
        winfo = (ParallelApplyWorkerInfo *) lfirst(lc);
 
        if (!winfo->in_use)
            return winfo;
    }
 
    /*
     * Start a new parallel apply worker.
     *
     * The worker info can be used for the lifetime of the worker process, so
     * create it in a permanent context.
     */
    oldcontext = MemoryContextSwitchTo(ApplyContext);
 
    winfo = (ParallelApplyWorkerInfo *) palloc0(sizeof(ParallelApplyWorkerInfo));
 
    /* Setup shared memory. */
    if (!pa_setup_dsm(winfo))
    {
        MemoryContextSwitchTo(oldcontext);
        pfree(winfo);
        return NULL;
    }
 
    launched = logicalrep_worker_launch(WORKERTYPE_PARALLEL_APPLY,
                                        MyLogicalRepWorker->dbid,
                                        MySubscription->oid,
                                        MySubscription->name,
                                        MyLogicalRepWorker->userid,
                                        InvalidOid,
                                        dsm_segment_handle(winfo->dsm_seg));
 
    if (launched)
    {
        ParallelApplyWorkerPool = lappend(ParallelApplyWorkerPool, winfo);
    }
    else
    {
        pa_free_worker_info(winfo);
        winfo = NULL;
    }
 
    MemoryContextSwitchTo(oldcontext);
 
    return winfo;
}

References ApplyContext, LogicalRepWorker::dbid, ParallelApplyWorkerInfo::dsm_seg, dsm_segment_handle(), ParallelApplyWorkerInfo::in_use, InvalidOid, lappend(), lfirst, logicalrep_worker_launch(), MemoryContextSwitchTo(), MyLogicalRepWorker, MySubscription, Subscription::name, Subscription::oid, pa_free_worker_info(), pa_setup_dsm(), palloc0(), ParallelApplyWorkerPool, pfree(), LogicalRepWorker::userid, and WORKERTYPE_PARALLEL_APPLY.

Referenced by pa_allocate_worker().

pa_lock_stream()

void pa_lock_stream	(	TransactionId	xid,
		LOCKMODE	lockmode
	)

Definition at line 1540 of file applyparallelworker.c.

{
    LockApplyTransactionForSession(MyLogicalRepWorker->subid, xid,
                                   PARALLEL_APPLY_LOCK_STREAM, lockmode);
}

References LockApplyTransactionForSession(), MyLogicalRepWorker, PARALLEL_APPLY_LOCK_STREAM, and LogicalRepWorker::subid.

Referenced by apply_handle_stream_abort(), apply_handle_stream_stop(), pa_decr_and_wait_stream_block(), pa_process_spooled_messages_if_required(), and pa_switch_to_partial_serialize().

pa_lock_transaction()

void pa_lock_transaction	(	TransactionId	xid,
		LOCKMODE	lockmode
	)

Definition at line 1573 of file applyparallelworker.c.

{
    LockApplyTransactionForSession(MyLogicalRepWorker->subid, xid,
                                   PARALLEL_APPLY_LOCK_XACT, lockmode);
}

References LockApplyTransactionForSession(), MyLogicalRepWorker, PARALLEL_APPLY_LOCK_XACT, and LogicalRepWorker::subid.

Referenced by apply_handle_stream_start(), and pa_wait_for_xact_finish().

pa_process_spooled_messages_if_required()

static bool pa_process_spooled_messages_if_required ( void  )

static

Definition at line 658 of file applyparallelworker.c.

{
    PartialFileSetState fileset_state;
 
    fileset_state = pa_get_fileset_state();
 
    if (fileset_state == FS_EMPTY)
        return false;
 
    /*
     * If the leader apply worker is busy serializing the partial changes then
     * acquire the stream lock now and wait for the leader worker to finish
     * serializing the changes. Otherwise, the parallel apply worker won't get
     * a chance to receive a STREAM_STOP (and acquire the stream lock) until
     * the leader had serialized all changes which can lead to undetected
     * deadlock.
     *
     * Note that the fileset state can be FS_SERIALIZE_DONE once the leader
     * worker has finished serializing the changes.
     */
    if (fileset_state == FS_SERIALIZE_IN_PROGRESS)
    {
        pa_lock_stream(MyParallelShared->xid, AccessShareLock);
        pa_unlock_stream(MyParallelShared->xid, AccessShareLock);
 
        fileset_state = pa_get_fileset_state();
    }
 
    /*
     * We cannot read the file immediately after the leader has serialized all
     * changes to the file because there may still be messages in the memory
     * queue. We will apply all spooled messages the next time we call this
     * function and that will ensure there are no messages left in the memory
     * queue.
     */
    if (fileset_state == FS_SERIALIZE_DONE)
    {
        pa_set_fileset_state(MyParallelShared, FS_READY);
    }
    else if (fileset_state == FS_READY)
    {
        apply_spooled_messages(&MyParallelShared->fileset,
                               MyParallelShared->xid,
                               InvalidXLogRecPtr);
        pa_set_fileset_state(MyParallelShared, FS_EMPTY);
    }
 
    return true;
}

References AccessShareLock, apply_spooled_messages(), ParallelApplyWorkerShared::fileset, FS_EMPTY, FS_READY, FS_SERIALIZE_DONE, FS_SERIALIZE_IN_PROGRESS, InvalidXLogRecPtr, MyParallelShared, pa_get_fileset_state(), pa_lock_stream(), pa_set_fileset_state(), pa_unlock_stream(), and ParallelApplyWorkerShared::xid.

Referenced by LogicalParallelApplyLoop().

pa_reset_subtrans()

void pa_reset_subtrans

(

void

)

Definition at line 1402 of file applyparallelworker.c.

{
    /*
     * We don't need to free this explicitly as the allocated memory will be
     * freed at the transaction end.
     */
    subxactlist = NIL;
}

References NIL, and subxactlist.

Referenced by apply_handle_stream_commit(), apply_handle_stream_prepare(), and pa_stream_abort().

pa_savepoint_name()

static void pa_savepoint_name ( Oid  suboid, TransactionId  xid, char *  spname, Size  szsp  )

static

Definition at line 1348 of file applyparallelworker.c.

{
    snprintf(spname, szsp, "pg_sp_%u_%u", suboid, xid);
}

References snprintf.

Referenced by pa_start_subtrans(), and pa_stream_abort().

pa_send_data()

bool pa_send_data	(	ParallelApplyWorkerInfo *	winfo,
		Size	nbytes,
		const void *	data
	)

Definition at line 1146 of file applyparallelworker.c.

{
    int         rc;
    shm_mq_result result;
    TimestampTz startTime = 0;
 
    Assert(!IsTransactionState());
    Assert(!winfo->serialize_changes);
 
    /*
     * We don't try to send data to parallel worker for 'immediate' mode. This
     * is primarily used for testing purposes.
     */
    if (unlikely(debug_logical_replication_streaming == DEBUG_LOGICAL_REP_STREAMING_IMMEDIATE))
        return false;
 
/*
 * This timeout is a bit arbitrary but testing revealed that it is sufficient
 * to send the message unless the parallel apply worker is waiting on some
 * lock or there is a serious resource crunch. See the comments atop this file
 * to know why we are using a non-blocking way to send the message.
 */
#define SHM_SEND_RETRY_INTERVAL_MS 1000
#define SHM_SEND_TIMEOUT_MS     (10000 - SHM_SEND_RETRY_INTERVAL_MS)
 
    for (;;)
    {
        result = shm_mq_send(winfo->mq_handle, nbytes, data, true, true);
 
        if (result == SHM_MQ_SUCCESS)
            return true;
        else if (result == SHM_MQ_DETACHED)
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("could not send data to shared-memory queue")));
 
        Assert(result == SHM_MQ_WOULD_BLOCK);
 
        /* Wait before retrying. */
        rc = WaitLatch(MyLatch,
                       WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
                       SHM_SEND_RETRY_INTERVAL_MS,
                       WAIT_EVENT_LOGICAL_APPLY_SEND_DATA);
 
        if (rc & WL_LATCH_SET)
        {
            ResetLatch(MyLatch);
            CHECK_FOR_INTERRUPTS();
        }
 
        if (startTime == 0)
            startTime = GetCurrentTimestamp();
        else if (TimestampDifferenceExceeds(startTime, GetCurrentTimestamp(),
                                            SHM_SEND_TIMEOUT_MS))
            return false;
    }
}

References Assert, CHECK_FOR_INTERRUPTS, data, DEBUG_LOGICAL_REP_STREAMING_IMMEDIATE, debug_logical_replication_streaming, ereport, errcode(), errmsg(), ERROR, GetCurrentTimestamp(), IsTransactionState(), ParallelApplyWorkerInfo::mq_handle, MyLatch, ResetLatch(), ParallelApplyWorkerInfo::serialize_changes, SHM_MQ_DETACHED, shm_mq_send(), SHM_MQ_SUCCESS, SHM_MQ_WOULD_BLOCK, SHM_SEND_RETRY_INTERVAL_MS, SHM_SEND_TIMEOUT_MS, TimestampDifferenceExceeds(), unlikely, WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

Referenced by apply_handle_stream_abort(), apply_handle_stream_commit(), apply_handle_stream_prepare(), apply_handle_stream_start(), apply_handle_stream_stop(), and handle_streamed_transaction().

pa_set_fileset_state()

void pa_set_fileset_state	(	ParallelApplyWorkerShared *	wshared,
		PartialFileSetState	fileset_state
	)

Definition at line 1498 of file applyparallelworker.c.

{
    SpinLockAcquire(&wshared->mutex);
    wshared->fileset_state = fileset_state;
 
    if (fileset_state == FS_SERIALIZE_DONE)
    {
        Assert(am_leader_apply_worker());
        Assert(MyLogicalRepWorker->stream_fileset);
        wshared->fileset = *MyLogicalRepWorker->stream_fileset;
    }
 
    SpinLockRelease(&wshared->mutex);
}

References am_leader_apply_worker(), Assert, ParallelApplyWorkerShared::fileset, ParallelApplyWorkerShared::fileset_state, FS_SERIALIZE_DONE, ParallelApplyWorkerShared::mutex, MyLogicalRepWorker, SpinLockAcquire, SpinLockRelease, and LogicalRepWorker::stream_fileset.

Referenced by apply_handle_stream_abort(), apply_handle_stream_commit(), apply_handle_stream_prepare(), pa_process_spooled_messages_if_required(), and pa_switch_to_partial_serialize().

pa_set_stream_apply_worker()

void pa_set_stream_apply_worker

(

ParallelApplyWorkerInfo *

winfo

)

Definition at line 1334 of file applyparallelworker.c.

{
    stream_apply_worker = winfo;
}

References stream_apply_worker.

Referenced by apply_handle_stream_start(), and apply_handle_stream_stop().

pa_set_xact_state()

void pa_set_xact_state	(	ParallelApplyWorkerShared *	wshared,
		ParallelTransState	xact_state
	)

Definition at line 1307 of file applyparallelworker.c.

{
    SpinLockAcquire(&wshared->mutex);
    wshared->xact_state = xact_state;
    SpinLockRelease(&wshared->mutex);
}

References ParallelApplyWorkerShared::mutex, SpinLockAcquire, SpinLockRelease, and ParallelApplyWorkerShared::xact_state.

Referenced by apply_handle_stream_commit(), apply_handle_stream_prepare(), apply_handle_stream_start(), and pa_stream_abort().

pa_setup_dsm()

static bool pa_setup_dsm ( ParallelApplyWorkerInfo *  winfo )

static

Definition at line 327 of file applyparallelworker.c.

{
    shm_toc_estimator e;
    Size        segsize;
    dsm_segment *seg;
    shm_toc    *toc;
    ParallelApplyWorkerShared *shared;
    shm_mq     *mq;
    Size        queue_size = DSM_QUEUE_SIZE;
    Size        error_queue_size = DSM_ERROR_QUEUE_SIZE;
 
    /*
     * Estimate how much shared memory we need.
     *
     * Because the TOC machinery may choose to insert padding of oddly-sized
     * requests, we must estimate each chunk separately.
     *
     * We need one key to register the location of the header, and two other
     * keys to track the locations of the message queue and the error message
     * queue.
     */
    shm_toc_initialize_estimator(&e);
    shm_toc_estimate_chunk(&e, sizeof(ParallelApplyWorkerShared));
    shm_toc_estimate_chunk(&e, queue_size);
    shm_toc_estimate_chunk(&e, error_queue_size);
 
    shm_toc_estimate_keys(&e, 3);
    segsize = shm_toc_estimate(&e);
 
    /* Create the shared memory segment and establish a table of contents. */
    seg = dsm_create(shm_toc_estimate(&e), 0);
    if (!seg)
        return false;
 
    toc = shm_toc_create(PG_LOGICAL_APPLY_SHM_MAGIC, dsm_segment_address(seg),
                         segsize);
 
    /* Set up the header region. */
    shared = shm_toc_allocate(toc, sizeof(ParallelApplyWorkerShared));
    SpinLockInit(&shared->mutex);
 
    shared->xact_state = PARALLEL_TRANS_UNKNOWN;
    pg_atomic_init_u32(&(shared->pending_stream_count), 0);
    shared->last_commit_end = InvalidXLogRecPtr;
    shared->fileset_state = FS_EMPTY;
 
    shm_toc_insert(toc, PARALLEL_APPLY_KEY_SHARED, shared);
 
    /* Set up message queue for the worker. */
    mq = shm_mq_create(shm_toc_allocate(toc, queue_size), queue_size);
    shm_toc_insert(toc, PARALLEL_APPLY_KEY_MQ, mq);
    shm_mq_set_sender(mq, MyProc);
 
    /* Attach the queue. */
    winfo->mq_handle = shm_mq_attach(mq, seg, NULL);
 
    /* Set up error queue for the worker. */
    mq = shm_mq_create(shm_toc_allocate(toc, error_queue_size),
                       error_queue_size);
    shm_toc_insert(toc, PARALLEL_APPLY_KEY_ERROR_QUEUE, mq);
    shm_mq_set_receiver(mq, MyProc);
 
    /* Attach the queue. */
    winfo->error_mq_handle = shm_mq_attach(mq, seg, NULL);
 
    /* Return results to caller. */
    winfo->dsm_seg = seg;
    winfo->shared = shared;
 
    return true;
}

References dsm_create(), DSM_ERROR_QUEUE_SIZE, DSM_QUEUE_SIZE, ParallelApplyWorkerInfo::dsm_seg, dsm_segment_address(), ParallelApplyWorkerInfo::error_mq_handle, ParallelApplyWorkerShared::fileset_state, FS_EMPTY, InvalidXLogRecPtr, ParallelApplyWorkerShared::last_commit_end, ParallelApplyWorkerInfo::mq_handle, ParallelApplyWorkerShared::mutex, MyProc, PARALLEL_APPLY_KEY_ERROR_QUEUE, PARALLEL_APPLY_KEY_MQ, PARALLEL_APPLY_KEY_SHARED, PARALLEL_TRANS_UNKNOWN, ParallelApplyWorkerShared::pending_stream_count, pg_atomic_init_u32(), PG_LOGICAL_APPLY_SHM_MAGIC, ParallelApplyWorkerInfo::shared, shm_mq_attach(), shm_mq_create(), shm_mq_set_receiver(), shm_mq_set_sender(), shm_toc_allocate(), shm_toc_create(), shm_toc_estimate(), shm_toc_estimate_chunk, shm_toc_estimate_keys, shm_toc_initialize_estimator, shm_toc_insert(), SpinLockInit, and ParallelApplyWorkerShared::xact_state.

Referenced by pa_launch_parallel_worker().

pa_shutdown()

static void pa_shutdown ( int  code, Datum  arg  )

static

Definition at line 844 of file applyparallelworker.c.

{
    SendProcSignal(MyLogicalRepWorker->leader_pid,
                   PROCSIG_PARALLEL_APPLY_MESSAGE,
                   INVALID_PROC_NUMBER);
 
    dsm_detach((dsm_segment *) DatumGetPointer(arg));
}

References arg, DatumGetPointer(), dsm_detach(), INVALID_PROC_NUMBER, LogicalRepWorker::leader_pid, MyLogicalRepWorker, PROCSIG_PARALLEL_APPLY_MESSAGE, and SendProcSignal().

Referenced by ParallelApplyWorkerMain().

pa_start_subtrans()

void pa_start_subtrans	(	TransactionId	current_xid,
		TransactionId	top_xid
	)

Definition at line 1362 of file applyparallelworker.c.

{
    if (current_xid != top_xid &&
        !list_member_xid(subxactlist, current_xid))
    {
        MemoryContext oldctx;
        char        spname[NAMEDATALEN];
 
        pa_savepoint_name(MySubscription->oid, current_xid,
                          spname, sizeof(spname));
 
        elog(DEBUG1, "defining savepoint %s in logical replication parallel apply worker", spname);
 
        /* We must be in transaction block to define the SAVEPOINT. */
        if (!IsTransactionBlock())
        {
            if (!IsTransactionState())
                StartTransactionCommand();
 
            BeginTransactionBlock();
            CommitTransactionCommand();
        }
 
        DefineSavepoint(spname);
 
        /*
         * CommitTransactionCommand is needed to start a subtransaction after
         * issuing a SAVEPOINT inside a transaction block (see
         * StartSubTransaction()).
         */
        CommitTransactionCommand();
 
        oldctx = MemoryContextSwitchTo(TopTransactionContext);
        subxactlist = lappend_xid(subxactlist, current_xid);
        MemoryContextSwitchTo(oldctx);
    }
}

References BeginTransactionBlock(), CommitTransactionCommand(), DEBUG1, DefineSavepoint(), elog, IsTransactionBlock(), IsTransactionState(), lappend_xid(), list_member_xid(), MemoryContextSwitchTo(), MySubscription, NAMEDATALEN, Subscription::oid, pa_savepoint_name(), StartTransactionCommand(), subxactlist, and TopTransactionContext.

Referenced by handle_streamed_transaction().

pa_stream_abort()

void pa_stream_abort

(

LogicalRepStreamAbortData *

abort_data

)

Definition at line 1416 of file applyparallelworker.c.

{
    TransactionId xid = abort_data->xid;
    TransactionId subxid = abort_data->subxid;
 
    /*
     * Update origin state so we can restart streaming from correct position
     * in case of crash.
     */
    replorigin_session_origin_lsn = abort_data->abort_lsn;
    replorigin_session_origin_timestamp = abort_data->abort_time;
 
    /*
     * If the two XIDs are the same, it's in fact abort of toplevel xact, so
     * just free the subxactlist.
     */
    if (subxid == xid)
    {
        pa_set_xact_state(MyParallelShared, PARALLEL_TRANS_FINISHED);
 
        /*
         * Release the lock as we might be processing an empty streaming
         * transaction in which case the lock won't be released during
         * transaction rollback.
         *
         * Note that it's ok to release the transaction lock before aborting
         * the transaction because even if the parallel apply worker dies due
         * to crash or some other reason, such a transaction would still be
         * considered aborted.
         */
        pa_unlock_transaction(xid, AccessExclusiveLock);
 
        AbortCurrentTransaction();
 
        if (IsTransactionBlock())
        {
            EndTransactionBlock(false);
            CommitTransactionCommand();
        }
 
        pa_reset_subtrans();
 
        pgstat_report_activity(STATE_IDLE, NULL);
    }
    else
    {
        /* OK, so it's a subxact. Rollback to the savepoint. */
        int         i;
        char        spname[NAMEDATALEN];
 
        pa_savepoint_name(MySubscription->oid, subxid, spname, sizeof(spname));
 
        elog(DEBUG1, "rolling back to savepoint %s in logical replication parallel apply worker", spname);
 
        /*
         * Search the subxactlist, determine the offset tracked for the
         * subxact, and truncate the list.
         *
         * Note that for an empty sub-transaction we won't find the subxid
         * here.
         */
        for (i = list_length(subxactlist) - 1; i >= 0; i--)
        {
            TransactionId xid_tmp = lfirst_xid(list_nth_cell(subxactlist, i));
 
            if (xid_tmp == subxid)
            {
                RollbackToSavepoint(spname);
                CommitTransactionCommand();
                subxactlist = list_truncate(subxactlist, i);
                break;
            }
        }
    }
}

References LogicalRepStreamAbortData::abort_lsn, LogicalRepStreamAbortData::abort_time, AbortCurrentTransaction(), AccessExclusiveLock, CommitTransactionCommand(), DEBUG1, elog, EndTransactionBlock(), i, IsTransactionBlock(), lfirst_xid, list_length(), list_nth_cell(), list_truncate(), MyParallelShared, MySubscription, NAMEDATALEN, Subscription::oid, pa_reset_subtrans(), pa_savepoint_name(), pa_set_xact_state(), pa_unlock_transaction(), PARALLEL_TRANS_FINISHED, pgstat_report_activity(), replorigin_session_origin_lsn, replorigin_session_origin_timestamp, RollbackToSavepoint(), STATE_IDLE, subxactlist, LogicalRepStreamAbortData::subxid, and LogicalRepStreamAbortData::xid.

Referenced by apply_handle_stream_abort().

pa_switch_to_partial_serialize()

void pa_switch_to_partial_serialize	(	ParallelApplyWorkerInfo *	winfo,
		bool	stream_locked
	)

Definition at line 1211 of file applyparallelworker.c.

{
    ereport(LOG,
            (errmsg("logical replication apply worker will serialize the remaining changes of remote transaction %u to a file",
                    winfo->shared->xid)));
 
    /*
     * The parallel apply worker could be stuck for some reason (say waiting
     * on some lock by other backend), so stop trying to send data directly to
     * it and start serializing data to the file instead.
     */
    winfo->serialize_changes = true;
 
    /* Initialize the stream fileset. */
    stream_start_internal(winfo->shared->xid, true);
 
    /*
     * Acquires the stream lock if not already to make sure that the parallel
     * apply worker will wait for the leader to release the stream lock until
     * the end of the transaction.
     */
    if (!stream_locked)
        pa_lock_stream(winfo->shared->xid, AccessExclusiveLock);
 
    pa_set_fileset_state(winfo->shared, FS_SERIALIZE_IN_PROGRESS);
}

References AccessExclusiveLock, ereport, errmsg(), FS_SERIALIZE_IN_PROGRESS, LOG, pa_lock_stream(), pa_set_fileset_state(), ParallelApplyWorkerInfo::serialize_changes, ParallelApplyWorkerInfo::shared, stream_start_internal(), and ParallelApplyWorkerShared::xid.

Referenced by apply_handle_stream_abort(), apply_handle_stream_commit(), apply_handle_stream_prepare(), apply_handle_stream_start(), apply_handle_stream_stop(), and handle_streamed_transaction().

pa_unlock_stream()

void pa_unlock_stream	(	TransactionId	xid,
		LOCKMODE	lockmode
	)

Definition at line 1547 of file applyparallelworker.c.

{
    UnlockApplyTransactionForSession(MyLogicalRepWorker->subid, xid,
                                     PARALLEL_APPLY_LOCK_STREAM, lockmode);
}

References MyLogicalRepWorker, PARALLEL_APPLY_LOCK_STREAM, LogicalRepWorker::subid, and UnlockApplyTransactionForSession().

Referenced by apply_handle_stream_abort(), apply_handle_stream_start(), pa_decr_and_wait_stream_block(), pa_process_spooled_messages_if_required(), and pa_xact_finish().

pa_unlock_transaction()

void pa_unlock_transaction	(	TransactionId	xid,
		LOCKMODE	lockmode
	)

Definition at line 1580 of file applyparallelworker.c.

{
    UnlockApplyTransactionForSession(MyLogicalRepWorker->subid, xid,
                                     PARALLEL_APPLY_LOCK_XACT, lockmode);
}

References MyLogicalRepWorker, PARALLEL_APPLY_LOCK_XACT, LogicalRepWorker::subid, and UnlockApplyTransactionForSession().

Referenced by apply_handle_stream_commit(), apply_handle_stream_prepare(), pa_stream_abort(), and pa_wait_for_xact_finish().

pa_wait_for_xact_finish()

static void pa_wait_for_xact_finish ( ParallelApplyWorkerInfo *  winfo )

static

Definition at line 1274 of file applyparallelworker.c.

{
    /*
     * Wait until the parallel apply worker set the state to
     * PARALLEL_TRANS_STARTED which means it has acquired the transaction
     * lock. This is to prevent leader apply worker from acquiring the
     * transaction lock earlier than the parallel apply worker.
     */
    pa_wait_for_xact_state(winfo, PARALLEL_TRANS_STARTED);
 
    /*
     * Wait for the transaction lock to be released. This is required to
     * detect deadlock among leader and parallel apply workers. Refer to the
     * comments atop this file.
     */
    pa_lock_transaction(winfo->shared->xid, AccessShareLock);
    pa_unlock_transaction(winfo->shared->xid, AccessShareLock);
 
    /*
     * Check if the state becomes PARALLEL_TRANS_FINISHED in case the parallel
     * apply worker failed while applying changes causing the lock to be
     * released.
     */
    if (pa_get_xact_state(winfo->shared) != PARALLEL_TRANS_FINISHED)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("lost connection to the logical replication parallel apply worker")));
}

References AccessShareLock, ereport, errcode(), errmsg(), ERROR, pa_get_xact_state(), pa_lock_transaction(), pa_unlock_transaction(), pa_wait_for_xact_state(), PARALLEL_TRANS_FINISHED, PARALLEL_TRANS_STARTED, ParallelApplyWorkerInfo::shared, and ParallelApplyWorkerShared::xid.

Referenced by pa_xact_finish().

pa_wait_for_xact_state()

static void pa_wait_for_xact_state ( ParallelApplyWorkerInfo *  winfo, ParallelTransState  xact_state  )

static

Definition at line 1244 of file applyparallelworker.c.

{
    for (;;)
    {
        /*
         * Stop if the transaction state has reached or exceeded the given
         * xact_state.
         */
        if (pa_get_xact_state(winfo->shared) >= xact_state)
            break;
 
        /* Wait to be signalled. */
        (void) WaitLatch(MyLatch,
                         WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
                         10L,
                         WAIT_EVENT_LOGICAL_PARALLEL_APPLY_STATE_CHANGE);
 
        /* Reset the latch so we don't spin. */
        ResetLatch(MyLatch);
 
        /* An interrupt may have occurred while we were waiting. */
        CHECK_FOR_INTERRUPTS();
    }
}

References CHECK_FOR_INTERRUPTS, MyLatch, pa_get_xact_state(), ResetLatch(), ParallelApplyWorkerInfo::shared, WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

Referenced by pa_wait_for_xact_finish().

pa_xact_finish()

void pa_xact_finish	(	ParallelApplyWorkerInfo *	winfo,
		XLogRecPtr	remote_lsn
	)

Definition at line 1618 of file applyparallelworker.c.

{
    Assert(am_leader_apply_worker());
 
    /*
     * Unlock the shared object lock so that parallel apply worker can
     * continue to receive and apply changes.
     */
    pa_unlock_stream(winfo->shared->xid, AccessExclusiveLock);
 
    /*
     * Wait for that worker to finish. This is necessary to maintain commit
     * order which avoids failures due to transaction dependencies and
     * deadlocks.
     */
    pa_wait_for_xact_finish(winfo);
 
    if (!XLogRecPtrIsInvalid(remote_lsn))
        store_flush_position(remote_lsn, winfo->shared->last_commit_end);
 
    pa_free_worker(winfo);
}

References AccessExclusiveLock, am_leader_apply_worker(), Assert, ParallelApplyWorkerShared::last_commit_end, pa_free_worker(), pa_unlock_stream(), pa_wait_for_xact_finish(), ParallelApplyWorkerInfo::shared, store_flush_position(), ParallelApplyWorkerShared::xid, and XLogRecPtrIsInvalid.

Referenced by apply_handle_stream_abort(), apply_handle_stream_commit(), and apply_handle_stream_prepare().

ParallelApplyWorkerMain()

void ParallelApplyWorkerMain

(

Datum

main_arg

)

Definition at line 857 of file applyparallelworker.c.

{
    ParallelApplyWorkerShared *shared;
    dsm_handle  handle;
    dsm_segment *seg;
    shm_toc    *toc;
    shm_mq     *mq;
    shm_mq_handle *mqh;
    shm_mq_handle *error_mqh;
    RepOriginId originid;
    int         worker_slot = DatumGetInt32(main_arg);
    char        originname[NAMEDATALEN];
 
    InitializingApplyWorker = true;
 
    /* Setup signal handling. */
    pqsignal(SIGHUP, SignalHandlerForConfigReload);
    pqsignal(SIGINT, SignalHandlerForShutdownRequest);
    pqsignal(SIGTERM, die);
    BackgroundWorkerUnblockSignals();
 
    /*
     * Attach to the dynamic shared memory segment for the parallel apply, and
     * find its table of contents.
     *
     * Like parallel query, we don't need resource owner by this time. See
     * ParallelWorkerMain.
     */
    memcpy(&handle, MyBgworkerEntry->bgw_extra, sizeof(dsm_handle));
    seg = dsm_attach(handle);
    if (!seg)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("could not map dynamic shared memory segment")));
 
    toc = shm_toc_attach(PG_LOGICAL_APPLY_SHM_MAGIC, dsm_segment_address(seg));
    if (!toc)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("invalid magic number in dynamic shared memory segment")));
 
    /* Look up the shared information. */
    shared = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_SHARED, false);
    MyParallelShared = shared;
 
    /*
     * Attach to the message queue.
     */
    mq = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_MQ, false);
    shm_mq_set_receiver(mq, MyProc);
    mqh = shm_mq_attach(mq, seg, NULL);
 
    /*
     * Primary initialization is complete. Now, we can attach to our slot.
     * This is to ensure that the leader apply worker does not write data to
     * the uninitialized memory queue.
     */
    logicalrep_worker_attach(worker_slot);
 
    /*
     * Register the shutdown callback after we are attached to the worker
     * slot. This is to ensure that MyLogicalRepWorker remains valid when this
     * callback is invoked.
     */
    before_shmem_exit(pa_shutdown, PointerGetDatum(seg));
 
    SpinLockAcquire(&MyParallelShared->mutex);
    MyParallelShared->logicalrep_worker_generation = MyLogicalRepWorker->generation;
    MyParallelShared->logicalrep_worker_slot_no = worker_slot;
    SpinLockRelease(&MyParallelShared->mutex);
 
    /*
     * Attach to the error queue.
     */
    mq = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_ERROR_QUEUE, false);
    shm_mq_set_sender(mq, MyProc);
    error_mqh = shm_mq_attach(mq, seg, NULL);
 
    pq_redirect_to_shm_mq(seg, error_mqh);
    pq_set_parallel_leader(MyLogicalRepWorker->leader_pid,
                           INVALID_PROC_NUMBER);
 
    MyLogicalRepWorker->last_send_time = MyLogicalRepWorker->last_recv_time =
        MyLogicalRepWorker->reply_time = 0;
 
    InitializeLogRepWorker();
 
    InitializingApplyWorker = false;
 
    /* Setup replication origin tracking. */
    StartTransactionCommand();
    ReplicationOriginNameForLogicalRep(MySubscription->oid, InvalidOid,
                                       originname, sizeof(originname));
    originid = replorigin_by_name(originname, false);
 
    /*
     * The parallel apply worker doesn't need to monopolize this replication
     * origin which was already acquired by its leader process.
     */
    replorigin_session_setup(originid, MyLogicalRepWorker->leader_pid);
    replorigin_session_origin = originid;
    CommitTransactionCommand();
 
    /*
     * Setup callback for syscache so that we know when something changes in
     * the subscription relation state.
     */
    CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
                                  invalidate_syncing_table_states,
                                  (Datum) 0);
 
    set_apply_error_context_origin(originname);
 
    LogicalParallelApplyLoop(mqh);
 
    /*
     * The parallel apply worker must not get here because the parallel apply
     * worker will only stop when it receives a SIGTERM or SIGINT from the
     * leader, or when there is an error. None of these cases will allow the
     * code to reach here.
     */
    Assert(false);
}

References Assert, BackgroundWorkerUnblockSignals(), before_shmem_exit(), BackgroundWorker::bgw_extra, CacheRegisterSyscacheCallback(), CommitTransactionCommand(), DatumGetInt32(), die, dsm_attach(), dsm_segment_address(), ereport, errcode(), errmsg(), ERROR, LogicalRepWorker::generation, InitializeLogRepWorker(), InitializingApplyWorker, INVALID_PROC_NUMBER, invalidate_syncing_table_states(), InvalidOid, LogicalRepWorker::last_recv_time, LogicalRepWorker::last_send_time, LogicalRepWorker::leader_pid, LogicalParallelApplyLoop(), logicalrep_worker_attach(), ParallelApplyWorkerShared::logicalrep_worker_generation, ParallelApplyWorkerShared::logicalrep_worker_slot_no, ParallelApplyWorkerShared::mutex, MyBgworkerEntry, MyLogicalRepWorker, MyParallelShared, MyProc, MySubscription, NAMEDATALEN, Subscription::oid, pa_shutdown(), PARALLEL_APPLY_KEY_ERROR_QUEUE, PARALLEL_APPLY_KEY_MQ, PARALLEL_APPLY_KEY_SHARED, PG_LOGICAL_APPLY_SHM_MAGIC, PointerGetDatum(), pq_redirect_to_shm_mq(), pq_set_parallel_leader(), pqsignal(), ReplicationOriginNameForLogicalRep(), replorigin_by_name(), replorigin_session_origin, replorigin_session_setup(), LogicalRepWorker::reply_time, set_apply_error_context_origin(), shm_mq_attach(), shm_mq_set_receiver(), shm_mq_set_sender(), shm_toc_attach(), shm_toc_lookup(), SIGHUP, SignalHandlerForConfigReload(), SignalHandlerForShutdownRequest(), SpinLockAcquire, SpinLockRelease, and StartTransactionCommand().

ProcessParallelApplyInterrupts()

static void ProcessParallelApplyInterrupts ( void  )

static

Definition at line 712 of file applyparallelworker.c.

{
    CHECK_FOR_INTERRUPTS();
 
    if (ShutdownRequestPending)
    {
        ereport(LOG,
                (errmsg("logical replication parallel apply worker for subscription \"%s\" has finished",
                        MySubscription->name)));
 
        proc_exit(0);
    }
 
    if (ConfigReloadPending)
    {
        ConfigReloadPending = false;
        ProcessConfigFile(PGC_SIGHUP);
    }
}

References CHECK_FOR_INTERRUPTS, ConfigReloadPending, ereport, errmsg(), LOG, MySubscription, Subscription::name, PGC_SIGHUP, proc_exit(), ProcessConfigFile(), and ShutdownRequestPending.

Referenced by LogicalParallelApplyLoop().

Variable Documentation

MyParallelShared

ParallelApplyWorkerShared* MyParallelShared = NULL

Definition at line 239 of file applyparallelworker.c.

Referenced by apply_handle_stream_commit(), apply_handle_stream_prepare(), apply_handle_stream_start(), pa_decr_and_wait_stream_block(), pa_get_fileset_state(), pa_process_spooled_messages_if_required(), pa_stream_abort(), and ParallelApplyWorkerMain().

ParallelApplyMessagePending

volatile sig_atomic_t ParallelApplyMessagePending = false

Definition at line 245 of file applyparallelworker.c.

Referenced by HandleParallelApplyMessageInterrupt(), HandleParallelApplyMessages(), and ProcessInterrupts().

ParallelApplyTxnHash

HTAB* ParallelApplyTxnHash = NULL

static

Definition at line 225 of file applyparallelworker.c.

Referenced by pa_allocate_worker(), pa_find_worker(), and pa_free_worker().

ParallelApplyWorkerPool

List* ParallelApplyWorkerPool = NIL

static

Definition at line 234 of file applyparallelworker.c.

Referenced by HandleParallelApplyMessages(), pa_detach_all_error_mq(), pa_free_worker(), pa_free_worker_info(), and pa_launch_parallel_worker().

stream_apply_worker

ParallelApplyWorkerInfo* stream_apply_worker = NULL

static

Definition at line 252 of file applyparallelworker.c.

Referenced by pa_find_worker(), and pa_set_stream_apply_worker().

subxactlist

List* subxactlist = NIL

static

Definition at line 255 of file applyparallelworker.c.

Referenced by pa_reset_subtrans(), pa_start_subtrans(), and pa_stream_abort().