parallel.h File Reference

#include <limits.h>
#include "pg_backup_archiver.h"

Include dependency graph for parallel.h:

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

Go to the source code of this file.

Data Structures
struct	ParallelState

Macros
#define	PG_MAX_JOBS   INT_MAX

Typedefs
typedef void(*	ParallelCompletionPtr) (ArchiveHandle *AH, TocEntry *te, int status, void *callback_data)
typedef struct ParallelSlot	ParallelSlot
typedef struct ParallelState	ParallelState

Enumerations
enum	WFW_WaitOption { WFW_NO_WAIT , WFW_GOT_STATUS , WFW_ONE_IDLE , WFW_ALL_IDLE }

Functions
void	init_parallel_dump_utils (void)
bool	IsEveryWorkerIdle (ParallelState *pstate)
void	WaitForWorkers (ArchiveHandle *AH, ParallelState *pstate, WFW_WaitOption mode)
ParallelState *	ParallelBackupStart (ArchiveHandle *AH)
void	DispatchJobForTocEntry (ArchiveHandle *AH, ParallelState *pstate, TocEntry *te, T_Action act, ParallelCompletionPtr callback, void *callback_data)
void	ParallelBackupEnd (ArchiveHandle *AH, ParallelState *pstate)
void	set_archive_cancel_info (ArchiveHandle *AH, PGconn *conn)

Macro Definition Documentation

PG_MAX_JOBS

#define PG_MAX_JOBS   INT_MAX

Definition at line 48 of file parallel.h.

Typedef Documentation

ParallelCompletionPtr

typedef void(* ParallelCompletionPtr) (ArchiveHandle *AH, TocEntry *te, int status, void *callback_data)

Definition at line 24 of file parallel.h.

ParallelSlot

typedef struct ParallelSlot ParallelSlot

Definition at line 24 of file parallel.h.

ParallelState

typedef struct ParallelState ParallelState

Enumeration Type Documentation

WFW_WaitOption

enum WFW_WaitOption

Enumerator
WFW_NO_WAIT
WFW_GOT_STATUS
WFW_ONE_IDLE
WFW_ALL_IDLE

Definition at line 30 of file parallel.h.

{
    WFW_NO_WAIT,
    WFW_GOT_STATUS,
    WFW_ONE_IDLE,
    WFW_ALL_IDLE
} WFW_WaitOption;

Function Documentation

DispatchJobForTocEntry()

void DispatchJobForTocEntry	(	ArchiveHandle *	AH,
		ParallelState *	pstate,
		TocEntry *	te,
		T_Action	act,
		ParallelCompletionPtr	callback,
		void *	callback_data
	)

Definition at line 1205 of file parallel.c.

{
    int         worker;
    char        buf[256];
 
    /* Get a worker, waiting if none are idle */
    while ((worker = GetIdleWorker(pstate)) == NO_SLOT)
        WaitForWorkers(AH, pstate, WFW_ONE_IDLE);
 
    /* Construct and send command string */
    buildWorkerCommand(AH, te, act, buf, sizeof(buf));
 
    sendMessageToWorker(pstate, worker, buf);
 
    /* Remember worker is busy, and which TocEntry it's working on */
    pstate->parallelSlot[worker].workerStatus = WRKR_WORKING;
    pstate->parallelSlot[worker].callback = callback;
    pstate->parallelSlot[worker].callback_data = callback_data;
    pstate->te[worker] = te;
}

References buf, buildWorkerCommand(), ParallelSlot::callback, callback(), ParallelSlot::callback_data, GetIdleWorker(), NO_SLOT, ParallelState::parallelSlot, sendMessageToWorker(), ParallelState::te, WaitForWorkers(), WFW_ONE_IDLE, ParallelSlot::workerStatus, and WRKR_WORKING.

Referenced by restore_toc_entries_parallel(), and WriteDataChunks().

init_parallel_dump_utils()

void init_parallel_dump_utils

(

void

)

Definition at line 236 of file parallel.c.

{
#ifdef WIN32
    if (!parallel_init_done)
    {
        WSADATA     wsaData;
        int         err;
 
        /* Prepare for threaded operation */
        tls_index = TlsAlloc();
        mainThreadId = GetCurrentThreadId();
 
        /* Initialize socket access */
        err = WSAStartup(MAKEWORD(2, 2), &wsaData);
        if (err != 0)
            pg_fatal("%s() failed: error code %d", "WSAStartup", err);
 
        parallel_init_done = true;
    }
#endif
}

References err(), and pg_fatal.

Referenced by main().

IsEveryWorkerIdle()

bool IsEveryWorkerIdle

(

ParallelState *

pstate

)

Definition at line 1268 of file parallel.c.

{
    int         i;
 
    for (i = 0; i < pstate->numWorkers; i++)
    {
        if (pstate->parallelSlot[i].workerStatus != WRKR_IDLE)
            return false;
    }
    return true;
}

References i, ParallelState::numWorkers, ParallelState::parallelSlot, ParallelSlot::workerStatus, and WRKR_IDLE.

Referenced by ParallelBackupEnd(), restore_toc_entries_parallel(), and WaitForWorkers().

ParallelBackupEnd()

void ParallelBackupEnd	(	ArchiveHandle *	AH,
		ParallelState *	pstate
	)

Definition at line 1059 of file parallel.c.

{
    int         i;
 
    /* No work if non-parallel */
    if (pstate->numWorkers == 1)
        return;
 
    /* There should not be any unfinished jobs */
    Assert(IsEveryWorkerIdle(pstate));
 
    /* Close the sockets so that the workers know they can exit */
    for (i = 0; i < pstate->numWorkers; i++)
    {
        closesocket(pstate->parallelSlot[i].pipeRead);
        closesocket(pstate->parallelSlot[i].pipeWrite);
    }
 
    /* Wait for them to exit */
    WaitForTerminatingWorkers(pstate);
 
    /*
     * Unlink pstate from shutdown_info, so the exit handler will not try to
     * use it; and likewise unlink from signal_info.
     */
    shutdown_info.pstate = NULL;
    set_cancel_pstate(NULL);
 
    /* Release state (mere neatnik-ism, since we're about to terminate) */
    free(pstate->te);
    free(pstate->parallelSlot);
    free(pstate);
}

References Assert(), closesocket, free, i, IsEveryWorkerIdle(), ParallelState::numWorkers, ParallelState::parallelSlot, ParallelSlot::pipeRead, ParallelSlot::pipeWrite, ShutdownInformation::pstate, set_cancel_pstate(), shutdown_info, ParallelState::te, and WaitForTerminatingWorkers().

Referenced by _CloseArchive(), and RestoreArchive().

ParallelBackupStart()

ParallelState* ParallelBackupStart

(

ArchiveHandle *

AH

)

Definition at line 897 of file parallel.c.

{
    ParallelState *pstate;
    int         i;
 
    Assert(AH->public.numWorkers > 0);
 
    pstate = (ParallelState *) pg_malloc(sizeof(ParallelState));
 
    pstate->numWorkers = AH->public.numWorkers;
    pstate->te = NULL;
    pstate->parallelSlot = NULL;
 
    if (AH->public.numWorkers == 1)
        return pstate;
 
    /* Create status arrays, being sure to initialize all fields to 0 */
    pstate->te = (TocEntry **)
        pg_malloc0(pstate->numWorkers * sizeof(TocEntry *));
    pstate->parallelSlot = (ParallelSlot *)
        pg_malloc0(pstate->numWorkers * sizeof(ParallelSlot));
 
#ifdef WIN32
    /* Make fmtId() and fmtQualifiedId() use thread-local storage */
    getLocalPQExpBuffer = getThreadLocalPQExpBuffer;
#endif
 
    /*
     * Set the pstate in shutdown_info, to tell the exit handler that it must
     * clean up workers as well as the main database connection.  But we don't
     * set this in signal_info yet, because we don't want child processes to
     * inherit non-NULL signal_info.pstate.
     */
    shutdown_info.pstate = pstate;
 
    /*
     * Temporarily disable query cancellation on the leader connection.  This
     * ensures that child processes won't inherit valid AH->connCancel
     * settings and thus won't try to issue cancels against the leader's
     * connection.  No harm is done if we fail while it's disabled, because
     * the leader connection is idle at this point anyway.
     */
    set_archive_cancel_info(AH, NULL);
 
    /* Ensure stdio state is quiesced before forking */
    fflush(NULL);
 
    /* Create desired number of workers */
    for (i = 0; i < pstate->numWorkers; i++)
    {
#ifdef WIN32
        WorkerInfo *wi;
        uintptr_t   handle;
#else
        pid_t       pid;
#endif
        ParallelSlot *slot = &(pstate->parallelSlot[i]);
        int         pipeMW[2],
                    pipeWM[2];
 
        /* Create communication pipes for this worker */
        if (pgpipe(pipeMW) < 0 || pgpipe(pipeWM) < 0)
            pg_fatal("could not create communication channels: %m");
 
        /* leader's ends of the pipes */
        slot->pipeRead = pipeWM[PIPE_READ];
        slot->pipeWrite = pipeMW[PIPE_WRITE];
        /* child's ends of the pipes */
        slot->pipeRevRead = pipeMW[PIPE_READ];
        slot->pipeRevWrite = pipeWM[PIPE_WRITE];
 
#ifdef WIN32
        /* Create transient structure to pass args to worker function */
        wi = (WorkerInfo *) pg_malloc(sizeof(WorkerInfo));
 
        wi->AH = AH;
        wi->slot = slot;
 
        handle = _beginthreadex(NULL, 0, (void *) &init_spawned_worker_win32,
                                wi, 0, &(slot->threadId));
        slot->hThread = handle;
        slot->workerStatus = WRKR_IDLE;
#else                           /* !WIN32 */
        pid = fork();
        if (pid == 0)
        {
            /* we are the worker */
            int         j;
 
            /* this is needed for GetMyPSlot() */
            slot->pid = getpid();
 
            /* instruct signal handler that we're in a worker now */
            signal_info.am_worker = true;
 
            /* close read end of Worker -> Leader */
            closesocket(pipeWM[PIPE_READ]);
            /* close write end of Leader -> Worker */
            closesocket(pipeMW[PIPE_WRITE]);
 
            /*
             * Close all inherited fds for communication of the leader with
             * previously-forked workers.
             */
            for (j = 0; j < i; j++)
            {
                closesocket(pstate->parallelSlot[j].pipeRead);
                closesocket(pstate->parallelSlot[j].pipeWrite);
            }
 
            /* Run the worker ... */
            RunWorker(AH, slot);
 
            /* We can just exit(0) when done */
            exit(0);
        }
        else if (pid < 0)
        {
            /* fork failed */
            pg_fatal("could not create worker process: %m");
        }
 
        /* In Leader after successful fork */
        slot->pid = pid;
        slot->workerStatus = WRKR_IDLE;
 
        /* close read end of Leader -> Worker */
        closesocket(pipeMW[PIPE_READ]);
        /* close write end of Worker -> Leader */
        closesocket(pipeWM[PIPE_WRITE]);
#endif                          /* WIN32 */
    }
 
    /*
     * Having forked off the workers, disable SIGPIPE so that leader isn't
     * killed if it tries to send a command to a dead worker.  We don't want
     * the workers to inherit this setting, though.
     */
#ifndef WIN32
    pqsignal(SIGPIPE, SIG_IGN);
#endif
 
    /*
     * Re-establish query cancellation on the leader connection.
     */
    set_archive_cancel_info(AH, AH->connection);
 
    /*
     * Tell the cancel signal handler to forward signals to worker processes,
     * too.  (As with query cancel, we did not need this earlier because the
     * workers have not yet been given anything to do; if we die before this
     * point, any already-started workers will see EOF and quit promptly.)
     */
    set_cancel_pstate(pstate);
 
    return pstate;
}

References DumpSignalInformation::am_worker, Assert(), closesocket, _archiveHandle::connection, exit(), fflush(), getLocalPQExpBuffer, i, j, ParallelState::numWorkers, Archive::numWorkers, ParallelState::parallelSlot, pg_fatal, pg_malloc(), pg_malloc0(), pgpipe, ParallelSlot::pid, PIPE_READ, PIPE_WRITE, ParallelSlot::pipeRead, ParallelSlot::pipeRevRead, ParallelSlot::pipeRevWrite, ParallelSlot::pipeWrite, pqsignal(), ShutdownInformation::pstate, _archiveHandle::public, RunWorker(), set_archive_cancel_info(), set_cancel_pstate(), shutdown_info, SIG_IGN, signal_info, SIGPIPE, ParallelState::te, ParallelSlot::workerStatus, and WRKR_IDLE.

Referenced by _CloseArchive(), and RestoreArchive().

set_archive_cancel_info()

void set_archive_cancel_info	(	ArchiveHandle *	AH,
		PGconn *	conn
	)

Definition at line 730 of file parallel.c.

{
    PGcancel   *oldConnCancel;
 
    /*
     * Activate the interrupt handler if we didn't yet in this process.  On
     * Windows, this also initializes signal_info_lock; therefore it's
     * important that this happen at least once before we fork off any
     * threads.
     */
    setup_cancel_handler();
 
    /*
     * On Unix, we assume that storing a pointer value is atomic with respect
     * to any possible signal interrupt.  On Windows, use a critical section.
     */
 
#ifdef WIN32
    EnterCriticalSection(&signal_info_lock);
#endif
 
    /* Free the old one if we have one */
    oldConnCancel = AH->connCancel;
    /* be sure interrupt handler doesn't use pointer while freeing */
    AH->connCancel = NULL;
 
    if (oldConnCancel != NULL)
        PQfreeCancel(oldConnCancel);
 
    /* Set the new one if specified */
    if (conn)
        AH->connCancel = PQgetCancel(conn);
 
    /*
     * On Unix, there's only ever one active ArchiveHandle per process, so we
     * can just set signal_info.myAH unconditionally.  On Windows, do that
     * only in the main thread; worker threads have to make sure their
     * ArchiveHandle appears in the pstate data, which is dealt with in
     * RunWorker().
     */
#ifndef WIN32
    signal_info.myAH = AH;
#else
    if (mainThreadId == GetCurrentThreadId())
        signal_info.myAH = AH;
#endif
 
#ifdef WIN32
    LeaveCriticalSection(&signal_info_lock);
#endif
}

References conn, _archiveHandle::connCancel, DumpSignalInformation::myAH, PQfreeCancel(), PQgetCancel(), setup_cancel_handler(), and signal_info.

Referenced by ConnectDatabase(), DisconnectDatabase(), and ParallelBackupStart().

WaitForWorkers()

void WaitForWorkers	(	ArchiveHandle *	AH,
		ParallelState *	pstate,
		WFW_WaitOption	mode
	)

Definition at line 1451 of file parallel.c.

{
    bool        do_wait = false;
 
    /*
     * In GOT_STATUS mode, always block waiting for a message, since we can't
     * return till we get something.  In other modes, we don't block the first
     * time through the loop.
     */
    if (mode == WFW_GOT_STATUS)
    {
        /* Assert that caller knows what it's doing */
        Assert(!IsEveryWorkerIdle(pstate));
        do_wait = true;
    }
 
    for (;;)
    {
        /*
         * Check for status messages, even if we don't need to block.  We do
         * not try very hard to reap all available messages, though, since
         * there's unlikely to be more than one.
         */
        if (ListenToWorkers(AH, pstate, do_wait))
        {
            /*
             * If we got a message, we are done by definition for GOT_STATUS
             * mode, and we can also be certain that there's at least one idle
             * worker.  So we're done in all but ALL_IDLE mode.
             */
            if (mode != WFW_ALL_IDLE)
                return;
        }
 
        /* Check whether we must wait for new status messages */
        switch (mode)
        {
            case WFW_NO_WAIT:
                return;         /* never wait */
            case WFW_GOT_STATUS:
                Assert(false);  /* can't get here, because we waited */
                break;
            case WFW_ONE_IDLE:
                if (GetIdleWorker(pstate) != NO_SLOT)
                    return;
                break;
            case WFW_ALL_IDLE:
                if (IsEveryWorkerIdle(pstate))
                    return;
                break;
        }
 
        /* Loop back, and this time wait for something to happen */
        do_wait = true;
    }
}

References Assert(), do_wait, GetIdleWorker(), IsEveryWorkerIdle(), ListenToWorkers(), mode, NO_SLOT, WFW_ALL_IDLE, WFW_GOT_STATUS, WFW_NO_WAIT, and WFW_ONE_IDLE.

Referenced by DispatchJobForTocEntry(), restore_toc_entries_parallel(), and WriteDataChunks().