task.c File Reference

#include "postgres_fe.h"
#include "common/connect.h"
#include "fe_utils/string_utils.h"
#include "pg_upgrade.h"

Include dependency graph for task.c:

Go to the source code of this file.

Data Structures
struct	UpgradeTaskStep
struct	UpgradeTask
struct	UpgradeTaskSlot

Typedefs
typedef struct UpgradeTaskStep	UpgradeTaskStep
typedef enum UpgradeTaskSlotState	UpgradeTaskSlotState
typedef struct UpgradeTaskSlot	UpgradeTaskSlot

Enumerations
enum	UpgradeTaskSlotState { FREE , CONNECTING , RUNNING_QUERIES }

Functions
UpgradeTask *	upgrade_task_create (void)
void	upgrade_task_free (UpgradeTask *task)
void	upgrade_task_add_step (UpgradeTask *task, const char *query, UpgradeTaskProcessCB process_cb, bool free_result, void *arg)
static void	start_conn (const ClusterInfo *cluster, UpgradeTaskSlot *slot)
static void	process_query_result (const ClusterInfo *cluster, UpgradeTaskSlot *slot, const UpgradeTask *task)
static void	process_slot (const ClusterInfo *cluster, UpgradeTaskSlot *slot, const UpgradeTask *task)
static int	select_loop (int maxFd, fd_set *input, fd_set *output)
static void	wait_on_slots (UpgradeTaskSlot *slots, int numslots)
void	upgrade_task_run (const UpgradeTask *task, const ClusterInfo *cluster)

Variables
static int	dbs_complete
static int	dbs_processing

Typedef Documentation

UpgradeTaskSlot

typedef struct UpgradeTaskSlot UpgradeTaskSlot

UpgradeTaskSlotState

typedef enum UpgradeTaskSlotState UpgradeTaskSlotState

UpgradeTaskStep

typedef struct UpgradeTaskStep UpgradeTaskStep

Enumeration Type Documentation

UpgradeTaskSlotState

enum UpgradeTaskSlotState

Enumerator
FREE
CONNECTING
RUNNING_QUERIES

Definition at line 92 of file task.c.

{
    FREE,                       /* slot available for use in a new database */
    CONNECTING,                 /* waiting for connection to be established */
    RUNNING_QUERIES,            /* running/processing queries in the task */
} UpgradeTaskSlotState;

Function Documentation

process_query_result()

static void process_query_result ( const ClusterInfo *  cluster, UpgradeTaskSlot *  slot, const UpgradeTask *  task  )

static

Definition at line 206 of file task.c.

{
    UpgradeTaskStep *steps = &task->steps[slot->step_idx];
    UpgradeTaskProcessCB process_cb = steps->process_cb;
    DbInfo     *dbinfo = &cluster->dbarr.dbs[slot->db_idx];
    PGresult   *res = PQgetResult(slot->conn);
 
    if (PQstatus(slot->conn) == CONNECTION_BAD ||
        (PQresultStatus(res) != PGRES_TUPLES_OK &&
         PQresultStatus(res) != PGRES_COMMAND_OK))
        pg_fatal("connection failure: %s", PQerrorMessage(slot->conn));
 
    /*
     * We assume that a NULL process_cb callback function means there's
     * nothing to process.  This is primarily intended for the initial step in
     * every task that sets a safe search_path.
     */
    if (process_cb)
        (*process_cb) (dbinfo, res, steps->arg);
 
    if (steps->free_result)
        PQclear(res);
}

References UpgradeTaskStep::arg, cluster(), UpgradeTaskSlot::conn, CONNECTION_BAD, UpgradeTaskSlot::db_idx, dbinfo, UpgradeTaskStep::free_result, pg_fatal, PGRES_COMMAND_OK, PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQgetResult(), PQresultStatus(), PQstatus(), UpgradeTaskStep::process_cb, res, UpgradeTaskSlot::step_idx, and UpgradeTask::steps.

Referenced by process_slot().

process_slot()

static void process_slot ( const ClusterInfo *  cluster, UpgradeTaskSlot *  slot, const UpgradeTask *  task  )

static

Definition at line 235 of file task.c.

{
    PostgresPollingStatusType status;
 
    if (!slot->ready)
        return;
 
    switch (slot->state)
    {
        case FREE:
 
            /*
             * If all of the databases in the cluster have been processed or
             * are currently being processed by other slots, we are done.
             */
            if (dbs_processing >= cluster->dbarr.ndbs)
                return;
 
            /*
             * Claim the next database in the cluster's array and initiate a
             * new connection.
             */
            slot->db_idx = dbs_processing++;
            slot->state = CONNECTING;
            start_conn(cluster, slot);
 
            return;
 
        case CONNECTING:
 
            /* Check for connection failure. */
            status = PQconnectPoll(slot->conn);
            if (status == PGRES_POLLING_FAILED)
                pg_fatal("connection failure: %s", PQerrorMessage(slot->conn));
 
            /* Check whether the connection is still establishing. */
            if (status != PGRES_POLLING_OK)
            {
                slot->select_mode = (status == PGRES_POLLING_READING);
                return;
            }
 
            /*
             * Move on to running/processing the queries in the task.
             */
            slot->state = RUNNING_QUERIES;
            slot->select_mode = true;   /* wait until ready for reading */
            if (!PQsendQuery(slot->conn, task->queries->data))
                pg_fatal("connection failure: %s", PQerrorMessage(slot->conn));
 
            return;
 
        case RUNNING_QUERIES:
 
            /*
             * Consume any available data and clear the read-ready indicator
             * for the connection.
             */
            if (!PQconsumeInput(slot->conn))
                pg_fatal("connection failure: %s", PQerrorMessage(slot->conn));
 
            /*
             * Process any results that are ready so that we can free up this
             * slot for another database as soon as possible.
             */
            for (; slot->step_idx < task->num_steps; slot->step_idx++)
            {
                /* If no more results are available yet, move on. */
                if (PQisBusy(slot->conn))
                    return;
 
                process_query_result(cluster, slot, task);
            }
 
            /*
             * If we just finished processing the result of the last step in
             * the task, free the slot.  We recursively call this function on
             * the newly-freed slot so that we can start initiating the next
             * connection immediately instead of waiting for the next loop
             * through the slots.
             */
            dbs_complete++;
            PQfinish(slot->conn);
            memset(slot, 0, sizeof(UpgradeTaskSlot));
            slot->ready = true;
 
            process_slot(cluster, slot, task);
 
            return;
    }
}

References cluster(), UpgradeTaskSlot::conn, CONNECTING, PQExpBufferData::data, UpgradeTaskSlot::db_idx, dbs_complete, dbs_processing, FREE, UpgradeTask::num_steps, pg_fatal, PGRES_POLLING_FAILED, PGRES_POLLING_OK, PGRES_POLLING_READING, PQconnectPoll(), PQconsumeInput(), PQerrorMessage(), PQfinish(), PQisBusy(), PQsendQuery(), process_query_result(), UpgradeTask::queries, UpgradeTaskSlot::ready, RUNNING_QUERIES, UpgradeTaskSlot::select_mode, start_conn(), UpgradeTaskSlot::state, and UpgradeTaskSlot::step_idx.

Referenced by upgrade_task_run().

select_loop()

static int select_loop ( int  maxFd, fd_set *  input, fd_set *  output  )

static

Definition at line 331 of file task.c.

{
    fd_set      save_input = *input;
    fd_set      save_output = *output;
 
    if (maxFd == 0)
        return 0;
 
    for (;;)
    {
        int         i;
 
        *input = save_input;
        *output = save_output;
 
        i = select(maxFd + 1, input, output, NULL, NULL);
 
#ifndef WIN32
        if (i < 0 && errno == EINTR)
            continue;
#else
        if (i == SOCKET_ERROR && WSAGetLastError() == WSAEINTR)
            continue;
#endif
        return i;
    }
}

References EINTR, i, input, output, and select.

Referenced by wait_on_slots().

start_conn()

static void start_conn ( const ClusterInfo *  cluster, UpgradeTaskSlot *  slot  )

static

Definition at line 174 of file task.c.

{
    PQExpBufferData conn_opts;
    DbInfo     *dbinfo = &cluster->dbarr.dbs[slot->db_idx];
 
    /* Build connection string with proper quoting */
    initPQExpBuffer(&conn_opts);
    appendPQExpBufferStr(&conn_opts, "dbname=");
    appendConnStrVal(&conn_opts, dbinfo->db_name);
    appendPQExpBufferStr(&conn_opts, " user=");
    appendConnStrVal(&conn_opts, os_info.user);
    appendPQExpBuffer(&conn_opts, " port=%d", cluster->port);
    if (cluster->sockdir)
    {
        appendPQExpBufferStr(&conn_opts, " host=");
        appendConnStrVal(&conn_opts, cluster->sockdir);
    }
 
    slot->conn = PQconnectStart(conn_opts.data);
 
    if (!slot->conn)
        pg_fatal("failed to create connection with connection string: \"%s\"",
                 conn_opts.data);
 
    termPQExpBuffer(&conn_opts);
}

References appendConnStrVal(), appendPQExpBuffer(), appendPQExpBufferStr(), cluster(), UpgradeTaskSlot::conn, PQExpBufferData::data, UpgradeTaskSlot::db_idx, dbinfo, initPQExpBuffer(), os_info, pg_fatal, PQconnectStart(), termPQExpBuffer(), and OSInfo::user.

Referenced by process_slot().

upgrade_task_add_step()

void upgrade_task_add_step	(	UpgradeTask *	task,
		const char *	query,
		UpgradeTaskProcessCB	process_cb,
		bool	free_result,
		void *	arg
	)

Definition at line 151 of file task.c.

{
    UpgradeTaskStep *new_step;
 
    task->steps = pg_realloc(task->steps,
                             ++task->num_steps * sizeof(UpgradeTaskStep));
 
    new_step = &task->steps[task->num_steps - 1];
    new_step->process_cb = process_cb;
    new_step->free_result = free_result;
    new_step->arg = arg;
 
    appendPQExpBuffer(task->queries, "%s;", query);
}

References appendPQExpBuffer(), arg, UpgradeTaskStep::arg, UpgradeTaskStep::free_result, UpgradeTask::num_steps, pg_realloc(), UpgradeTaskStep::process_cb, UpgradeTask::queries, and UpgradeTask::steps.

Referenced by check_for_data_types_usage(), check_for_incompatible_polymorphics(), check_for_isn_and_int8_passing_mismatch(), check_for_tables_with_oids(), check_for_user_defined_encoding_conversions(), check_for_user_defined_postfix_ops(), check_old_cluster_subscription_state(), get_db_rel_and_slot_infos(), get_loadable_libraries(), report_extension_updates(), and upgrade_task_create().

upgrade_task_create()

UpgradeTask* upgrade_task_create

(

void

)

Definition at line 117 of file task.c.

{
    UpgradeTask *task = pg_malloc0(sizeof(UpgradeTask));
 
    task->queries = createPQExpBuffer();
 
    /* All tasks must first set a secure search_path. */
    upgrade_task_add_step(task, ALWAYS_SECURE_SEARCH_PATH_SQL, NULL, true, NULL);
 
    return task;
}

References ALWAYS_SECURE_SEARCH_PATH_SQL, createPQExpBuffer(), pg_malloc0(), UpgradeTask::queries, and upgrade_task_add_step().

Referenced by check_for_data_types_usage(), check_for_incompatible_polymorphics(), check_for_isn_and_int8_passing_mismatch(), check_for_tables_with_oids(), check_for_user_defined_encoding_conversions(), check_for_user_defined_postfix_ops(), check_old_cluster_subscription_state(), get_db_rel_and_slot_infos(), get_loadable_libraries(), and report_extension_updates().

upgrade_task_free()

void upgrade_task_free

(

UpgradeTask *

task

)

Definition at line 133 of file task.c.

{
    destroyPQExpBuffer(task->queries);
    pg_free(task->steps);
    pg_free(task);
}

References destroyPQExpBuffer(), pg_free(), UpgradeTask::queries, and UpgradeTask::steps.

Referenced by check_for_data_types_usage(), check_for_incompatible_polymorphics(), check_for_isn_and_int8_passing_mismatch(), check_for_tables_with_oids(), check_for_user_defined_encoding_conversions(), check_for_user_defined_postfix_ops(), check_old_cluster_subscription_state(), get_db_rel_and_slot_infos(), get_loadable_libraries(), and report_extension_updates().

upgrade_task_run()

void upgrade_task_run	(	const UpgradeTask *	task,
		const ClusterInfo *	cluster
	)

Definition at line 420 of file task.c.

{
    int         jobs = Max(1, user_opts.jobs);
    UpgradeTaskSlot *slots = pg_malloc0(sizeof(UpgradeTaskSlot) * jobs);
 
    dbs_complete = 0;
    dbs_processing = 0;
 
    /*
     * Process every slot the first time round.
     */
    for (int i = 0; i < jobs; i++)
        slots[i].ready = true;
 
    while (dbs_complete < cluster->dbarr.ndbs)
    {
        for (int i = 0; i < jobs; i++)
            process_slot(cluster, &slots[i], task);
 
        wait_on_slots(slots, jobs);
    }
 
    pg_free(slots);
}

References cluster(), dbs_complete, dbs_processing, i, UserOpts::jobs, Max, pg_free(), pg_malloc0(), process_slot(), user_opts, and wait_on_slots().

Referenced by check_for_data_types_usage(), check_for_incompatible_polymorphics(), check_for_isn_and_int8_passing_mismatch(), check_for_tables_with_oids(), check_for_user_defined_encoding_conversions(), check_for_user_defined_postfix_ops(), check_old_cluster_subscription_state(), get_db_rel_and_slot_infos(), get_loadable_libraries(), and report_extension_updates().

wait_on_slots()

static void wait_on_slots ( UpgradeTaskSlot *  slots, int  numslots  )

static

Definition at line 364 of file task.c.

{
    fd_set      input;
    fd_set      output;
    int         maxFd = 0;
 
    FD_ZERO(&input);
    FD_ZERO(&output);
 
    for (int i = 0; i < numslots; i++)
    {
        /*
         * We assume the previous call to process_slot() handled everything
         * that was marked ready in the previous call to wait_on_slots(), if
         * any.
         */
        slots[i].ready = false;
 
        /*
         * This function should only ever see free slots as we are finishing
         * processing the last few databases, at which point we don't have any
         * databases left for them to process.  We'll never use these slots
         * again, so we can safely ignore them.
         */
        if (slots[i].state == FREE)
            continue;
 
        /*
         * Add the socket to the set.
         */
        slots[i].sock = PQsocket(slots[i].conn);
        if (slots[i].sock < 0)
            pg_fatal("invalid socket");
        FD_SET(slots[i].sock, slots[i].select_mode ? &input : &output);
        maxFd = Max(maxFd, slots[i].sock);
    }
 
    /*
     * If we found socket(s) to wait on, wait.
     */
    if (select_loop(maxFd, &input, &output) == -1)
        pg_fatal("select() failed: %m");
 
    /*
     * Mark which sockets appear to be ready.
     */
    for (int i = 0; i < numslots; i++)
        slots[i].ready |= (FD_ISSET(slots[i].sock, &input) ||
                           FD_ISSET(slots[i].sock, &output));
}

References conn, FREE, i, input, Max, output, pg_fatal, PQsocket(), UpgradeTaskSlot::ready, select_loop(), and UpgradeTaskSlot::sock.

Referenced by upgrade_task_run().

Variable Documentation

dbs_complete

int dbs_complete

static

Definition at line 56 of file task.c.

Referenced by process_slot(), and upgrade_task_run().

dbs_processing

int dbs_processing

static

Definition at line 63 of file task.c.

Referenced by process_slot(), and upgrade_task_run().