parallel_slot.c File Reference

#include "postgres_fe.h"
#include <sys/select.h>
#include "common/logging.h"
#include "fe_utils/cancel.h"
#include "fe_utils/parallel_slot.h"
#include "fe_utils/query_utils.h"

Include dependency graph for parallel_slot.c:

Go to the source code of this file.

Macros
#define	ERRCODE_UNDEFINED_TABLE   "42P01"

Functions
static int	select_loop (int maxFd, fd_set *workerset)
static bool	processQueryResult (ParallelSlot *slot, PGresult *result)
static bool	consumeQueryResult (ParallelSlot *slot)
static int	find_matching_idle_slot (const ParallelSlotArray *sa, const char *dbname)
static int	find_unconnected_slot (const ParallelSlotArray *sa)
static int	find_any_idle_slot (const ParallelSlotArray *sa)
static bool	wait_on_slots (ParallelSlotArray *sa)
static void	connect_slot (ParallelSlotArray *sa, int slotno, const char *dbname)
ParallelSlot *	ParallelSlotsGetIdle (ParallelSlotArray *sa, const char *dbname)
ParallelSlotArray *	ParallelSlotsSetup (int numslots, ConnParams *cparams, const char *progname, bool echo, const char *initcmd)
void	ParallelSlotsAdoptConn (ParallelSlotArray *sa, PGconn *conn)
void	ParallelSlotsTerminate (ParallelSlotArray *sa)
bool	ParallelSlotsWaitCompletion (ParallelSlotArray *sa)
bool	TableCommandResultHandler (PGresult *res, PGconn *conn, void *context)

Macro Definition Documentation

ERRCODE_UNDEFINED_TABLE

#define ERRCODE_UNDEFINED_TABLE   "42P01"

Definition at line 28 of file parallel_slot.c.

Function Documentation

connect_slot()

static void connect_slot ( ParallelSlotArray *  sa, int  slotno, const char *  dbname  )

static

Definition at line 287 of file parallel_slot.c.

{
    const char *old_override;
    ParallelSlot *slot = &sa->slots[slotno];
 
    old_override = sa->cparams->override_dbname;
    if (dbname)
        sa->cparams->override_dbname = dbname;
    slot->connection = connectDatabase(sa->cparams, sa->progname, sa->echo, false, true);
    sa->cparams->override_dbname = old_override;
 
    /*
     * POSIX defines FD_SETSIZE as the highest file descriptor acceptable to
     * FD_SET() and allied macros.  Windows defines it as a ceiling on the
     * count of file descriptors in the set, not a ceiling on the value of
     * each file descriptor; see
     * https://learn.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-select
     * and
     * https://learn.microsoft.com/en-us/windows/win32/api/winsock/ns-winsock-fd_set.
     * We can't ignore that, because Windows starts file descriptors at a
     * higher value, delays reuse, and skips values.  With less than ten
     * concurrent file descriptors, opened and closed rapidly, one can reach
     * file descriptor 1024.
     *
     * Doing a hard exit here is a bit grotty, but it doesn't seem worth
     * complicating the API to make it less grotty.
     */
#ifdef WIN32
    if (slotno >= FD_SETSIZE)
    {
        pg_log_error("too many jobs for this platform: %d", slotno);
        exit(1);
    }
#else
    {
        int         fd = PQsocket(slot->connection);
 
        if (fd >= FD_SETSIZE)
        {
            pg_log_error("socket file descriptor out of range for select(): %d",
                         fd);
            pg_log_error_hint("Try fewer jobs.");
            exit(1);
        }
    }
#endif
 
    /* Setup the connection using the supplied command, if any. */
    if (sa->initcmd)
        executeCommand(slot->connection, sa->initcmd, sa->echo);
}

References connectDatabase(), ParallelSlot::connection, dbname, executeCommand(), fd(), pg_log_error, pg_log_error_hint, and PQsocket().

Referenced by ParallelSlotsGetIdle().

consumeQueryResult()

static bool consumeQueryResult ( ParallelSlot *  slot )

static

Definition at line 58 of file parallel_slot.c.

{
    bool        ok = true;
    PGresult   *result;
 
    SetCancelConn(slot->connection);
    while ((result = PQgetResult(slot->connection)) != NULL)
    {
        if (!processQueryResult(slot, result))
            ok = false;
    }
    ResetCancelConn();
    return ok;
}

References ParallelSlot::connection, PQgetResult, processQueryResult(), ResetCancelConn(), and SetCancelConn().

Referenced by ParallelSlotsWaitCompletion().

find_any_idle_slot()

static int find_any_idle_slot ( const ParallelSlotArray *  sa )

static

Definition at line 179 of file parallel_slot.c.

{
    int         i;
 
    for (i = 0; i < sa->numslots; i++)
        if (!sa->slots[i].inUse)
            return i;
 
    return -1;
}

References i.

Referenced by ParallelSlotsGetIdle().

find_matching_idle_slot()

static int find_matching_idle_slot ( const ParallelSlotArray *  sa, const char *  dbname  )

static

Definition at line 135 of file parallel_slot.c.

{
    int         i;
 
    for (i = 0; i < sa->numslots; i++)
    {
        if (sa->slots[i].inUse)
            continue;
 
        if (sa->slots[i].connection == NULL)
            continue;
 
        if (dbname == NULL ||
            strcmp(PQdb(sa->slots[i].connection), dbname) == 0)
            return i;
    }
    return -1;
}

References dbname, i, and PQdb().

Referenced by ParallelSlotsGetIdle().

find_unconnected_slot()

static int find_unconnected_slot ( const ParallelSlotArray *  sa )

static

Definition at line 159 of file parallel_slot.c.

{
    int         i;
 
    for (i = 0; i < sa->numslots; i++)
    {
        if (sa->slots[i].inUse)
            continue;
 
        if (sa->slots[i].connection == NULL)
            return i;
    }
 
    return -1;
}

References i.

Referenced by ParallelSlotsAdoptConn(), and ParallelSlotsGetIdle().

ParallelSlotsAdoptConn()

void ParallelSlotsAdoptConn	(	ParallelSlotArray *	sa,
		PGconn *	conn
	)

Definition at line 460 of file parallel_slot.c.

{
    int         offset;
 
    offset = find_unconnected_slot(sa);
    if (offset >= 0)
        sa->slots[offset].connection = conn;
    else
        disconnectDatabase(conn);
}

References conn, disconnectDatabase(), and find_unconnected_slot().

Referenced by main(), reindex_one_database(), and vacuum_one_database().

ParallelSlotsGetIdle()

ParallelSlot * ParallelSlotsGetIdle	(	ParallelSlotArray *	sa,
		const char *	dbname
	)

Definition at line 371 of file parallel_slot.c.

{
    int         offset;
 
    Assert(sa);
    Assert(sa->numslots > 0);
 
    while (1)
    {
        /* First choice: a slot already connected to the desired database. */
        offset = find_matching_idle_slot(sa, dbname);
        if (offset >= 0)
        {
            sa->slots[offset].inUse = true;
            return &sa->slots[offset];
        }
 
        /* Second choice: a slot not connected to any database. */
        offset = find_unconnected_slot(sa);
        if (offset >= 0)
        {
            connect_slot(sa, offset, dbname);
            sa->slots[offset].inUse = true;
            return &sa->slots[offset];
        }
 
        /* Third choice: a slot connected to the wrong database. */
        offset = find_any_idle_slot(sa);
        if (offset >= 0)
        {
            disconnectDatabase(sa->slots[offset].connection);
            sa->slots[offset].connection = NULL;
            connect_slot(sa, offset, dbname);
            sa->slots[offset].inUse = true;
            return &sa->slots[offset];
        }
 
        /*
         * Fourth choice: block until one or more slots become available. If
         * any slots hit a fatal error, we'll find out about that here and
         * return NULL.
         */
        if (!wait_on_slots(sa))
            return NULL;
    }
}

References Assert(), connect_slot(), dbname, disconnectDatabase(), find_any_idle_slot(), find_matching_idle_slot(), find_unconnected_slot(), and wait_on_slots().

Referenced by main(), reindex_one_database(), and vacuum_one_database().

ParallelSlotsSetup()

ParallelSlotArray * ParallelSlotsSetup	(	int	numslots,
		ConnParams *	cparams,
		const char *	progname,
		bool	echo,
		const char *	initcmd
	)

Definition at line 428 of file parallel_slot.c.

{
    ParallelSlotArray *sa;
 
    Assert(numslots > 0);
    Assert(cparams != NULL);
    Assert(progname != NULL);
 
    sa = (ParallelSlotArray *) palloc0(offsetof(ParallelSlotArray, slots) +
                                       numslots * sizeof(ParallelSlot));
 
    sa->numslots = numslots;
    sa->cparams = cparams;
    sa->progname = progname;
    sa->echo = echo;
    sa->initcmd = initcmd;
 
    return sa;
}

References Assert(), palloc0(), and progname.

Referenced by main(), reindex_one_database(), and vacuum_one_database().

ParallelSlotsTerminate()

void ParallelSlotsTerminate

(

ParallelSlotArray *

sa

)

Definition at line 479 of file parallel_slot.c.

{
    int         i;
 
    for (i = 0; i < sa->numslots; i++)
    {
        PGconn     *conn = sa->slots[i].connection;
 
        if (conn == NULL)
            continue;
 
        disconnectDatabase(conn);
    }
}

References conn, disconnectDatabase(), and i.

Referenced by main(), reindex_one_database(), and vacuum_one_database().

ParallelSlotsWaitCompletion()

bool ParallelSlotsWaitCompletion

(

ParallelSlotArray *

sa

)

Definition at line 501 of file parallel_slot.c.

{
    int         i;
 
    for (i = 0; i < sa->numslots; i++)
    {
        if (sa->slots[i].connection == NULL)
            continue;
        if (!consumeQueryResult(&sa->slots[i]))
            return false;
        /* Mark connection as idle */
        sa->slots[i].inUse = false;
        ParallelSlotClearHandler(&sa->slots[i]);
    }
 
    return true;
}

References consumeQueryResult(), i, and ParallelSlotClearHandler().

Referenced by main(), reindex_one_database(), and vacuum_one_database().

processQueryResult()

static bool processQueryResult ( ParallelSlot *  slot, PGresult *  result  )

static

Definition at line 39 of file parallel_slot.c.

{
    Assert(slot->handler != NULL);
 
    /* On failure, the handler should return NULL after freeing the result */
    if (!slot->handler(result, slot->connection, slot->handler_context))
        return false;
 
    /* Ok, we have to free it ourself */
    PQclear(result);
    return true;
}

References Assert(), ParallelSlot::connection, ParallelSlot::handler, ParallelSlot::handler_context, and PQclear.

Referenced by consumeQueryResult(), and wait_on_slots().

select_loop()

static int select_loop ( int  maxFd, fd_set *  workerset  )

static

Definition at line 80 of file parallel_slot.c.

{
    int         i;
    fd_set      saveSet = *workerset;
 
    if (CancelRequested)
        return -1;
 
    for (;;)
    {
        /*
         * On Windows, we need to check once in a while for cancel requests;
         * on other platforms we rely on select() returning when interrupted.
         */
        struct timeval *tvp;
#ifdef WIN32
        struct timeval tv = {0, 1000000};
 
        tvp = &tv;
#else
        tvp = NULL;
#endif
 
        *workerset = saveSet;
        i = select(maxFd + 1, workerset, NULL, NULL, tvp);
 
#ifdef WIN32
        if (i == SOCKET_ERROR)
        {
            i = -1;
 
            if (WSAGetLastError() == WSAEINTR)
                errno = EINTR;
        }
#endif
 
        if (i < 0 && errno == EINTR)
            continue;           /* ignore this */
        if (i < 0 || CancelRequested)
            return -1;          /* but not this */
        if (i == 0)
            continue;           /* timeout (Win32 only) */
        break;
    }
 
    return i;
}

References CancelRequested, EINTR, i, and select.

Referenced by wait_on_slots().

TableCommandResultHandler()

bool TableCommandResultHandler	(	PGresult *	res,
		PGconn *	conn,
		void *	context
	)

Definition at line 540 of file parallel_slot.c.

{
    Assert(res != NULL);
    Assert(conn != NULL);
 
    /*
     * If it's an error, report it.  Errors about a missing table are harmless
     * so we continue processing; but die for other errors.
     */
    if (PQresultStatus(res) != PGRES_COMMAND_OK)
    {
        char       *sqlState = PQresultErrorField(res, PG_DIAG_SQLSTATE);
 
        pg_log_error("processing of database \"%s\" failed: %s",
                     PQdb(conn), PQerrorMessage(conn));
 
        if (sqlState && strcmp(sqlState, ERRCODE_UNDEFINED_TABLE) != 0)
        {
            PQclear(res);
            return false;
        }
    }
 
    return true;
}

References Assert(), conn, ERRCODE_UNDEFINED_TABLE, PG_DIAG_SQLSTATE, pg_log_error, PGRES_COMMAND_OK, PQclear, PQdb(), PQerrorMessage(), PQresultErrorField, and PQresultStatus.

Referenced by reindex_one_database(), and vacuum_one_database().

wait_on_slots()

static bool wait_on_slots ( ParallelSlotArray *  sa )

static

Definition at line 196 of file parallel_slot.c.

{
    int         i;
    fd_set      slotset;
    int         maxFd = 0;
    PGconn     *cancelconn = NULL;
 
    /* We must reconstruct the fd_set for each call to select_loop */
    FD_ZERO(&slotset);
 
    for (i = 0; i < sa->numslots; i++)
    {
        int         sock;
 
        /* We shouldn't get here if we still have slots without connections */
        Assert(sa->slots[i].connection != NULL);
 
        sock = PQsocket(sa->slots[i].connection);
 
        /*
         * We don't really expect any connections to lose their sockets after
         * startup, but just in case, cope by ignoring them.
         */
        if (sock < 0)
            continue;
 
        /* Keep track of the first valid connection we see. */
        if (cancelconn == NULL)
            cancelconn = sa->slots[i].connection;
 
        FD_SET(sock, &slotset);
        if (sock > maxFd)
            maxFd = sock;
    }
 
    /*
     * If we get this far with no valid connections, processing cannot
     * continue.
     */
    if (cancelconn == NULL)
        return false;
 
    SetCancelConn(cancelconn);
    i = select_loop(maxFd, &slotset);
    ResetCancelConn();
 
    /* failure? */
    if (i < 0)
        return false;
 
    for (i = 0; i < sa->numslots; i++)
    {
        int         sock;
 
        sock = PQsocket(sa->slots[i].connection);
 
        if (sock >= 0 && FD_ISSET(sock, &slotset))
        {
            /* select() says input is available, so consume it */
            PQconsumeInput(sa->slots[i].connection);
        }
 
        /* Collect result(s) as long as any are available */
        while (!PQisBusy(sa->slots[i].connection))
        {
            PGresult   *result = PQgetResult(sa->slots[i].connection);
 
            if (result != NULL)
            {
                /* Handle and discard the command result */
                if (!processQueryResult(&sa->slots[i], result))
                    return false;
            }
            else
            {
                /* This connection has become idle */
                sa->slots[i].inUse = false;
                ParallelSlotClearHandler(&sa->slots[i]);
                break;
            }
        }
    }
    return true;
}

References Assert(), i, ParallelSlotClearHandler(), PQconsumeInput(), PQgetResult, PQisBusy(), PQsocket(), processQueryResult(), ResetCancelConn(), select_loop(), and SetCancelConn().

Referenced by ParallelSlotsGetIdle().