Commit 7f7485a0 authored by Robert Haas's avatar Robert Haas

Allow background workers to be started dynamically.

There is a new API, RegisterDynamicBackgroundWorker, which allows
an ordinary user backend to register a new background writer during
normal running.  This means that it's no longer necessary for all
background workers to be registered during processing of
shared_preload_libraries, although the option of registering workers
at that time remains available.

When a background worker exits and will not be restarted, the
slot previously used by that background worker is automatically
released and becomes available for reuse.  Slots used by background
workers that are configured for automatic restart can't (yet) be
released without shutting down the system.

This commit adds a new source file, bgworker.c, and moves some
of the existing control logic for background workers there.
Previously, there was little enough logic that it made sense to
keep everything in postmaster.c, but not any more.

This commit also makes the worker_spi contrib module into an
extension and adds a new function, worker_spi_launch, which can
be used to demonstrate the new facility.
parent 233bfe06
...@@ -2,6 +2,9 @@ ...@@ -2,6 +2,9 @@
MODULES = worker_spi MODULES = worker_spi
EXTENSION = worker_spi
DATA = worker_spi--1.0.sql
ifdef USE_PGXS ifdef USE_PGXS
PG_CONFIG = pg_config PG_CONFIG = pg_config
PGXS := $(shell $(PG_CONFIG) --pgxs) PGXS := $(shell $(PG_CONFIG) --pgxs)
......
/* contrib/worker_spi/worker_spi--1.0.sql */
-- complain if script is sourced in psql, rather than via CREATE EXTENSION
\echo Use "CREATE EXTENSION worker_spi" to load this file. \quit
CREATE FUNCTION worker_spi_launch(pg_catalog.int4)
RETURNS pg_catalog.bool STRICT
AS 'MODULE_PATHNAME'
LANGUAGE C;
...@@ -42,8 +42,11 @@ ...@@ -42,8 +42,11 @@
#include "tcop/utility.h" #include "tcop/utility.h"
PG_MODULE_MAGIC; PG_MODULE_MAGIC;
PG_FUNCTION_INFO_V1(worker_spi_launch);
void _PG_init(void); void _PG_init(void);
void worker_spi_main(Datum);
Datum worker_spi_launch(PG_FUNCTION_ARGS);
/* flags set by signal handlers */ /* flags set by signal handlers */
static volatile sig_atomic_t got_sighup = false; static volatile sig_atomic_t got_sighup = false;
...@@ -153,11 +156,22 @@ initialize_worker_spi(worktable *table) ...@@ -153,11 +156,22 @@ initialize_worker_spi(worktable *table)
pgstat_report_activity(STATE_IDLE, NULL); pgstat_report_activity(STATE_IDLE, NULL);
} }
static void void
worker_spi_main(void *main_arg) worker_spi_main(Datum main_arg)
{ {
worktable *table = (worktable *) main_arg; int index = DatumGetInt32(main_arg);
worktable *table;
StringInfoData buf; StringInfoData buf;
char name[20];
table = palloc(sizeof(worktable));
sprintf(name, "schema%d", index);
table->schema = pstrdup(name);
table->name = pstrdup("counted");
/* Establish signal handlers before unblocking signals. */
pqsignal(SIGHUP, worker_spi_sighup);
pqsignal(SIGTERM, worker_spi_sigterm);
/* We're now ready to receive signals */ /* We're now ready to receive signals */
BackgroundWorkerUnblockSignals(); BackgroundWorkerUnblockSignals();
...@@ -279,7 +293,7 @@ worker_spi_main(void *main_arg) ...@@ -279,7 +293,7 @@ worker_spi_main(void *main_arg)
pgstat_report_activity(STATE_IDLE, NULL); pgstat_report_activity(STATE_IDLE, NULL);
} }
proc_exit(0); proc_exit(1);
} }
/* /*
...@@ -292,9 +306,7 @@ void ...@@ -292,9 +306,7 @@ void
_PG_init(void) _PG_init(void)
{ {
BackgroundWorker worker; BackgroundWorker worker;
worktable *table;
unsigned int i; unsigned int i;
char name[20];
/* get the configuration */ /* get the configuration */
DefineCustomIntVariable("worker_spi.naptime", DefineCustomIntVariable("worker_spi.naptime",
...@@ -309,6 +321,10 @@ _PG_init(void) ...@@ -309,6 +321,10 @@ _PG_init(void)
NULL, NULL,
NULL, NULL,
NULL); NULL);
if (!process_shared_preload_libraries_in_progress)
return;
DefineCustomIntVariable("worker_spi.total_workers", DefineCustomIntVariable("worker_spi.total_workers",
"Number of workers.", "Number of workers.",
NULL, NULL,
...@@ -328,23 +344,41 @@ _PG_init(void) ...@@ -328,23 +344,41 @@ _PG_init(void)
worker.bgw_start_time = BgWorkerStart_RecoveryFinished; worker.bgw_start_time = BgWorkerStart_RecoveryFinished;
worker.bgw_restart_time = BGW_NEVER_RESTART; worker.bgw_restart_time = BGW_NEVER_RESTART;
worker.bgw_main = worker_spi_main; worker.bgw_main = worker_spi_main;
worker.bgw_sighup = worker_spi_sighup; worker.bgw_sighup = NULL;
worker.bgw_sigterm = worker_spi_sigterm; worker.bgw_sigterm = NULL;
/* /*
* Now fill in worker-specific data, and do the actual registrations. * Now fill in worker-specific data, and do the actual registrations.
*/ */
for (i = 1; i <= worker_spi_total_workers; i++) for (i = 1; i <= worker_spi_total_workers; i++)
{ {
sprintf(name, "worker %d", i); snprintf(worker.bgw_name, BGW_MAXLEN, "worker %d", i);
worker.bgw_name = pstrdup(name); worker.bgw_main_arg = Int32GetDatum(i);
table = palloc(sizeof(worktable));
sprintf(name, "schema%d", i);
table->schema = pstrdup(name);
table->name = pstrdup("counted");
worker.bgw_main_arg = (void *) table;
RegisterBackgroundWorker(&worker); RegisterBackgroundWorker(&worker);
} }
} }
/*
* Dynamically launch an SPI worker.
*/
Datum
worker_spi_launch(PG_FUNCTION_ARGS)
{
int32 i = PG_GETARG_INT32(0);
BackgroundWorker worker;
worker.bgw_flags = BGWORKER_SHMEM_ACCESS |
BGWORKER_BACKEND_DATABASE_CONNECTION;
worker.bgw_start_time = BgWorkerStart_RecoveryFinished;
worker.bgw_restart_time = BGW_NEVER_RESTART;
worker.bgw_main = NULL; /* new worker might not have library loaded */
sprintf(worker.bgw_library_name, "worker_spi");
sprintf(worker.bgw_function_name, "worker_spi_main");
worker.bgw_sighup = NULL; /* new worker might not have library loaded */
worker.bgw_sigterm = NULL; /* new worker might not have library loaded */
snprintf(worker.bgw_name, BGW_MAXLEN, "worker %d", i);
worker.bgw_main_arg = Int32GetDatum(i);
PG_RETURN_BOOL(RegisterDynamicBackgroundWorker(&worker));
}
# worker_spi extension
comment = 'Sample background worker'
default_version = '1.0'
module_pathname = '$libdir/worker_spi'
relocatable = true
...@@ -30,23 +30,35 @@ ...@@ -30,23 +30,35 @@
</warning> </warning>
<para> <para>
Only modules listed in <varname>shared_preload_libraries</> can run Background workers can be initialized at the time that
background workers. A module wishing to run a background worker needs <productname>PostgreSQL</> is started including the module name in
to register it by calling <varname>shared_preload_libraries</>. A module wishing to run a background
worker can register it by calling
<function>RegisterBackgroundWorker(<type>BackgroundWorker *worker</type>)</function> <function>RegisterBackgroundWorker(<type>BackgroundWorker *worker</type>)</function>
from its <function>_PG_init()</>. from its <function>_PG_init()</>. Background workers can also be started
after the system is up and running by calling the function
<function>RegisterDynamicBackgroundWorker</function>(<type>BackgroundWorker
*worker</type>). Unlike <function>RegisterBackgroundWorker</>, which can
only be called from within the postmaster,
<function>RegisterDynamicBackgroundWorker</function> must be called from
a regular backend.
</para>
<para>
The structure <structname>BackgroundWorker</structname> is defined thus: The structure <structname>BackgroundWorker</structname> is defined thus:
<programlisting> <programlisting>
typedef void (*bgworker_main_type)(void *main_arg); typedef void (*bgworker_main_type)(void *main_arg);
typedef void (*bgworker_sighdlr_type)(SIGNAL_ARGS); typedef void (*bgworker_sighdlr_type)(SIGNAL_ARGS);
typedef struct BackgroundWorker typedef struct BackgroundWorker
{ {
char *bgw_name; char bgw_name[BGW_MAXLEN];
int bgw_flags; int bgw_flags;
BgWorkerStartTime bgw_start_time; BgWorkerStartTime bgw_start_time;
int bgw_restart_time; /* in seconds, or BGW_NEVER_RESTART */ int bgw_restart_time; /* in seconds, or BGW_NEVER_RESTART */
bgworker_main_type bgw_main; bgworker_main_type bgw_main;
void *bgw_main_arg; char bgw_library_name[BGW_MAXLEN]; /* only if bgw_main is NULL */
char bgw_function_name[BGW_MAXLEN]; /* only if bgw_main is NULL */
Datum bgw_main_arg;
bgworker_sighdlr_type bgw_sighup; bgworker_sighdlr_type bgw_sighup;
bgworker_sighdlr_type bgw_sigterm; bgworker_sighdlr_type bgw_sigterm;
} BackgroundWorker; } BackgroundWorker;
...@@ -101,7 +113,29 @@ typedef struct BackgroundWorker ...@@ -101,7 +113,29 @@ typedef struct BackgroundWorker
<structfield>bgw_main_arg</structfield> will be passed to it as its only <structfield>bgw_main_arg</structfield> will be passed to it as its only
argument. Note that the global variable <literal>MyBgworkerEntry</literal> argument. Note that the global variable <literal>MyBgworkerEntry</literal>
points to a copy of the <structname>BackgroundWorker</structname> structure points to a copy of the <structname>BackgroundWorker</structname> structure
passed at registration time. passed at registration time. <structfield>bgw_main</structfield> may be
NULL; in that case, <structfield>bgw_library_name</structfield> and
<structfield>bgw_function_name</structfield> will be used to determine
the entrypoint. This is useful for background workers launched after
postmaster startup, where the postmaster does not have the requisite
library loaded.
</para>
<para>
<structfield>bgw_library_name</structfield> is the name of a library in
which the initial entrypoint for the background worker should be sought.
It is ignored unless <structfield>bgw_main</structfield> is NULL.
But if <structfield>bgw_main</structfield> is NULL, then the named library
will be dynamically loaded by the worker process and
<structfield>bgw_function_name</structfield> will be used to identify
the function to be called.
</para>
<para>
<structfield>bgw_function_name</structfield> is the name of a function in
a dynamically loaded library which should be used as the initial entrypoint
for a new background worker. It is ignored unless
<structfield>bgw_main</structfield> is NULL.
</para> </para>
<para> <para>
...@@ -109,7 +143,10 @@ typedef struct BackgroundWorker ...@@ -109,7 +143,10 @@ typedef struct BackgroundWorker
pointers to functions that will be installed as signal handlers for the new pointers to functions that will be installed as signal handlers for the new
process. If <structfield>bgw_sighup</> is NULL, then <literal>SIG_IGN</> process. If <structfield>bgw_sighup</> is NULL, then <literal>SIG_IGN</>
is used; if <structfield>bgw_sigterm</> is NULL, a handler is installed that is used; if <structfield>bgw_sigterm</> is NULL, a handler is installed that
will terminate the process after logging a suitable message. will terminate the process after logging a suitable message. These
fields should not be used if <structfield>bgw_main</> is NULL; instead,
the worker process should set its own signal handlers before calling
<function>BackgroundWorkerUnblockSignals()</function>.
</para> </para>
<para>Once running, the process can connect to a database by calling <para>Once running, the process can connect to a database by calling
......
...@@ -12,7 +12,7 @@ subdir = src/backend/postmaster ...@@ -12,7 +12,7 @@ subdir = src/backend/postmaster
top_builddir = ../../.. top_builddir = ../../..
include $(top_builddir)/src/Makefile.global include $(top_builddir)/src/Makefile.global
OBJS = autovacuum.o bgwriter.o fork_process.o pgarch.o pgstat.o postmaster.o \ OBJS = autovacuum.o bgworker.o bgwriter.o checkpointer.o fork_process.o \
startup.o syslogger.o walwriter.o checkpointer.o pgarch.o pgstat.o postmaster.o startup.o syslogger.o walwriter.o
include $(top_srcdir)/src/backend/common.mk include $(top_srcdir)/src/backend/common.mk
This diff is collapsed.
This diff is collapsed.
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include "miscadmin.h" #include "miscadmin.h"
#include "pgstat.h" #include "pgstat.h"
#include "postmaster/autovacuum.h" #include "postmaster/autovacuum.h"
#include "postmaster/bgworker_internals.h"
#include "postmaster/bgwriter.h" #include "postmaster/bgwriter.h"
#include "postmaster/postmaster.h" #include "postmaster/postmaster.h"
#include "replication/walreceiver.h" #include "replication/walreceiver.h"
...@@ -113,6 +114,7 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port) ...@@ -113,6 +114,7 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
size = add_size(size, CLOGShmemSize()); size = add_size(size, CLOGShmemSize());
size = add_size(size, SUBTRANSShmemSize()); size = add_size(size, SUBTRANSShmemSize());
size = add_size(size, TwoPhaseShmemSize()); size = add_size(size, TwoPhaseShmemSize());
size = add_size(size, BackgroundWorkerShmemSize());
size = add_size(size, MultiXactShmemSize()); size = add_size(size, MultiXactShmemSize());
size = add_size(size, LWLockShmemSize()); size = add_size(size, LWLockShmemSize());
size = add_size(size, ProcArrayShmemSize()); size = add_size(size, ProcArrayShmemSize());
...@@ -214,6 +216,7 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port) ...@@ -214,6 +216,7 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
CreateSharedProcArray(); CreateSharedProcArray();
CreateSharedBackendStatus(); CreateSharedBackendStatus();
TwoPhaseShmemInit(); TwoPhaseShmemInit();
BackgroundWorkerShmemInit();
/* /*
* Set up shared-inval messaging * Set up shared-inval messaging
......
...@@ -52,7 +52,7 @@ ...@@ -52,7 +52,7 @@
#define BGWORKER_BACKEND_DATABASE_CONNECTION 0x0002 #define BGWORKER_BACKEND_DATABASE_CONNECTION 0x0002
typedef void (*bgworker_main_type) (void *main_arg); typedef void (*bgworker_main_type) (Datum main_arg);
typedef void (*bgworker_sighdlr_type) (SIGNAL_ARGS); typedef void (*bgworker_sighdlr_type) (SIGNAL_ARGS);
/* /*
...@@ -67,22 +67,28 @@ typedef enum ...@@ -67,22 +67,28 @@ typedef enum
#define BGW_DEFAULT_RESTART_INTERVAL 60 #define BGW_DEFAULT_RESTART_INTERVAL 60
#define BGW_NEVER_RESTART -1 #define BGW_NEVER_RESTART -1
#define BGW_MAXLEN 64
typedef struct BackgroundWorker typedef struct BackgroundWorker
{ {
char *bgw_name; char bgw_name[BGW_MAXLEN];
int bgw_flags; int bgw_flags;
BgWorkerStartTime bgw_start_time; BgWorkerStartTime bgw_start_time;
int bgw_restart_time; /* in seconds, or BGW_NEVER_RESTART */ int bgw_restart_time; /* in seconds, or BGW_NEVER_RESTART */
bgworker_main_type bgw_main; bgworker_main_type bgw_main;
void *bgw_main_arg; char bgw_library_name[BGW_MAXLEN]; /* only if bgw_main is NULL */
char bgw_function_name[BGW_MAXLEN]; /* only if bgw_main is NULL */
Datum bgw_main_arg;
bgworker_sighdlr_type bgw_sighup; bgworker_sighdlr_type bgw_sighup;
bgworker_sighdlr_type bgw_sigterm; bgworker_sighdlr_type bgw_sigterm;
} BackgroundWorker; } BackgroundWorker;
/* Register a new bgworker */ /* Register a new bgworker during shared_preload_libraries */
extern void RegisterBackgroundWorker(BackgroundWorker *worker); extern void RegisterBackgroundWorker(BackgroundWorker *worker);
/* Register a new bgworker from a regular backend */
extern bool RegisterDynamicBackgroundWorker(BackgroundWorker *worker);
/* This is valid in a running worker */ /* This is valid in a running worker */
extern BackgroundWorker *MyBgworkerEntry; extern BackgroundWorker *MyBgworkerEntry;
......
/*--------------------------------------------------------------------
* bgworker_internals.h
* POSTGRES pluggable background workers internals
*
* Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
* IDENTIFICATION
* src/include/postmaster/bgworker.h
*--------------------------------------------------------------------
*/
#ifndef BGWORKER_INTERNALS_H
#define BGWORKER_INTERNALS_H
#include "datatype/timestamp.h"
#include "lib/ilist.h"
#include "postmaster/bgworker.h"
/*
* List of background workers, private to postmaster.
*
* A worker that requests a database connection during registration will have
* rw_backend set, and will be present in BackendList. Note: do not rely on
* rw_backend being non-NULL for shmem-connected workers!
*/
typedef struct RegisteredBgWorker
{
BackgroundWorker rw_worker; /* its registry entry */
struct bkend *rw_backend; /* its BackendList entry, or NULL */
pid_t rw_pid; /* 0 if not running */
int rw_child_slot;
TimestampTz rw_crashed_at; /* if not 0, time it last crashed */
int rw_shmem_slot;
slist_node rw_lnode; /* list link */
} RegisteredBgWorker;
extern slist_head BackgroundWorkerList;
extern Size BackgroundWorkerShmemSize(void);
extern void BackgroundWorkerShmemInit(void);
extern void BackgroundWorkerStateChange(void);
extern void ForgetBackgroundWorker(RegisteredBgWorker *);
#ifdef EXEC_BACKEND
extern BackgroundWorker *BackgroundWorkerEntry(int slotno);
#endif
#endif /* BGWORKER_INTERNLS_H */
...@@ -79,6 +79,7 @@ typedef enum LWLockId ...@@ -79,6 +79,7 @@ typedef enum LWLockId
SerializablePredicateLockListLock, SerializablePredicateLockListLock,
OldSerXidLock, OldSerXidLock,
SyncRepLock, SyncRepLock,
BackgroundWorkerLock,
/* Individual lock IDs end here */ /* Individual lock IDs end here */
FirstBufMappingLock, FirstBufMappingLock,
FirstLockMgrLock = FirstBufMappingLock + NUM_BUFFER_PARTITIONS, FirstLockMgrLock = FirstBufMappingLock + NUM_BUFFER_PARTITIONS,
......
...@@ -28,6 +28,7 @@ typedef enum ...@@ -28,6 +28,7 @@ typedef enum
PMSIGNAL_ROTATE_LOGFILE, /* send SIGUSR1 to syslogger to rotate logfile */ PMSIGNAL_ROTATE_LOGFILE, /* send SIGUSR1 to syslogger to rotate logfile */
PMSIGNAL_START_AUTOVAC_LAUNCHER, /* start an autovacuum launcher */ PMSIGNAL_START_AUTOVAC_LAUNCHER, /* start an autovacuum launcher */
PMSIGNAL_START_AUTOVAC_WORKER, /* start an autovacuum worker */ PMSIGNAL_START_AUTOVAC_WORKER, /* start an autovacuum worker */
PMSIGNAL_BACKGROUND_WORKER_CHANGE, /* background worker state change */
PMSIGNAL_START_WALRECEIVER, /* start a walreceiver */ PMSIGNAL_START_WALRECEIVER, /* start a walreceiver */
PMSIGNAL_ADVANCE_STATE_MACHINE, /* advance postmaster's state machine */ PMSIGNAL_ADVANCE_STATE_MACHINE, /* advance postmaster's state machine */
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment