PostgreSQL Source Code  git master
worker_spi.c File Reference
#include "postgres.h"
#include "miscadmin.h"
#include "postmaster/bgworker.h"
#include "postmaster/interrupt.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 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 282 of file worker_spi.c.

283 {
284  BackgroundWorker worker;
285 
286  /* get the configuration */
287  DefineCustomIntVariable("worker_spi.naptime",
288  "Duration between each check (in seconds).",
289  NULL,
290  &worker_spi_naptime,
291  10,
292  1,
293  INT_MAX,
294  PGC_SIGHUP,
295  0,
296  NULL,
297  NULL,
298  NULL);
299 
301  return;
302 
303  DefineCustomIntVariable("worker_spi.total_workers",
304  "Number of workers.",
305  NULL,
306  &worker_spi_total_workers,
307  2,
308  1,
309  100,
311  0,
312  NULL,
313  NULL,
314  NULL);
315 
316  DefineCustomStringVariable("worker_spi.database",
317  "Database to connect to.",
318  NULL,
319  &worker_spi_database,
320  "postgres",
322  0,
323  NULL, NULL, NULL);
324 
325  MarkGUCPrefixReserved("worker_spi");
326 
327  /* set up common data for all our workers */
328  memset(&worker, 0, sizeof(worker));
333  sprintf(worker.bgw_library_name, "worker_spi");
334  sprintf(worker.bgw_function_name, "worker_spi_main");
335  worker.bgw_notify_pid = 0;
336 
337  /*
338  * Now fill in worker-specific data, and do the actual registrations.
339  */
340  for (int i = 1; i <= worker_spi_total_workers; i++)
341  {
342  snprintf(worker.bgw_name, BGW_MAXLEN, "worker_spi worker %d", i);
343  snprintf(worker.bgw_type, BGW_MAXLEN, "worker_spi");
344  worker.bgw_main_arg = Int32GetDatum(i);
345 
346  RegisterBackgroundWorker(&worker);
347  }
348 }
void RegisterBackgroundWorker(BackgroundWorker *worker)
Definition: bgworker.c:872
#define BGW_NEVER_RESTART
Definition: bgworker.h:85
@ BgWorkerStart_RecoveryFinished
Definition: bgworker.h:81
#define BGWORKER_BACKEND_DATABASE_CONNECTION
Definition: bgworker.h:60
#define BGWORKER_SHMEM_ACCESS
Definition: bgworker.h:53
#define BGW_MAXLEN
Definition: bgworker.h:86
void DefineCustomStringVariable(const char *name, const char *short_desc, const char *long_desc, char **valueAddr, const char *bootValue, GucContext context, int flags, GucStringCheckHook check_hook, GucStringAssignHook assign_hook, GucShowHook show_hook)
Definition: guc.c:9562
void MarkGUCPrefixReserved(const char *className)
Definition: guc.c:9623
void DefineCustomIntVariable(const char *name, const char *short_desc, const char *long_desc, int *valueAddr, int bootValue, int minValue, int maxValue, GucContext context, int flags, GucIntCheckHook check_hook, GucIntAssignHook assign_hook, GucShowHook show_hook)
Definition: guc.c:9502
@ PGC_POSTMASTER
Definition: guc.h:71
@ PGC_SIGHUP
Definition: guc.h:72
int i
Definition: isn.c:73
bool process_shared_preload_libraries_in_progress
Definition: miscinit.c:1618
#define sprintf
Definition: port.h:227
#define snprintf
Definition: port.h:225
#define Int32GetDatum(X)
Definition: postgres.h:523
char bgw_function_name[BGW_MAXLEN]
Definition: bgworker.h:97
Datum bgw_main_arg
Definition: bgworker.h:98
char bgw_name[BGW_MAXLEN]
Definition: bgworker.h:91
int bgw_restart_time
Definition: bgworker.h:95
char bgw_type[BGW_MAXLEN]
Definition: bgworker.h:92
BgWorkerStartTime bgw_start_time
Definition: bgworker.h:94
pid_t bgw_notify_pid
Definition: bgworker.h:100
char bgw_library_name[BGW_MAXLEN]
Definition: bgworker.h:96

References 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, DefineCustomIntVariable(), DefineCustomStringVariable(), i, Int32GetDatum, MarkGUCPrefixReserved(), PGC_POSTMASTER, PGC_SIGHUP, process_shared_preload_libraries_in_progress, RegisterBackgroundWorker(), snprintf, and sprintf.

◆ initialize_worker_spi()

static void initialize_worker_spi ( worktable table)
static

Definition at line 69 of file worker_spi.c.

70 {
71  int ret;
72  int ntup;
73  bool isnull;
75 
78  SPI_connect();
80  pgstat_report_activity(STATE_RUNNING, "initializing worker_spi schema");
81 
82  /* XXX could we use CREATE SCHEMA IF NOT EXISTS? */
84  appendStringInfo(&buf, "select count(*) from pg_namespace where nspname = '%s'",
85  table->schema);
86 
87  debug_query_string = buf.data;
88  ret = SPI_execute(buf.data, true, 0);
89  if (ret != SPI_OK_SELECT)
90  elog(FATAL, "SPI_execute failed: error code %d", ret);
91 
92  if (SPI_processed != 1)
93  elog(FATAL, "not a singleton result");
94 
97  1, &isnull));
98  if (isnull)
99  elog(FATAL, "null result");
100 
101  if (ntup == 0)
102  {
103  debug_query_string = NULL;
106  "CREATE SCHEMA \"%s\" "
107  "CREATE TABLE \"%s\" ("
108  " type text CHECK (type IN ('total', 'delta')), "
109  " value integer)"
110  "CREATE UNIQUE INDEX \"%s_unique_total\" ON \"%s\" (type) "
111  "WHERE type = 'total'",
112  table->schema, table->name, table->name, table->name);
113 
114  /* set statement start time */
116 
117  debug_query_string = buf.data;
118  ret = SPI_execute(buf.data, false, 0);
119 
120  if (ret != SPI_OK_UTILITY)
121  elog(FATAL, "failed to create my schema");
122 
123  debug_query_string = NULL; /* rest is not statement-specific */
124  }
125 
126  SPI_finish();
129  debug_query_string = NULL;
131 }
void pgstat_report_activity(BackendState state, const char *cmd_str)
@ STATE_IDLE
@ STATE_RUNNING
#define FATAL
Definition: elog.h:35
#define elog(elevel,...)
Definition: elog.h:218
static char * buf
Definition: pg_test_fsync.c:67
const char * debug_query_string
Definition: postgres.c:89
#define DatumGetInt64(X)
Definition: postgres.h:651
Snapshot GetTransactionSnapshot(void)
Definition: snapmgr.c:250
void PopActiveSnapshot(void)
Definition: snapmgr.c:776
void PushActiveSnapshot(Snapshot snap)
Definition: snapmgr.c:682
uint64 SPI_processed
Definition: spi.c:45
SPITupleTable * SPI_tuptable
Definition: spi.c:46
int SPI_connect(void)
Definition: spi.c:95
int SPI_finish(void)
Definition: spi.c:183
int SPI_execute(const char *src, bool read_only, long tcount)
Definition: spi.c:594
Datum SPI_getbinval(HeapTuple tuple, TupleDesc tupdesc, int fnumber, bool *isnull)
Definition: spi.c:1250
#define SPI_OK_UTILITY
Definition: spi.h:85
#define SPI_OK_SELECT
Definition: spi.h:86
void resetStringInfo(StringInfo str)
Definition: stringinfo.c:75
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:91
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
TupleDesc tupdesc
Definition: spi.h:25
HeapTuple * vals
Definition: spi.h:26
void StartTransactionCommand(void)
Definition: xact.c:2925
void SetCurrentStatementStartTimestamp(void)
Definition: xact.c:886
void CommitTransactionCommand(void)
Definition: xact.c:3022

References appendStringInfo(), buf, CommitTransactionCommand(), DatumGetInt64, debug_query_string, elog, FATAL, GetTransactionSnapshot(), initStringInfo(), pgstat_report_activity(), PopActiveSnapshot(), PushActiveSnapshot(), resetStringInfo(), SetCurrentStatementStartTimestamp(), SPI_connect(), SPI_execute(), SPI_finish(), SPI_getbinval(), SPI_OK_SELECT, SPI_OK_UTILITY, SPI_processed, SPI_tuptable, StartTransactionCommand(), STATE_IDLE, STATE_RUNNING, SPITupleTable::tupdesc, and SPITupleTable::vals.

◆ PG_FUNCTION_INFO_V1()

PG_FUNCTION_INFO_V1 ( worker_spi_launch  )

◆ worker_spi_launch()

Datum worker_spi_launch ( PG_FUNCTION_ARGS  )

Definition at line 354 of file worker_spi.c.

355 {
356  int32 i = PG_GETARG_INT32(0);
357  BackgroundWorker worker;
358  BackgroundWorkerHandle *handle;
360  pid_t pid;
361 
362  memset(&worker, 0, sizeof(worker));
367  sprintf(worker.bgw_library_name, "worker_spi");
368  sprintf(worker.bgw_function_name, "worker_spi_main");
369  snprintf(worker.bgw_name, BGW_MAXLEN, "worker_spi worker %d", i);
370  snprintf(worker.bgw_type, BGW_MAXLEN, "worker_spi");
371  worker.bgw_main_arg = Int32GetDatum(i);
372  /* set bgw_notify_pid so that we can use WaitForBackgroundWorkerStartup */
373  worker.bgw_notify_pid = MyProcPid;
374 
375  if (!RegisterDynamicBackgroundWorker(&worker, &handle))
376  PG_RETURN_NULL();
377 
378  status = WaitForBackgroundWorkerStartup(handle, &pid);
379 
380  if (status == BGWH_STOPPED)
381  ereport(ERROR,
382  (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
383  errmsg("could not start background process"),
384  errhint("More details may be available in the server log.")));
386  ereport(ERROR,
387  (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
388  errmsg("cannot start background processes without postmaster"),
389  errhint("Kill all remaining database processes and restart the database.")));
391 
392  PG_RETURN_INT32(pid);
393 }
BgwHandleStatus WaitForBackgroundWorkerStartup(BackgroundWorkerHandle *handle, pid_t *pidp)
Definition: bgworker.c:1123
bool RegisterDynamicBackgroundWorker(BackgroundWorker *worker, BackgroundWorkerHandle **handle)
Definition: bgworker.c:956
BgwHandleStatus
Definition: bgworker.h:104
@ BGWH_POSTMASTER_DIED
Definition: bgworker.h:108
@ BGWH_STARTED
Definition: bgworker.h:105
@ BGWH_STOPPED
Definition: bgworker.h:107
signed int int32
Definition: c.h:429
int errhint(const char *fmt,...)
Definition: elog.c:1151
int errcode(int sqlerrcode)
Definition: elog.c:693
int errmsg(const char *fmt,...)
Definition: elog.c:904
#define ERROR
Definition: elog.h:33
#define ereport(elevel,...)
Definition: elog.h:143
#define PG_RETURN_NULL()
Definition: fmgr.h:345
#define PG_RETURN_INT32(x)
Definition: fmgr.h:354
#define PG_GETARG_INT32(n)
Definition: fmgr.h:269
int MyProcPid
Definition: globals.c:44
Assert(fmt[strlen(fmt) - 1] !='\n')
static void static void status(const char *fmt,...) pg_attribute_printf(1
Definition: pg_regress.c:229

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, sprintf, status(), and WaitForBackgroundWorkerStartup().

◆ worker_spi_main()

void worker_spi_main ( Datum  main_arg)

Definition at line 50 of file worker_spi.c.

59 {
60  const char *schema;
61  const char *name;
62 } worktable;
const char * name
Definition: encode.c:561
worktable
Definition: worker_spi.c:62

Variable Documentation

◆ PG_MODULE_MAGIC

PG_MODULE_MAGIC

Definition at line 45 of file worker_spi.c.

◆ worktable

worktable

Definition at line 62 of file worker_spi.c.