PostgreSQL Source Code  git master
worker_spi.c
Go to the documentation of this file.
1 /* -------------------------------------------------------------------------
2  *
3  * worker_spi.c
4  * Sample background worker code that demonstrates various coding
5  * patterns: establishing a database connection; starting and committing
6  * transactions; using GUC variables, and heeding SIGHUP to reread
7  * the configuration file; reporting to pg_stat_activity; using the
8  * process latch to sleep and exit in case of postmaster death.
9  *
10  * This code connects to a database, creates a schema and table, and summarizes
11  * the numbers contained therein. To see it working, insert an initial value
12  * with "total" type and some initial value; then insert some other rows with
13  * "delta" type. Delta rows will be deleted by this worker and their values
14  * aggregated into the total.
15  *
16  * Copyright (c) 2013-2022, PostgreSQL Global Development Group
17  *
18  * IDENTIFICATION
19  * src/test/modules/worker_spi/worker_spi.c
20  *
21  * -------------------------------------------------------------------------
22  */
23 #include "postgres.h"
24 
25 /* These are always necessary for a bgworker */
26 #include "miscadmin.h"
27 #include "postmaster/bgworker.h"
28 #include "postmaster/interrupt.h"
29 #include "storage/ipc.h"
30 #include "storage/latch.h"
31 #include "storage/lwlock.h"
32 #include "storage/proc.h"
33 #include "storage/shmem.h"
34 
35 /* these headers are used by this particular worker's code */
36 #include "access/xact.h"
37 #include "executor/spi.h"
38 #include "fmgr.h"
39 #include "lib/stringinfo.h"
40 #include "pgstat.h"
41 #include "utils/builtins.h"
42 #include "utils/snapmgr.h"
43 #include "tcop/utility.h"
44 
46 
48 
50 
51 /* GUC variables */
52 static int worker_spi_naptime = 10;
53 static int worker_spi_total_workers = 2;
54 static char *worker_spi_database = NULL;
55 
56 
57 typedef struct worktable
58 {
59  const char *schema;
60  const char *name;
62 
63 /*
64  * Initialize workspace for a worker process: create the schema if it doesn't
65  * already exist.
66  */
67 static void
69 {
70  int ret;
71  int ntup;
72  bool isnull;
74 
77  SPI_connect();
79  pgstat_report_activity(STATE_RUNNING, "initializing worker_spi schema");
80 
81  /* XXX could we use CREATE SCHEMA IF NOT EXISTS? */
83  appendStringInfo(&buf, "select count(*) from pg_namespace where nspname = '%s'",
84  table->schema);
85 
86  debug_query_string = buf.data;
87  ret = SPI_execute(buf.data, true, 0);
88  if (ret != SPI_OK_SELECT)
89  elog(FATAL, "SPI_execute failed: error code %d", ret);
90 
91  if (SPI_processed != 1)
92  elog(FATAL, "not a singleton result");
93 
96  1, &isnull));
97  if (isnull)
98  elog(FATAL, "null result");
99 
100  if (ntup == 0)
101  {
102  debug_query_string = NULL;
105  "CREATE SCHEMA \"%s\" "
106  "CREATE TABLE \"%s\" ("
107  " type text CHECK (type IN ('total', 'delta')), "
108  " value integer)"
109  "CREATE UNIQUE INDEX \"%s_unique_total\" ON \"%s\" (type) "
110  "WHERE type = 'total'",
111  table->schema, table->name, table->name, table->name);
112 
113  /* set statement start time */
115 
116  debug_query_string = buf.data;
117  ret = SPI_execute(buf.data, false, 0);
118 
119  if (ret != SPI_OK_UTILITY)
120  elog(FATAL, "failed to create my schema");
121 
122  debug_query_string = NULL; /* rest is not statement-specific */
123  }
124 
125  SPI_finish();
128  debug_query_string = NULL;
130 }
131 
132 void
133 worker_spi_main(Datum main_arg)
134 {
135  int index = DatumGetInt32(main_arg);
136  worktable *table;
138  char name[20];
139 
140  table = palloc(sizeof(worktable));
141  sprintf(name, "schema%d", index);
142  table->schema = pstrdup(name);
143  table->name = pstrdup("counted");
144 
145  /* Establish signal handlers before unblocking signals. */
147  pqsignal(SIGTERM, die);
148 
149  /* We're now ready to receive signals */
151 
152  /* Connect to our database */
153  BackgroundWorkerInitializeConnection(worker_spi_database, NULL, 0);
154 
155  elog(LOG, "%s initialized with %s.%s",
156  MyBgworkerEntry->bgw_name, table->schema, table->name);
157  initialize_worker_spi(table);
158 
159  /*
160  * Quote identifiers passed to us. Note that this must be done after
161  * initialize_worker_spi, because that routine assumes the names are not
162  * quoted.
163  *
164  * Note some memory might be leaked here.
165  */
166  table->schema = quote_identifier(table->schema);
167  table->name = quote_identifier(table->name);
168 
171  "WITH deleted AS (DELETE "
172  "FROM %s.%s "
173  "WHERE type = 'delta' RETURNING value), "
174  "total AS (SELECT coalesce(sum(value), 0) as sum "
175  "FROM deleted) "
176  "UPDATE %s.%s "
177  "SET value = %s.value + total.sum "
178  "FROM total WHERE type = 'total' "
179  "RETURNING %s.value",
180  table->schema, table->name,
181  table->schema, table->name,
182  table->name,
183  table->name);
184 
185  /*
186  * Main loop: do this until SIGTERM is received and processed by
187  * ProcessInterrupts.
188  */
189  for (;;)
190  {
191  int ret;
192 
193  /*
194  * Background workers mustn't call usleep() or any direct equivalent:
195  * instead, they may wait on their process latch, which sleeps as
196  * necessary, but is awakened if postmaster dies. That way the
197  * background process goes away immediately in an emergency.
198  */
199  (void) WaitLatch(MyLatch,
201  worker_spi_naptime * 1000L,
204 
206 
207  /*
208  * In case of a SIGHUP, just reload the configuration.
209  */
211  {
212  ConfigReloadPending = false;
214  }
215 
216  /*
217  * Start a transaction on which we can run queries. Note that each
218  * StartTransactionCommand() call should be preceded by a
219  * SetCurrentStatementStartTimestamp() call, which sets both the time
220  * for the statement we're about the run, and also the transaction
221  * start time. Also, each other query sent to SPI should probably be
222  * preceded by SetCurrentStatementStartTimestamp(), so that statement
223  * start time is always up to date.
224  *
225  * The SPI_connect() call lets us run queries through the SPI manager,
226  * and the PushActiveSnapshot() call creates an "active" snapshot
227  * which is necessary for queries to have MVCC data to work on.
228  *
229  * The pgstat_report_activity() call makes our activity visible
230  * through the pgstat views.
231  */
234  SPI_connect();
236  debug_query_string = buf.data;
238 
239  /* We can now execute queries via SPI */
240  ret = SPI_execute(buf.data, false, 0);
241 
242  if (ret != SPI_OK_UPDATE_RETURNING)
243  elog(FATAL, "cannot select from table %s.%s: error code %d",
244  table->schema, table->name, ret);
245 
246  if (SPI_processed > 0)
247  {
248  bool isnull;
249  int32 val;
250 
253  1, &isnull));
254  if (!isnull)
255  elog(LOG, "%s: count in %s.%s is now %d",
257  table->schema, table->name, val);
258  }
259 
260  /*
261  * And finish our transaction.
262  */
263  SPI_finish();
266  debug_query_string = NULL;
267  pgstat_report_stat(true);
269  }
270 
271  /* Not reachable */
272 }
273 
274 /*
275  * Entrypoint of this module.
276  *
277  * We register more than one worker process here, to demonstrate how that can
278  * be done.
279  */
280 void
281 _PG_init(void)
282 {
283  BackgroundWorker worker;
284 
285  /* get the configuration */
286  DefineCustomIntVariable("worker_spi.naptime",
287  "Duration between each check (in seconds).",
288  NULL,
289  &worker_spi_naptime,
290  10,
291  1,
292  INT_MAX,
293  PGC_SIGHUP,
294  0,
295  NULL,
296  NULL,
297  NULL);
298 
300  return;
301 
302  DefineCustomIntVariable("worker_spi.total_workers",
303  "Number of workers.",
304  NULL,
305  &worker_spi_total_workers,
306  2,
307  1,
308  100,
310  0,
311  NULL,
312  NULL,
313  NULL);
314 
315  DefineCustomStringVariable("worker_spi.database",
316  "Database to connect to.",
317  NULL,
318  &worker_spi_database,
319  "postgres",
321  0,
322  NULL, NULL, NULL);
323 
324  MarkGUCPrefixReserved("worker_spi");
325 
326  /* set up common data for all our workers */
327  memset(&worker, 0, sizeof(worker));
332  sprintf(worker.bgw_library_name, "worker_spi");
333  sprintf(worker.bgw_function_name, "worker_spi_main");
334  worker.bgw_notify_pid = 0;
335 
336  /*
337  * Now fill in worker-specific data, and do the actual registrations.
338  */
339  for (int i = 1; i <= worker_spi_total_workers; i++)
340  {
341  snprintf(worker.bgw_name, BGW_MAXLEN, "worker_spi worker %d", i);
342  snprintf(worker.bgw_type, BGW_MAXLEN, "worker_spi");
343  worker.bgw_main_arg = Int32GetDatum(i);
344 
345  RegisterBackgroundWorker(&worker);
346  }
347 }
348 
349 /*
350  * Dynamically launch an SPI worker.
351  */
352 Datum
354 {
355  int32 i = PG_GETARG_INT32(0);
356  BackgroundWorker worker;
357  BackgroundWorkerHandle *handle;
359  pid_t pid;
360 
361  memset(&worker, 0, sizeof(worker));
366  sprintf(worker.bgw_library_name, "worker_spi");
367  sprintf(worker.bgw_function_name, "worker_spi_main");
368  snprintf(worker.bgw_name, BGW_MAXLEN, "worker_spi worker %d", i);
369  snprintf(worker.bgw_type, BGW_MAXLEN, "worker_spi");
370  worker.bgw_main_arg = Int32GetDatum(i);
371  /* set bgw_notify_pid so that we can use WaitForBackgroundWorkerStartup */
372  worker.bgw_notify_pid = MyProcPid;
373 
374  if (!RegisterDynamicBackgroundWorker(&worker, &handle))
375  PG_RETURN_NULL();
376 
377  status = WaitForBackgroundWorkerStartup(handle, &pid);
378 
379  if (status == BGWH_STOPPED)
380  ereport(ERROR,
381  (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
382  errmsg("could not start background process"),
383  errhint("More details may be available in the server log.")));
385  ereport(ERROR,
386  (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
387  errmsg("cannot start background processes without postmaster"),
388  errhint("Kill all remaining database processes and restart the database.")));
390 
391  PG_RETURN_INT32(pid);
392 }
void pgstat_report_activity(BackendState state, const char *cmd_str)
@ STATE_IDLE
@ STATE_RUNNING
void RegisterBackgroundWorker(BackgroundWorker *worker)
Definition: bgworker.c:872
BgwHandleStatus WaitForBackgroundWorkerStartup(BackgroundWorkerHandle *handle, pid_t *pidp)
Definition: bgworker.c:1123
bool RegisterDynamicBackgroundWorker(BackgroundWorker *worker, BackgroundWorkerHandle **handle)
Definition: bgworker.c:956
#define BGW_NEVER_RESTART
Definition: bgworker.h:85
BgwHandleStatus
Definition: bgworker.h:104
@ BGWH_POSTMASTER_DIED
Definition: bgworker.h:108
@ BGWH_STARTED
Definition: bgworker.h:105
@ BGWH_STOPPED
Definition: bgworker.h:107
@ 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
signed int int32
Definition: c.h:430
#define pg_attribute_noreturn()
Definition: c.h:201
#define PGDLLEXPORT
Definition: c.h:1267
int errhint(const char *fmt,...)
Definition: elog.c:1153
int errcode(int sqlerrcode)
Definition: elog.c:695
int errmsg(const char *fmt,...)
Definition: elog.c:906
#define LOG
Definition: elog.h:27
#define FATAL
Definition: elog.h:37
#define ERROR
Definition: elog.h:35
#define ereport(elevel,...)
Definition: elog.h:145
const char * name
Definition: encode.c:561
#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
#define PG_FUNCTION_ARGS
Definition: fmgr.h:193
int MyProcPid
Definition: globals.c:44
struct Latch * MyLatch
Definition: globals.c:58
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:5026
void MarkGUCPrefixReserved(const char *className)
Definition: guc.c:5087
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:4966
@ PGC_POSTMASTER
Definition: guc.h:70
@ PGC_SIGHUP
Definition: guc.h:71
void ProcessConfigFile(GucContext context)
long val
Definition: informix.c:664
volatile sig_atomic_t ConfigReloadPending
Definition: interrupt.c:27
void SignalHandlerForConfigReload(SIGNAL_ARGS)
Definition: interrupt.c:61
int i
Definition: isn.c:73
void ResetLatch(Latch *latch)
Definition: latch.c:683
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:476
#define WL_TIMEOUT
Definition: latch.h:128
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:130
#define WL_LATCH_SET
Definition: latch.h:125
Assert(fmt[strlen(fmt) - 1] !='\n')
char * pstrdup(const char *in)
Definition: mcxt.c:1483
void * palloc(Size size)
Definition: mcxt.c:1199
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:121
bool process_shared_preload_libraries_in_progress
Definition: miscinit.c:1765
static void static void status(const char *fmt,...) pg_attribute_printf(1
Definition: pg_regress.c:225
#define die(msg)
Definition: pg_test_fsync.c:95
static char * buf
Definition: pg_test_fsync.c:67
long pgstat_report_stat(bool force)
Definition: pgstat.c:565
#define sprintf
Definition: port.h:240
pqsigfunc pqsignal(int signo, pqsigfunc func)
#define snprintf
Definition: port.h:238
const char * debug_query_string
Definition: postgres.c:82
static int64 DatumGetInt64(Datum X)
Definition: postgres.h:733
uintptr_t Datum
Definition: postgres.h:412
static Datum Int32GetDatum(int32 X)
Definition: postgres.h:560
static int32 DatumGetInt32(Datum X)
Definition: postgres.h:550
void BackgroundWorkerInitializeConnection(const char *dbname, const char *username, uint32 flags)
Definition: postmaster.c:5541
void BackgroundWorkerUnblockSignals(void)
Definition: postmaster.c:5601
BackgroundWorker * MyBgworkerEntry
Definition: postmaster.c:194
const char * quote_identifier(const char *ident)
Definition: ruleutils.c:11529
Snapshot GetTransactionSnapshot(void)
Definition: snapmgr.c:251
void PushActiveSnapshot(Snapshot snapshot)
Definition: snapmgr.c:683
void PopActiveSnapshot(void)
Definition: snapmgr.c:778
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_UPDATE_RETURNING
Definition: spi.h:94
#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
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
TupleDesc tupdesc
Definition: spi.h:25
HeapTuple * vals
Definition: spi.h:26
Definition: type.h:95
#define PG_WAIT_EXTENSION
Definition: wait_event.h:23
#define SIGHUP
Definition: win32_port.h:176
static void initialize_worker_spi(worktable *table)
Definition: worker_spi.c:68
void _PG_init(void)
Definition: worker_spi.c:281
PG_MODULE_MAGIC
Definition: worker_spi.c:45
Datum worker_spi_launch(PG_FUNCTION_ARGS)
Definition: worker_spi.c:353
PGDLLEXPORT void worker_spi_main(Datum)
Definition: worker_spi.c:49
worktable
Definition: worker_spi.c:61
PG_FUNCTION_INFO_V1(worker_spi_launch)
void StartTransactionCommand(void)
Definition: xact.c:2925
void SetCurrentStatementStartTimestamp(void)
Definition: xact.c:898
void CommitTransactionCommand(void)
Definition: xact.c:3022