worker_spi.c File Reference

#include "postgres.h"
#include "miscadmin.h"
#include "postmaster/bgworker.h"
#include "storage/ipc.h"
#include "storage/latch.h"
#include "storage/lwlock.h"
#include "storage/proc.h"
#include "storage/shmem.h"
#include "access/xact.h"
#include "executor/spi.h"
#include "fmgr.h"
#include "lib/stringinfo.h"
#include "pgstat.h"
#include "utils/builtins.h"
#include "utils/snapmgr.h"
#include "tcop/utility.h"

Include dependency graph for worker_spi.c:

Go to the source code of this file.

Functions
	PG_FUNCTION_INFO_V1 (worker_spi_launch)
void	_PG_init (void)
void	worker_spi_main (Datum)
static void	worker_spi_sigterm (SIGNAL_ARGS)
static void	worker_spi_sighup (SIGNAL_ARGS)
static void	initialize_worker_spi (worktable *table)
Datum	worker_spi_launch (PG_FUNCTION_ARGS)

Variables
	PG_MODULE_MAGIC
	worktable

Function Documentation

_PG_init()

void _PG_init

(

void

)

Definition at line 312 of file worker_spi.c.

References auth_delay_checks(), auth_delay_milliseconds, BackgroundWorker::bgw_flags, BackgroundWorker::bgw_function_name, BackgroundWorker::bgw_library_name, BackgroundWorker::bgw_main_arg, BGW_MAXLEN, BackgroundWorker::bgw_name, BGW_NEVER_RESTART, BackgroundWorker::bgw_notify_pid, BackgroundWorker::bgw_restart_time, BackgroundWorker::bgw_start_time, BackgroundWorker::bgw_type, BGWORKER_BACKEND_DATABASE_CONNECTION, BGWORKER_SHMEM_ACCESS, BgWorkerStart_RecoveryFinished, ClientAuthentication_hook, DefineCustomIntVariable(), GUC_UNIT_MS, i, Int32GetDatum, original_client_auth_hook, PGC_POSTMASTER, PGC_SIGHUP, process_shared_preload_libraries_in_progress, RegisterBackgroundWorker(), and snprintf().

{
    BackgroundWorker worker;
    unsigned int i;

    /* get the configuration */
    DefineCustomIntVariable("worker_spi.naptime",
                            "Duration between each check (in seconds).",
                            NULL,
                            &worker_spi_naptime,
                            10,
                            1,
                            INT_MAX,
                            PGC_SIGHUP,
                            0,
                            NULL,
                            NULL,
                            NULL);

    if (!process_shared_preload_libraries_in_progress)
        return;

    DefineCustomIntVariable("worker_spi.total_workers",
                            "Number of workers.",
                            NULL,
                            &worker_spi_total_workers,
                            2,
                            1,
                            100,
                            PGC_POSTMASTER,
                            0,
                            NULL,
                            NULL,
                            NULL);

    /* set up common data for all our workers */
    memset(&worker, 0, sizeof(worker));
    worker.bgw_flags = BGWORKER_SHMEM_ACCESS |
        BGWORKER_BACKEND_DATABASE_CONNECTION;
    worker.bgw_start_time = BgWorkerStart_RecoveryFinished;
    worker.bgw_restart_time = BGW_NEVER_RESTART;
    sprintf(worker.bgw_library_name, "worker_spi");
    sprintf(worker.bgw_function_name, "worker_spi_main");
    worker.bgw_notify_pid = 0;

    /*
     * Now fill in worker-specific data, and do the actual registrations.
     */
    for (i = 1; i <= worker_spi_total_workers; i++)
    {
        snprintf(worker.bgw_name, BGW_MAXLEN, "worker_spi worker %d", i);
        snprintf(worker.bgw_type, BGW_MAXLEN, "worker_spi");
        worker.bgw_main_arg = Int32GetDatum(i);

        RegisterBackgroundWorker(&worker);
    }
}

initialize_worker_spi()

static void initialize_worker_spi ( worktable *  table )

static

Definition at line 103 of file worker_spi.c.

References appendStringInfo(), BackgroundWorkerInitializeConnection(), BackgroundWorkerUnblockSignals(), BackgroundWorker::bgw_name, buf, CHECK_FOR_INTERRUPTS, CommitTransactionCommand(), StringInfoData::data, DatumGetInt32, DatumGetInt64, elog, FATAL, GetTransactionSnapshot(), got_sighup, got_sigterm, initStringInfo(), LOG, MyBgworkerEntry, MyLatch, name, palloc(), PG_WAIT_EXTENSION, PGC_SIGHUP, pgstat_report_activity(), pgstat_report_stat(), PopActiveSnapshot(), pqsignal(), proc_exit(), ProcessConfigFile(), pstrdup(), PushActiveSnapshot(), quote_identifier(), ResetLatch(), resetStringInfo(), SetCurrentStatementStartTimestamp(), SIGHUP, SPI_connect(), SPI_execute(), SPI_finish(), SPI_getbinval(), SPI_OK_SELECT, SPI_OK_UPDATE_RETURNING, SPI_OK_UTILITY, SPI_processed, SPI_tuptable, StartTransactionCommand(), STATE_IDLE, STATE_RUNNING, SPITupleTable::tupdesc, val, SPITupleTable::vals, WaitLatch(), WL_LATCH_SET, WL_POSTMASTER_DEATH, WL_TIMEOUT, worker_spi_main(), worker_spi_sighup(), worker_spi_sigterm(), and worktable.

{
    int         ret;
    int         ntup;
    bool        isnull;
    StringInfoData buf;

    SetCurrentStatementStartTimestamp();
    StartTransactionCommand();
    SPI_connect();
    PushActiveSnapshot(GetTransactionSnapshot());
    pgstat_report_activity(STATE_RUNNING, "initializing worker_spi schema");

    /* XXX could we use CREATE SCHEMA IF NOT EXISTS? */
    initStringInfo(&buf);
    appendStringInfo(&buf, "select count(*) from pg_namespace where nspname = '%s'",
                     table->schema);

    ret = SPI_execute(buf.data, true, 0);
    if (ret != SPI_OK_SELECT)
        elog(FATAL, "SPI_execute failed: error code %d", ret);

    if (SPI_processed != 1)
        elog(FATAL, "not a singleton result");

    ntup = DatumGetInt64(SPI_getbinval(SPI_tuptable->vals[0],
                                       SPI_tuptable->tupdesc,
                                       1, &isnull));
    if (isnull)
        elog(FATAL, "null result");

    if (ntup == 0)
    {
        resetStringInfo(&buf);
        appendStringInfo(&buf,
                         "CREATE SCHEMA \"%s\" "
                         "CREATE TABLE \"%s\" ("
                         "      type text CHECK (type IN ('total', 'delta')), "
                         "      value   integer)"
                         "CREATE UNIQUE INDEX \"%s_unique_total\" ON \"%s\" (type) "
                         "WHERE type = 'total'",
                         table->schema, table->name, table->name, table->name);

        /* set statement start time */
        SetCurrentStatementStartTimestamp();

        ret = SPI_execute(buf.data, false, 0);

        if (ret != SPI_OK_UTILITY)
            elog(FATAL, "failed to create my schema");
    }

    SPI_finish();
    PopActiveSnapshot();
    CommitTransactionCommand();
    pgstat_report_activity(STATE_IDLE, NULL);
}

PG_FUNCTION_INFO_V1()

PG_FUNCTION_INFO_V1

(

worker_spi_launch

)

worker_spi_launch()

Datum worker_spi_launch

(

PG_FUNCTION_ARGS

)

Definition at line 374 of file worker_spi.c.

References Assert, BackgroundWorker::bgw_flags, BackgroundWorker::bgw_function_name, BackgroundWorker::bgw_library_name, BackgroundWorker::bgw_main_arg, BGW_MAXLEN, BackgroundWorker::bgw_name, BGW_NEVER_RESTART, BackgroundWorker::bgw_notify_pid, BackgroundWorker::bgw_restart_time, BackgroundWorker::bgw_start_time, BackgroundWorker::bgw_type, BGWH_POSTMASTER_DIED, BGWH_STARTED, BGWH_STOPPED, BGWORKER_BACKEND_DATABASE_CONNECTION, BGWORKER_SHMEM_ACCESS, BgWorkerStart_RecoveryFinished, ereport, errcode(), errhint(), errmsg(), ERROR, i, Int32GetDatum, MyProcPid, PG_GETARG_INT32, PG_RETURN_INT32, PG_RETURN_NULL, RegisterDynamicBackgroundWorker(), snprintf(), status(), and WaitForBackgroundWorkerStartup().

{
    int32       i = PG_GETARG_INT32(0);
    BackgroundWorker worker;
    BackgroundWorkerHandle *handle;
    BgwHandleStatus status;
    pid_t       pid;

    memset(&worker, 0, sizeof(worker));
    worker.bgw_flags = BGWORKER_SHMEM_ACCESS |
        BGWORKER_BACKEND_DATABASE_CONNECTION;
    worker.bgw_start_time = BgWorkerStart_RecoveryFinished;
    worker.bgw_restart_time = BGW_NEVER_RESTART;
    sprintf(worker.bgw_library_name, "worker_spi");
    sprintf(worker.bgw_function_name, "worker_spi_main");
    snprintf(worker.bgw_name, BGW_MAXLEN, "worker_spi worker %d", i);
    snprintf(worker.bgw_type, BGW_MAXLEN, "worker_spi");
    worker.bgw_main_arg = Int32GetDatum(i);
    /* set bgw_notify_pid so that we can use WaitForBackgroundWorkerStartup */
    worker.bgw_notify_pid = MyProcPid;

    if (!RegisterDynamicBackgroundWorker(&worker, &handle))
        PG_RETURN_NULL();

    status = WaitForBackgroundWorkerStartup(handle, &pid);

    if (status == BGWH_STOPPED)
        ereport(ERROR,
                (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
                 errmsg("could not start background process"),
                 errhint("More details may be available in the server log.")));
    if (status == BGWH_POSTMASTER_DIED)
        ereport(ERROR,
                (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
                 errmsg("cannot start background processes without postmaster"),
                 errhint("Kill all remaining database processes and restart the database.")));
    Assert(status == BGWH_STARTED);

    PG_RETURN_INT32(pid);
}

worker_spi_main()

void worker_spi_main

(

Datum

main_arg

)

Definition at line 49 of file worker_spi.c.

Referenced by initialize_worker_spi().

{
    const char *schema;
    const char *name;
} worktable;

worker_spi_sighup()

static void worker_spi_sighup ( SIGNAL_ARGS  )

static

Definition at line 88 of file worker_spi.c.

References got_sighup, MyLatch, and SetLatch().

Referenced by initialize_worker_spi().

{
    int         save_errno = errno;

    got_sighup = true;
    SetLatch(MyLatch);

    errno = save_errno;
}

worker_spi_sigterm()

static void worker_spi_sigterm ( SIGNAL_ARGS  )

static

Definition at line 72 of file worker_spi.c.

References got_sigterm, MyLatch, and SetLatch().

Referenced by initialize_worker_spi().

{
    int         save_errno = errno;

    got_sigterm = true;
    SetLatch(MyLatch);

    errno = save_errno;
}

Variable Documentation

PG_MODULE_MAGIC

PG_MODULE_MAGIC

Definition at line 44 of file worker_spi.c.

worktable

worktable

Definition at line 64 of file worker_spi.c.

Referenced by initialize_worker_spi().