parallel.c File Reference

#include "postgres_fe.h"
#include <sys/wait.h>
#include "pg_upgrade.h"

Include dependency graph for parallel.c:

Go to the source code of this file.

Functions
void	parallel_exec_prog (const char *log_file, const char *opt_log_file, const char *fmt,...)
void	parallel_transfer_all_new_dbs (DbInfoArr *old_db_arr, DbInfoArr *new_db_arr, char *old_pgdata, char *new_pgdata, char *old_tablespace, char *new_tablespace)
bool	reap_child (bool wait_for_child)

Variables
static int	parallel_jobs

Function Documentation

parallel_exec_prog()

void parallel_exec_prog	(	const char *	log_file,
		const char *	opt_log_file,
		const char *	fmt,
			...
	)

Definition at line 63 of file parallel.c.

{
    va_list     args;
    char        cmd[MAX_STRING];
 
#ifndef WIN32
    pid_t       child;
#else
    HANDLE      child;
    exec_thread_arg *new_arg;
#endif
 
    va_start(args, fmt);
    vsnprintf(cmd, sizeof(cmd), fmt, args);
    va_end(args);
 
    if (user_opts.jobs <= 1)
        /* exit_on_error must be true to allow jobs */
        exec_prog(log_file, opt_log_file, true, true, "%s", cmd);
    else
    {
        /* parallel */
#ifdef WIN32
        if (thread_handles == NULL)
            thread_handles = pg_malloc(user_opts.jobs * sizeof(HANDLE));
 
        if (exec_thread_args == NULL)
        {
            int         i;
 
            exec_thread_args = pg_malloc(user_opts.jobs * sizeof(exec_thread_arg *));
 
            /*
             * For safety and performance, we keep the args allocated during
             * the entire life of the process, and we don't free the args in a
             * thread different from the one that allocated it.
             */
            for (i = 0; i < user_opts.jobs; i++)
                exec_thread_args[i] = pg_malloc0(sizeof(exec_thread_arg));
        }
 
        cur_thread_args = (void **) exec_thread_args;
#endif
        /* harvest any dead children */
        while (reap_child(false) == true)
            ;
 
        /* must we wait for a dead child? */
        if (parallel_jobs >= user_opts.jobs)
            reap_child(true);
 
        /* set this before we start the job */
        parallel_jobs++;
 
        /* Ensure stdio state is quiesced before forking */
        fflush(NULL);
 
#ifndef WIN32
        child = fork();
        if (child == 0)
            /* use _exit to skip atexit() functions */
            _exit(!exec_prog(log_file, opt_log_file, true, true, "%s", cmd));
        else if (child < 0)
            /* fork failed */
            pg_fatal("could not create worker process: %m");
#else
        /* empty array element are always at the end */
        new_arg = exec_thread_args[parallel_jobs - 1];
 
        /* Can only pass one pointer into the function, so use a struct */
        pg_free(new_arg->log_file);
        new_arg->log_file = pg_strdup(log_file);
        pg_free(new_arg->opt_log_file);
        new_arg->opt_log_file = opt_log_file ? pg_strdup(opt_log_file) : NULL;
        pg_free(new_arg->cmd);
        new_arg->cmd = pg_strdup(cmd);
 
        child = (HANDLE) _beginthreadex(NULL, 0, (void *) win32_exec_prog,
                                        new_arg, 0, NULL);
        if (child == 0)
            pg_fatal("could not create worker thread: %m");
 
        thread_handles[parallel_jobs - 1] = child;
#endif
    }
}

References generate_unaccent_rules::args, exec_prog(), i, UserOpts::jobs, log_file, MAX_STRING, parallel_jobs, pg_fatal, pg_free(), pg_malloc(), pg_malloc0(), pg_strdup(), reap_child(), user_opts, and vsnprintf.

Referenced by create_new_objects(), and generate_old_dump().

parallel_transfer_all_new_dbs()

void parallel_transfer_all_new_dbs	(	DbInfoArr *	old_db_arr,
		DbInfoArr *	new_db_arr,
		char *	old_pgdata,
		char *	new_pgdata,
		char *	old_tablespace,
		char *	new_tablespace
	)

Definition at line 173 of file parallel.c.

{
#ifndef WIN32
    pid_t       child;
#else
    HANDLE      child;
    transfer_thread_arg *new_arg;
#endif
 
    if (user_opts.jobs <= 1)
        transfer_all_new_dbs(old_db_arr, new_db_arr, old_pgdata, new_pgdata, NULL, NULL);
    else
    {
        /* parallel */
#ifdef WIN32
        if (thread_handles == NULL)
            thread_handles = pg_malloc(user_opts.jobs * sizeof(HANDLE));
 
        if (transfer_thread_args == NULL)
        {
            int         i;
 
            transfer_thread_args = pg_malloc(user_opts.jobs * sizeof(transfer_thread_arg *));
 
            /*
             * For safety and performance, we keep the args allocated during
             * the entire life of the process, and we don't free the args in a
             * thread different from the one that allocated it.
             */
            for (i = 0; i < user_opts.jobs; i++)
                transfer_thread_args[i] = pg_malloc0(sizeof(transfer_thread_arg));
        }
 
        cur_thread_args = (void **) transfer_thread_args;
#endif
        /* harvest any dead children */
        while (reap_child(false) == true)
            ;
 
        /* must we wait for a dead child? */
        if (parallel_jobs >= user_opts.jobs)
            reap_child(true);
 
        /* set this before we start the job */
        parallel_jobs++;
 
        /* Ensure stdio state is quiesced before forking */
        fflush(NULL);
 
#ifndef WIN32
        child = fork();
        if (child == 0)
        {
            transfer_all_new_dbs(old_db_arr, new_db_arr, old_pgdata, new_pgdata,
                                 old_tablespace, new_tablespace);
            /* if we take another exit path, it will be non-zero */
            /* use _exit to skip atexit() functions */
            _exit(0);
        }
        else if (child < 0)
            /* fork failed */
            pg_fatal("could not create worker process: %m");
#else
        /* empty array element are always at the end */
        new_arg = transfer_thread_args[parallel_jobs - 1];
 
        /* Can only pass one pointer into the function, so use a struct */
        new_arg->old_db_arr = old_db_arr;
        new_arg->new_db_arr = new_db_arr;
        pg_free(new_arg->old_pgdata);
        new_arg->old_pgdata = pg_strdup(old_pgdata);
        pg_free(new_arg->new_pgdata);
        new_arg->new_pgdata = pg_strdup(new_pgdata);
        pg_free(new_arg->old_tablespace);
        new_arg->old_tablespace = old_tablespace ? pg_strdup(old_tablespace) : NULL;
        new_arg->new_tablespace = new_tablespace ? pg_strdup(new_tablespace) : NULL;
 
        child = (HANDLE) _beginthreadex(NULL, 0, (void *) win32_transfer_all_new_dbs,
                                        new_arg, 0, NULL);
        if (child == 0)
            pg_fatal("could not create worker thread: %m");
 
        thread_handles[parallel_jobs - 1] = child;
#endif
    }
}

References i, UserOpts::jobs, parallel_jobs, pg_fatal, pg_free(), pg_malloc(), pg_malloc0(), pg_strdup(), reap_child(), transfer_all_new_dbs(), and user_opts.

Referenced by transfer_all_new_tablespaces().

reap_child()

bool reap_child

(

bool

wait_for_child

)

Definition at line 281 of file parallel.c.

{
#ifndef WIN32
    int         work_status;
    pid_t       child;
#else
    int         thread_num;
    DWORD       res;
#endif
 
    if (user_opts.jobs <= 1 || parallel_jobs == 0)
        return false;
 
#ifndef WIN32
    child = waitpid(-1, &work_status, wait_for_child ? 0 : WNOHANG);
    if (child == (pid_t) -1)
        pg_fatal("%s() failed: %m", "waitpid");
    if (child == 0)
        return false;           /* no children, or no dead children */
    if (work_status != 0)
        pg_fatal("child process exited abnormally: status %d", work_status);
#else
    /* wait for one to finish */
    thread_num = WaitForMultipleObjects(parallel_jobs, thread_handles,
                                        false, wait_for_child ? INFINITE : 0);
 
    if (thread_num == WAIT_TIMEOUT || thread_num == WAIT_FAILED)
        return false;
 
    /* compute thread index in active_threads */
    thread_num -= WAIT_OBJECT_0;
 
    /* get the result */
    GetExitCodeThread(thread_handles[thread_num], &res);
    if (res != 0)
        pg_fatal("child worker exited abnormally: %m");
 
    /* dispose of handle to stop leaks */
    CloseHandle(thread_handles[thread_num]);
 
    /* Move last slot into dead child's position */
    if (thread_num != parallel_jobs - 1)
    {
        void       *tmp_args;
 
        thread_handles[thread_num] = thread_handles[parallel_jobs - 1];
 
        /*
         * Move last active thread arg struct into the now-dead slot, and the
         * now-dead slot to the end for reuse by the next thread. Though the
         * thread struct is in use by another thread, we can safely swap the
         * struct pointers within the array.
         */
        tmp_args = cur_thread_args[thread_num];
        cur_thread_args[thread_num] = cur_thread_args[parallel_jobs - 1];
        cur_thread_args[parallel_jobs - 1] = tmp_args;
    }
#endif
 
    /* do this after job has been removed */
    parallel_jobs--;
 
    return true;
}

References UserOpts::jobs, parallel_jobs, pg_fatal, and user_opts.

Referenced by create_new_objects(), generate_old_dump(), parallel_exec_prog(), parallel_transfer_all_new_dbs(), and transfer_all_new_tablespaces().

Variable Documentation

parallel_jobs

int parallel_jobs

static

Definition at line 19 of file parallel.c.

Referenced by parallel_exec_prog(), parallel_transfer_all_new_dbs(), and reap_child().