On Wed, Nov 12, 2014 at 11:36 AM, Robert Haas <robertmh...@gmail.com> wrote:
> On Wed, Nov 12, 2014 at 11:19 AM, Andres Freund <and...@2ndquadrant.com>
> wrote:
>> The question is whether the library is actually loaded in that case?
>> Because that normally only happens early during startup - which is why
>> it's a PGC_BACKEND guc.
>
> It looks like that does not work.
>
> [rhaas pgsql]$ PGOPTIONS='-c local_preload_libraries=auto_explain' psql
> psql (9.5devel)
> Type "help" for help.
>
> rhaas=# select * from pg_background_result(pg_background_launch('show
> auto_explain.log_min_duration')) as (x text);
> ERROR: unrecognized configuration parameter "auto_explain.log_min_duration"
> CONTEXT: background worker, pid 31316
>
> So, there's more to be done here. Rats.
It turned out to be quite simple to fix both problems. This
particular case fails because the call that loads the libraries
specified by session_preload_libraries and local_preload_libraries is
in PostgresMain() and thus never gets called by pg_background. I
fixed that by adding that call to pg_background in the appropriate
place. While I was at it, I added the REVOKE statements we discussed
earlier to pg_background's .sql file.
The other problem was due to this code in set_config_option:
/*
* If a PGC_BACKEND or PGC_SU_BACKEND
parameter is changed in
* the config file, we want to accept
the new value in the
* postmaster (whence it will propagate to
* subsequently-started backends), but
ignore it in existing
* backends. This is a tad klugy, but
necessary because we
* don't re-read the config file
during backend start.
*
* In EXEC_BACKEND builds, this works
differently: we load all
* nondefault settings from the
CONFIG_EXEC_PARAMS file during
* backend start. In that case we
must accept PGC_SIGHUP
* settings, so as to have the same
value as if we'd forked
* from the postmaster. We detect
this situation by checking
* IsInitProcessingMode, which is a
bit ugly, but it doesn't
* seem worth passing down an explicit
flag saying we're doing
* read_nondefault_variables().
*/
#ifdef EXEC_BACKEND
if (IsUnderPostmaster &&
!IsInitProcessingMode())
return -1;
#else
if (IsUnderPostmaster)
return -1;
#endif
When restoring variables via RestoreGUCState(), we need the same kind
of special-handling that we do when running in EXEC_BACKEND mode and
restoring variables via read_nondefault_variables(). Extending the
IsInitProcessingMode kludge() doesn't look appealing, so I instead
added the flag contemplated by the comment.
Updated patches attached.
--
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company
commit dc8980bdc9be89acf6d61bbfa5b1d7391992e8e8
Author: Robert Haas <rh...@postgresql.org>
Date: Thu Jul 17 07:58:32 2014 -0400
Add infrastructure to save and restore GUC values.
Amit Khandekar, Noah Misch, Robert Haas
V2: Fix misuse of NULL vs. NUL/zero-byte.
V2: Check return value of set_config_option.
V2: Add an explicit is_reload flag to set_config_option.
diff --git a/src/backend/commands/extension.c b/src/backend/commands/extension.c
index 9a0afa4..6692bb5 100644
--- a/src/backend/commands/extension.c
+++ b/src/backend/commands/extension.c
@@ -814,11 +814,11 @@ execute_extension_script(Oid extensionOid, ExtensionControlFile *control,
if (client_min_messages < WARNING)
(void) set_config_option("client_min_messages", "warning",
PGC_USERSET, PGC_S_SESSION,
- GUC_ACTION_SAVE, true, 0);
+ GUC_ACTION_SAVE, true, 0, false);
if (log_min_messages < WARNING)
(void) set_config_option("log_min_messages", "warning",
PGC_SUSET, PGC_S_SESSION,
- GUC_ACTION_SAVE, true, 0);
+ GUC_ACTION_SAVE, true, 0, false);
/*
* Set up the search path to contain the target schema, then the schemas
@@ -843,7 +843,7 @@ execute_extension_script(Oid extensionOid, ExtensionControlFile *control,
(void) set_config_option("search_path", pathbuf.data,
PGC_USERSET, PGC_S_SESSION,
- GUC_ACTION_SAVE, true, 0);
+ GUC_ACTION_SAVE, true, 0, false);
/*
* Set creating_extension and related variables so that
diff --git a/src/backend/utils/adt/ri_triggers.c b/src/backend/utils/adt/ri_triggers.c
index c0156fa..2f02303 100644
--- a/src/backend/utils/adt/ri_triggers.c
+++ b/src/backend/utils/adt/ri_triggers.c
@@ -2422,7 +2422,7 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
snprintf(workmembuf, sizeof(workmembuf), "%d", maintenance_work_mem);
(void) set_config_option("work_mem", workmembuf,
PGC_USERSET, PGC_S_SESSION,
- GUC_ACTION_SAVE, true, 0);
+ GUC_ACTION_SAVE, true, 0, false);
if (SPI_connect() != SPI_OK_CONNECT)
elog(ERROR, "SPI_connect failed");
diff --git a/src/backend/utils/misc/guc-file.l b/src/backend/utils/misc/guc-file.l
index bb9207a..6dca5df 100644
--- a/src/backend/utils/misc/guc-file.l
+++ b/src/backend/utils/misc/guc-file.l
@@ -318,7 +318,7 @@ ProcessConfigFile(GucContext context)
/* Now we can re-apply the wired-in default (i.e., the boot_val) */
if (set_config_option(gconf->name, NULL,
context, PGC_S_DEFAULT,
- GUC_ACTION_SET, true, 0) > 0)
+ GUC_ACTION_SET, true, 0, false) > 0)
{
/* Log the change if appropriate */
if (context == PGC_SIGHUP)
@@ -373,7 +373,7 @@ ProcessConfigFile(GucContext context)
scres = set_config_option(item->name, item->value,
context, PGC_S_FILE,
- GUC_ACTION_SET, true, 0);
+ GUC_ACTION_SET, true, 0, false);
if (scres > 0)
{
/* variable was updated, so log the change if appropriate */
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 6547dfa..f5a706e 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -110,6 +110,12 @@
#define S_PER_D (60 * 60 * 24)
#define MS_PER_D (1000 * 60 * 60 * 24)
+/*
+ * Precision with which REAL type guc values are to be printed for GUC
+ * serialization.
+ */
+#define REALTYPE_PRECISION 17
+
/* XXX these should appear in other modules' header files */
extern bool Log_disconnections;
extern int CommitDelay;
@@ -136,6 +142,10 @@ char *GUC_check_errmsg_string;
char *GUC_check_errdetail_string;
char *GUC_check_errhint_string;
+static void
+do_serialize(char **destptr, Size *maxbytes, const char *fmt,...)
+/* This lets gcc check the format string for consistency. */
+__attribute__((format(PG_PRINTF_ATTRIBUTE, 3, 4)));
static void set_config_sourcefile(const char *name, char *sourcefile,
int sourceline);
@@ -5612,7 +5622,8 @@ validate_conf_option(struct config_generic * record, const char *name,
int
set_config_option(const char *name, const char *value,
GucContext context, GucSource source,
- GucAction action, bool changeVal, int elevel)
+ GucAction action, bool changeVal, int elevel,
+ bool is_reload)
{
struct config_generic *record;
bool prohibitValueChange = false;
@@ -5726,18 +5737,13 @@ set_config_option(const char *name, const char *value,
* nondefault settings from the CONFIG_EXEC_PARAMS file during
* backend start. In that case we must accept PGC_SIGHUP
* settings, so as to have the same value as if we'd forked
- * from the postmaster. We detect this situation by checking
- * IsInitProcessingMode, which is a bit ugly, but it doesn't
- * seem worth passing down an explicit flag saying we're doing
- * read_nondefault_variables().
+ * from the postmaster. This can also happen when using
+ * RestoreGUCState() within a background worker that needs to
+ * have the same settings as the user backend that started it.
+ * is_reload will be true when either situation applies.
*/
-#ifdef EXEC_BACKEND
- if (IsUnderPostmaster && !IsInitProcessingMode())
- return -1;
-#else
- if (IsUnderPostmaster)
+ if (IsUnderPostmaster && !is_reload)
return -1;
-#endif
}
else if (context != PGC_POSTMASTER &&
context != PGC_BACKEND &&
@@ -6343,7 +6349,7 @@ SetConfigOption(const char *name, const char *value,
GucContext context, GucSource source)
{
(void) set_config_option(name, value, context, source,
- GUC_ACTION_SET, true, 0);
+ GUC_ACTION_SET, true, 0, false);
}
@@ -6923,9 +6929,7 @@ ExecSetVariableStmt(VariableSetStmt *stmt, bool isTopLevel)
ExtractSetVariableArgs(stmt),
(superuser() ? PGC_SUSET : PGC_USERSET),
PGC_S_SESSION,
- action,
- true,
- 0);
+ action, true, 0, false);
break;
case VAR_SET_MULTI:
@@ -7012,9 +7016,7 @@ ExecSetVariableStmt(VariableSetStmt *stmt, bool isTopLevel)
NULL,
(superuser() ? PGC_SUSET : PGC_USERSET),
PGC_S_SESSION,
- action,
- true,
- 0);
+ action, true, 0, false);
break;
case VAR_RESET_ALL:
ResetAllOptions();
@@ -7059,8 +7061,7 @@ SetPGVariable(const char *name, List *args, bool is_local)
(superuser() ? PGC_SUSET : PGC_USERSET),
PGC_S_SESSION,
is_local ? GUC_ACTION_LOCAL : GUC_ACTION_SET,
- true,
- 0);
+ true, 0, false);
}
/*
@@ -7103,8 +7104,7 @@ set_config_by_name(PG_FUNCTION_ARGS)
(superuser() ? PGC_SUSET : PGC_USERSET),
PGC_S_SESSION,
is_local ? GUC_ACTION_LOCAL : GUC_ACTION_SET,
- true,
- 0);
+ true, 0, false);
/* get the new current value */
new_value = GetConfigOptionByName(name, NULL);
@@ -7225,7 +7225,7 @@ define_custom_variable(struct config_generic * variable)
(void) set_config_option(name, pHolder->reset_val,
pHolder->gen.reset_scontext,
pHolder->gen.reset_source,
- GUC_ACTION_SET, true, WARNING);
+ GUC_ACTION_SET, true, WARNING, false);
/* That should not have resulted in stacking anything */
Assert(variable->stack == NULL);
@@ -7281,30 +7281,35 @@ reapply_stacked_values(struct config_generic * variable,
case GUC_SAVE:
(void) set_config_option(name, curvalue,
curscontext, cursource,
- GUC_ACTION_SAVE, true, WARNING);
+ GUC_ACTION_SAVE, true,
+ WARNING, false);
break;
case GUC_SET:
(void) set_config_option(name, curvalue,
curscontext, cursource,
- GUC_ACTION_SET, true, WARNING);
+ GUC_ACTION_SET, true,
+ WARNING, false);
break;
case GUC_LOCAL:
(void) set_config_option(name, curvalue,
curscontext, cursource,
- GUC_ACTION_LOCAL, true, WARNING);
+ GUC_ACTION_LOCAL, true,
+ WARNING, false);
break;
case GUC_SET_LOCAL:
/* first, apply the masked value as SET */
(void) set_config_option(name, stack->masked.val.stringval,
stack->masked_scontext, PGC_S_SESSION,
- GUC_ACTION_SET, true, WARNING);
+ GUC_ACTION_SET, true,
+ WARNING, false);
/* then apply the current value as LOCAL */
(void) set_config_option(name, curvalue,
curscontext, cursource,
- GUC_ACTION_LOCAL, true, WARNING);
+ GUC_ACTION_LOCAL, true,
+ WARNING, false);
break;
}
@@ -7328,7 +7333,7 @@ reapply_stacked_values(struct config_generic * variable,
{
(void) set_config_option(name, curvalue,
curscontext, cursource,
- GUC_ACTION_SET, true, WARNING);
+ GUC_ACTION_SET, true, WARNING, false);
variable->stack = NULL;
}
}
@@ -8452,7 +8457,7 @@ read_nondefault_variables(void)
(void) set_config_option(varname, varvalue,
varscontext, varsource,
- GUC_ACTION_SET, true, 0);
+ GUC_ACTION_SET, true, 0, true);
if (varsourcefile[0])
set_config_sourcefile(varname, varsourcefile, varsourceline);
@@ -8465,6 +8470,398 @@ read_nondefault_variables(void)
}
#endif /* EXEC_BACKEND */
+/*
+ * can_skip_gucvar:
+ * When serializing, determine whether to skip this GUC. When restoring, the
+ * negation of this test determines whether to restore the compiled-in default
+ * value before processing serialized values.
+ *
+ * A PGC_S_DEFAULT setting on the serialize side will typically match new
+ * postmaster children, but that can be false when got_SIGHUP == true and the
+ * pending configuration change modifies this setting. Nonetheless, we omit
+ * PGC_S_DEFAULT settings from serialization and make up for that by restoring
+ * defaults before applying serialized values.
+ *
+ * PGC_POSTMASTER variables always have the same value in every child of a
+ * particular postmaster. Most PGC_INTERNAL variables are compile-time
+ * constants; a few, like server_encoding and lc_ctype, are handled specially
+ * outside the serialize/restore procedure. Therefore, SerializeGUCState()
+ * never sends these, and and RestoreGUCState() never changes them.
+ */
+static bool
+can_skip_gucvar(struct config_generic * gconf)
+{
+ return gconf->context == PGC_POSTMASTER ||
+ gconf->context == PGC_INTERNAL || gconf->source == PGC_S_DEFAULT;
+}
+
+/*
+ * estimate_variable_size:
+ * Estimate max size for dumping the given GUC variable.
+ */
+static Size
+estimate_variable_size(struct config_generic * gconf)
+{
+ Size size;
+ Size valsize;
+
+ if (can_skip_gucvar(gconf))
+ return 0;
+
+ size = 0;
+
+ size = add_size(size, strlen(gconf->name) + 1);
+
+ /* Get the maximum display length of the GUC value. */
+ switch (gconf->vartype)
+ {
+ case PGC_BOOL:
+ {
+ valsize = 5; /* max(strlen('true'), strlen('false')) */
+ }
+ break;
+
+ case PGC_INT:
+ {
+ struct config_int *conf = (struct config_int *) gconf;
+
+ /*
+ * Instead of getting the exact display length, use max
+ * length. Also reduce the max length for typical ranges of
+ * small values. Maximum value is 2147483647, i.e. 10 chars.
+ * Include one byte for sign.
+ */
+ if (abs(*conf->variable) < 1000)
+ valsize = 3 + 1;
+ else
+ valsize = 10 + 1;
+ }
+ break;
+
+ case PGC_REAL:
+ {
+ /*
+ * We are going to print it with %.17g. Account for sign,
+ * decimal point, and e+nnn notation. E.g.
+ * -3.9932904234000002e+110
+ */
+ valsize = REALTYPE_PRECISION + 1 + 1 + 5;
+ }
+ break;
+
+ case PGC_STRING:
+ {
+ struct config_string *conf = (struct config_string *) gconf;
+
+ valsize = strlen(*conf->variable);
+ }
+ break;
+
+ case PGC_ENUM:
+ {
+ struct config_enum *conf = (struct config_enum *) gconf;
+
+ valsize = strlen(config_enum_lookup_by_value(conf, *conf->variable));
+ }
+ break;
+ }
+
+ /* Allow space for terminating zero-byte */
+ size = add_size(size, valsize + 1);
+
+ if (gconf->sourcefile)
+ size = add_size(size, strlen(gconf->sourcefile));
+
+ /* Allow space for terminating zero-byte */
+ size = add_size(size, 1);
+
+ /* Include line whenever we include file. */
+ if (gconf->sourcefile)
+ size = add_size(size, sizeof(gconf->sourceline));
+
+ size = add_size(size, sizeof(gconf->source));
+ size = add_size(size, sizeof(gconf->scontext));
+
+ return size;
+}
+
+/*
+ * EstimateGUCStateSpace:
+ * Returns the size needed to store the GUC state for the current process
+ */
+Size
+EstimateGUCStateSpace(void)
+{
+ Size size;
+ int i;
+
+ /* Add space reqd for saving the data size of the guc state */
+ size = sizeof(Size);
+
+ /* Add up the space needed for each GUC variable */
+ for (i = 0; i < num_guc_variables; i++)
+ size = add_size(size,
+ estimate_variable_size(guc_variables[i]));
+
+ return size;
+}
+
+/*
+ * do_serialize:
+ * Copies the formatted string into the destination. Moves ahead the
+ * destination pointer, and decrements the maxbytes by that many bytes. If
+ * maxbytes is not sufficient to copy the string, error out.
+ */
+static void
+do_serialize(char **destptr, Size *maxbytes, const char *fmt,...)
+{
+ va_list vargs;
+ int n;
+
+ if (*maxbytes <= 0)
+ elog(ERROR, "not enough space to serialize GUC state");
+
+ va_start(vargs, fmt);
+ n = vsnprintf(*destptr, *maxbytes, fmt, vargs);
+ va_end(vargs);
+
+ /*
+ * Cater to portability hazards in the vsnprintf() return value just like
+ * appendPQExpBufferVA() does. Note that this requires an extra byte of
+ * slack at the end of the buffer. Since serialize_variable() ends with a
+ * do_serialize_binary() rather than a do_serialize(), we'll always have
+ * that slack; estimate_variable_size() need not add a byte for it.
+ */
+ if (n < 0 || n >= *maxbytes - 1)
+ {
+ if (n < 0 && errno != 0 && errno != ENOMEM)
+ /* Shouldn't happen. Better show errno description. */
+ elog(ERROR, "vsnprintf failed: %m");
+ else
+ elog(ERROR, "not enough space to serialize GUC state");
+ }
+
+ /* Shift the destptr ahead of the null terminator */
+ *destptr += n + 1;
+ *maxbytes -= n + 1;
+}
+
+/* Binary copy version of do_serialize() */
+static void
+do_serialize_binary(char **destptr, Size *maxbytes, void *val, Size valsize)
+{
+ if (valsize > *maxbytes)
+ elog(ERROR, "not enough space to serialize GUC state");
+
+ memcpy(*destptr, val, valsize);
+ *destptr += valsize;
+ *maxbytes -= valsize;
+}
+
+/*
+ * serialize_variable:
+ * Dumps name, value and other information of a GUC variable into destptr.
+ */
+static void
+serialize_variable(char **destptr, Size *maxbytes,
+ struct config_generic * gconf)
+{
+ if (can_skip_gucvar(gconf))
+ return;
+
+ do_serialize(destptr, maxbytes, "%s", gconf->name);
+
+ switch (gconf->vartype)
+ {
+ case PGC_BOOL:
+ {
+ struct config_bool *conf = (struct config_bool *) gconf;
+
+ do_serialize(destptr, maxbytes,
+ (*conf->variable ? "true" : "false"));
+ }
+ break;
+
+ case PGC_INT:
+ {
+ struct config_int *conf = (struct config_int *) gconf;
+
+ do_serialize(destptr, maxbytes, "%d", *conf->variable);
+ }
+ break;
+
+ case PGC_REAL:
+ {
+ struct config_real *conf = (struct config_real *) gconf;
+
+ do_serialize(destptr, maxbytes, "%.*g",
+ REALTYPE_PRECISION, *conf->variable);
+ }
+ break;
+
+ case PGC_STRING:
+ {
+ struct config_string *conf = (struct config_string *) gconf;
+
+ do_serialize(destptr, maxbytes, "%s", *conf->variable);
+ }
+ break;
+
+ case PGC_ENUM:
+ {
+ struct config_enum *conf = (struct config_enum *) gconf;
+
+ do_serialize(destptr, maxbytes, "%s",
+ config_enum_lookup_by_value(conf, *conf->variable));
+ }
+ break;
+ }
+
+ do_serialize(destptr, maxbytes, "%s",
+ (gconf->sourcefile ? gconf->sourcefile : ""));
+
+ if (gconf->sourcefile)
+ do_serialize_binary(destptr, maxbytes, &gconf->sourceline,
+ sizeof(gconf->sourceline));
+
+ do_serialize_binary(destptr, maxbytes, &gconf->source,
+ sizeof(gconf->source));
+ do_serialize_binary(destptr, maxbytes, &gconf->scontext,
+ sizeof(gconf->scontext));
+}
+
+/*
+ * SerializeGUCState:
+ * Dumps the complete GUC state onto the memory location at start_address.
+ */
+void
+SerializeGUCState(Size maxsize, char *start_address)
+{
+ char *curptr;
+ Size actual_size;
+ Size bytes_left;
+ int i;
+ int i_role;
+
+ /* Reserve space for saving the actual size of the guc state */
+ curptr = start_address + sizeof(actual_size);
+ bytes_left = maxsize - sizeof(actual_size);
+
+ for (i = 0; i < num_guc_variables; i++)
+ {
+ /*
+ * It's pretty ugly, but we've got to force "role" to be initialized
+ * after "session_authorization"; otherwise, the latter will override
+ * the former.
+ */
+ if (strcmp(guc_variables[i]->name, "role") == 0)
+ i_role = i;
+ else
+ serialize_variable(&curptr, &bytes_left, guc_variables[i]);
+ }
+ serialize_variable(&curptr, &bytes_left, guc_variables[i_role]);
+
+ /* Store actual size without assuming alignment of start_address. */
+ actual_size = maxsize - bytes_left - sizeof(actual_size);
+ memcpy(start_address, &actual_size, sizeof(actual_size));
+}
+
+/*
+ * read_gucstate:
+ * Actually it does not read anything, just returns the srcptr. But it does
+ * move the srcptr past the terminating zero byte, so that the caller is ready
+ * to read the next string.
+ */
+static char *
+read_gucstate(char **srcptr, char *srcend)
+{
+ char *retptr = *srcptr;
+ char *ptr;
+
+ if (*srcptr >= srcend)
+ elog(ERROR, "incomplete GUC state");
+
+ /* The string variables are all null terminated */
+ for (ptr = *srcptr; ptr < srcend && *ptr != '\0'; ptr++)
+ ;
+
+ if (ptr > srcend)
+ elog(ERROR, "could not find null terminator in GUC state");
+
+ /* Set the new position to the byte following the terminating NUL */
+ *srcptr = ptr + 1;
+
+ return retptr;
+}
+
+/* Binary read version of read_gucstate(). Copies into dest */
+static void
+read_gucstate_binary(char **srcptr, char *srcend, void *dest, Size size)
+{
+ if (*srcptr + size > srcend)
+ elog(ERROR, "incomplete GUC state");
+
+ memcpy(dest, *srcptr, size);
+ *srcptr += size;
+}
+
+/*
+ * RestoreGUCState:
+ * Reads the GUC state at the specified address and updates the GUCs with the
+ * values read from the GUC state.
+ */
+void
+RestoreGUCState(void *gucstate)
+{
+ char *varname,
+ *varvalue,
+ *varsourcefile;
+ int varsourceline;
+ GucSource varsource;
+ GucContext varscontext;
+ char *srcptr = (char *) gucstate;
+ char *srcend;
+ Size len;
+ int i;
+
+ /* See comment at can_skip_gucvar(). */
+ for (i = 0; i < num_guc_variables; i++)
+ if (!can_skip_gucvar(guc_variables[i]))
+ InitializeOneGUCOption(guc_variables[i]);
+
+ /* First item is the length of the subsequent data */
+ memcpy(&len, gucstate, sizeof(len));
+
+ srcptr += sizeof(len);
+ srcend = srcptr + len;
+
+ while (srcptr < srcend)
+ {
+ int result;
+
+ if ((varname = read_gucstate(&srcptr, srcend)) == NULL)
+ break;
+
+ varvalue = read_gucstate(&srcptr, srcend);
+ varsourcefile = read_gucstate(&srcptr, srcend);
+ if (varsourcefile[0])
+ read_gucstate_binary(&srcptr, srcend,
+ &varsourceline, sizeof(varsourceline));
+ read_gucstate_binary(&srcptr, srcend,
+ &varsource, sizeof(varsource));
+ read_gucstate_binary(&srcptr, srcend,
+ &varscontext, sizeof(varscontext));
+
+ result = set_config_option(varname, varvalue, varscontext, varsource,
+ GUC_ACTION_SET, true, ERROR, true);
+ if (result <= 0)
+ ereport(ERROR,
+ (errcode(ERRCODE_INTERNAL_ERROR),
+ errmsg("parameter \"%s\" could not be set", varname)));
+ if (varsourcefile[0])
+ set_config_sourcefile(varname, varsourcefile, varsourceline);
+ }
+}
/*
* A little "long argument" simulation, although not quite GNU
@@ -8555,7 +8952,7 @@ ProcessGUCArray(ArrayType *array,
(void) set_config_option(name, value,
context, source,
- action, true, 0);
+ action, true, 0, false);
free(name);
if (value)
@@ -8858,7 +9255,7 @@ validate_option_array_item(const char *name, const char *value,
/* test for permissions and valid option value */
(void) set_config_option(name, value,
superuser() ? PGC_SUSET : PGC_USERSET,
- PGC_S_TEST, GUC_ACTION_SET, false, 0);
+ PGC_S_TEST, GUC_ACTION_SET, false, 0, false);
return true;
}
diff --git a/src/include/utils/guc.h b/src/include/utils/guc.h
index 2b2aaf4..395858b 100644
--- a/src/include/utils/guc.h
+++ b/src/include/utils/guc.h
@@ -339,7 +339,8 @@ extern bool parse_int(const char *value, int *result, int flags,
extern bool parse_real(const char *value, double *result);
extern int set_config_option(const char *name, const char *value,
GucContext context, GucSource source,
- GucAction action, bool changeVal, int elevel);
+ GucAction action, bool changeVal, int elevel,
+ bool is_reload);
extern void AlterSystemSetConfigFile(AlterSystemStmt *setstmt);
extern char *GetConfigOptionByName(const char *name, const char **varname);
extern void GetConfigOptionByNum(int varnum, const char **values, bool *noshow);
@@ -363,6 +364,11 @@ extern void write_nondefault_variables(GucContext context);
extern void read_nondefault_variables(void);
#endif
+/* GUC serialization */
+extern Size EstimateGUCStateSpace(void);
+extern void SerializeGUCState(Size maxsize, char *start_address);
+extern void RestoreGUCState(void *gucstate);
+
/* Support for messages reported from GUC check hooks */
extern PGDLLIMPORT char *GUC_check_errmsg_string;
commit be835602577c179a1adf6562c3391266465db71a
Author: Robert Haas <rh...@postgresql.org>
Date: Fri Jul 11 09:53:40 2014 -0400
pg_background: Run commands in a background worker, and get the results.
The currently-active GUC values from the user session will be copied
to the background worker. If the command returns a result set, you
can retrieve the result set; if not, you can retrieve the command
tags. If the command fails with an error, the same error will be
thrown in the launching process when the results are retrieved.
Warnings and other messages generated by the background worker, and
notifications received by it, are also propagated to the foreground
process.
Patch by me; review by Amit Kapila and Andres Freund.
V2: Refactor to reduce differences with exec_simple_query; other
cleanups per Amit.
V3: Per Andres, fix whitespace damage and use NameData instead of
char[NAMEDATALEN].
V4: dsm_(un)keep_mapping -> dsm_(un)pin_mapping. Error context:
background worker, pid %d. Permissions check: user must have
permission to read results.
V5: Revoke permissions on pg_background functions from public.
Add missing call to process_session_preload_libraries().
diff --git a/contrib/Makefile b/contrib/Makefile
index b37d0dd..11d6116 100644
--- a/contrib/Makefile
+++ b/contrib/Makefile
@@ -30,6 +30,7 @@ SUBDIRS = \
pageinspect \
passwordcheck \
pg_archivecleanup \
+ pg_background \
pg_buffercache \
pg_freespacemap \
pg_prewarm \
diff --git a/contrib/pg_background/Makefile b/contrib/pg_background/Makefile
new file mode 100644
index 0000000..c4e717d
--- /dev/null
+++ b/contrib/pg_background/Makefile
@@ -0,0 +1,18 @@
+# contrib/pg_background/Makefile
+
+MODULE_big = pg_background
+OBJS = pg_background.o
+
+EXTENSION = pg_background
+DATA = pg_background--1.0.sql
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = contrib/pg_background
+top_builddir = ../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/contrib/pg_background/pg_background--1.0.sql b/contrib/pg_background/pg_background--1.0.sql
new file mode 100644
index 0000000..8ac804a
--- /dev/null
+++ b/contrib/pg_background/pg_background--1.0.sql
@@ -0,0 +1,24 @@
+/* contrib/pg_background/pg_background--1.0.sql */
+
+-- complain if script is sourced in psql, rather than via CREATE EXTENSION
+\echo Use "CREATE EXTENSION pg_background" to load this file. \quit
+
+CREATE FUNCTION pg_background_launch(sql pg_catalog.text,
+ queue_size pg_catalog.int4 DEFAULT 65536)
+ RETURNS pg_catalog.int4 STRICT
+ AS 'MODULE_PATHNAME' LANGUAGE C;
+
+CREATE FUNCTION pg_background_result(pid pg_catalog.int4)
+ RETURNS SETOF pg_catalog.record STRICT
+ AS 'MODULE_PATHNAME' LANGUAGE C;
+
+CREATE FUNCTION pg_background_detach(pid pg_catalog.int4)
+ RETURNS pg_catalog.void STRICT
+ AS 'MODULE_PATHNAME' LANGUAGE C;
+
+REVOKE ALL ON FUNCTION pg_background_launch(pg_catalog.text, pg_catalog.int4)
+ FROM public;
+REVOKE ALL ON FUNCTION pg_background_result(pg_catalog.int4)
+ FROM public;
+REVOKE ALL ON FUNCTION pg_background_detach(pg_catalog.int4)
+ FROM public;
diff --git a/contrib/pg_background/pg_background.c b/contrib/pg_background/pg_background.c
new file mode 100644
index 0000000..c04de26
--- /dev/null
+++ b/contrib/pg_background/pg_background.c
@@ -0,0 +1,1000 @@
+/*--------------------------------------------------------------------------
+ *
+ * pg_background.c
+ * Run SQL commands using a background worker.
+ *
+ * Copyright (C) 2014, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * contrib/pg_background/pg_background.c
+ *
+ * -------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "fmgr.h"
+
+#include "access/htup_details.h"
+#include "access/printtup.h"
+#include "access/xact.h"
+#include "catalog/pg_type.h"
+#include "commands/async.h"
+#include "commands/dbcommands.h"
+#include "funcapi.h"
+#include "libpq/libpq.h"
+#include "libpq/pqformat.h"
+#include "libpq/pqmq.h"
+#include "miscadmin.h"
+#include "parser/analyze.h"
+#include "pgstat.h"
+#include "storage/dsm.h"
+#include "storage/ipc.h"
+#include "storage/shm_mq.h"
+#include "storage/shm_toc.h"
+#include "tcop/pquery.h"
+#include "tcop/utility.h"
+#include "utils/acl.h"
+#include "utils/builtins.h"
+#include "utils/lsyscache.h"
+#include "utils/memutils.h"
+#include "utils/ps_status.h"
+#include "utils/snapmgr.h"
+#include "utils/timeout.h"
+
+/* Table-of-contents constants for our dynamic shared memory segment. */
+#define PG_BACKGROUND_MAGIC 0x50674267
+#define PG_BACKGROUND_KEY_FIXED_DATA 0
+#define PG_BACKGROUND_KEY_SQL 1
+#define PG_BACKGROUND_KEY_GUC 2
+#define PG_BACKGROUND_KEY_QUEUE 3
+#define PG_BACKGROUND_NKEYS 4
+
+/* Fixed-size data passed via our dynamic shared memory segment. */
+typedef struct pg_background_fixed_data
+{
+ Oid database_id;
+ Oid authenticated_user_id;
+ Oid current_user_id;
+ int sec_context;
+ NameData database;
+ NameData authenticated_user;
+} pg_background_fixed_data;
+
+/* Private state maintained by the launching backend for IPC. */
+typedef struct pg_background_worker_info
+{
+ pid_t pid;
+ Oid current_user_id;
+ dsm_segment *seg;
+ BackgroundWorkerHandle *handle;
+ shm_mq_handle *responseq;
+ bool consumed;
+} pg_background_worker_info;
+
+/* Private state maintained across calls to pg_background_result. */
+typedef struct pg_background_result_state
+{
+ pg_background_worker_info *info;
+ FmgrInfo *receive_functions;
+ Oid *typioparams;
+ bool has_row_description;
+ List *command_tags;
+ bool complete;
+} pg_background_result_state;
+
+static HTAB *worker_hash;
+
+static void cleanup_worker_info(dsm_segment *, Datum pid_datum);
+static pg_background_worker_info *find_worker_info(pid_t pid);
+static void check_rights(pg_background_worker_info *info);
+static void save_worker_info(pid_t pid, dsm_segment *seg,
+ BackgroundWorkerHandle *handle,
+ shm_mq_handle *responseq);
+static void pg_background_error_callback(void *arg);
+
+static HeapTuple form_result_tuple(pg_background_result_state *state,
+ TupleDesc tupdesc, StringInfo msg);
+
+static void handle_sigterm(SIGNAL_ARGS);
+static void execute_sql_string(const char *sql);
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(pg_background_launch);
+PG_FUNCTION_INFO_V1(pg_background_result);
+PG_FUNCTION_INFO_V1(pg_background_detach);
+
+void pg_background_worker_main(Datum);
+
+/*
+ * Start a dynamic background worker to run a user-specified SQL command.
+ */
+Datum
+pg_background_launch(PG_FUNCTION_ARGS)
+{
+ text *sql = PG_GETARG_TEXT_PP(0);
+ int32 queue_size = PG_GETARG_INT32(1);
+ int32 sql_len = VARSIZE_ANY_EXHDR(sql);
+ Size guc_len;
+ Size segsize;
+ dsm_segment *seg;
+ shm_toc_estimator e;
+ shm_toc *toc;
+ char *sqlp;
+ char *gucstate;
+ shm_mq *mq;
+ BackgroundWorker worker;
+ BackgroundWorkerHandle *worker_handle;
+ pg_background_fixed_data *fdata;
+ pid_t pid;
+ shm_mq_handle *responseq;
+ MemoryContext oldcontext;
+
+ /* Ensure a valid queue size. */
+ if (queue_size < 0 || ((uint64) queue_size) < shm_mq_minimum_size)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("queue size must be at least %zu bytes",
+ shm_mq_minimum_size)));
+
+ /* Create dynamic shared memory and table of contents. */
+ shm_toc_initialize_estimator(&e);
+ shm_toc_estimate_chunk(&e, sizeof(pg_background_fixed_data));
+ shm_toc_estimate_chunk(&e, sql_len + 1);
+ guc_len = EstimateGUCStateSpace();
+ shm_toc_estimate_chunk(&e, guc_len);
+ shm_toc_estimate_chunk(&e, (Size) queue_size);
+ shm_toc_estimate_keys(&e, PG_BACKGROUND_NKEYS);
+ segsize = shm_toc_estimate(&e);
+ seg = dsm_create(segsize);
+ toc = shm_toc_create(PG_BACKGROUND_MAGIC, dsm_segment_address(seg),
+ segsize);
+
+ /* Store fixed-size data in dynamic shared memory. */
+ fdata = shm_toc_allocate(toc, sizeof(pg_background_fixed_data));
+ fdata->database_id = MyDatabaseId;
+ fdata->authenticated_user_id = GetAuthenticatedUserId();
+ GetUserIdAndSecContext(&fdata->current_user_id, &fdata->sec_context);
+ namestrcpy(&fdata->database, get_database_name(MyDatabaseId));
+ namestrcpy(&fdata->authenticated_user,
+ GetUserNameFromId(fdata->authenticated_user_id));
+ shm_toc_insert(toc, PG_BACKGROUND_KEY_FIXED_DATA, fdata);
+
+ /* Store SQL query in dynamic shared memory. */
+ sqlp = shm_toc_allocate(toc, sql_len + 1);
+ memcpy(sqlp, VARDATA(sql), sql_len);
+ sqlp[sql_len] = '\0';
+ shm_toc_insert(toc, PG_BACKGROUND_KEY_SQL, sqlp);
+
+ /* Store GUC state in dynamic shared memory. */
+ gucstate = shm_toc_allocate(toc, guc_len);
+ SerializeGUCState(guc_len, gucstate);
+ shm_toc_insert(toc, PG_BACKGROUND_KEY_GUC, gucstate);
+
+ /* Establish message queue in dynamic shared memory. */
+ mq = shm_mq_create(shm_toc_allocate(toc, (Size) queue_size),
+ (Size) queue_size);
+ shm_toc_insert(toc, PG_BACKGROUND_KEY_QUEUE, mq);
+ shm_mq_set_receiver(mq, MyProc);
+
+ /*
+ * Attach the queue before launching a worker, so that we'll automatically
+ * detach the queue if we error out. (Otherwise, the worker might sit
+ * there trying to write the queue long after we've gone away.)
+ */
+ oldcontext = MemoryContextSwitchTo(TopMemoryContext);
+ responseq = shm_mq_attach(mq, seg, NULL);
+ MemoryContextSwitchTo(oldcontext);
+
+ /* Configure a worker. */
+ worker.bgw_flags =
+ BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION;
+ worker.bgw_start_time = BgWorkerStart_ConsistentState;
+ worker.bgw_restart_time = BGW_NEVER_RESTART;
+ worker.bgw_main = NULL; /* new worker might not have library loaded */
+ sprintf(worker.bgw_library_name, "pg_background");
+ sprintf(worker.bgw_function_name, "pg_background_worker_main");
+ snprintf(worker.bgw_name, BGW_MAXLEN,
+ "pg_background by PID %d", MyProcPid);
+ worker.bgw_main_arg = UInt32GetDatum(dsm_segment_handle(seg));
+ /* set bgw_notify_pid, so we can detect if the worker stops */
+ worker.bgw_notify_pid = MyProcPid;
+
+ /*
+ * Register the worker.
+ *
+ * We switch contexts so that the background worker handle can outlast
+ * this transaction.
+ */
+ oldcontext = MemoryContextSwitchTo(TopMemoryContext);
+ if (!RegisterDynamicBackgroundWorker(&worker, &worker_handle))
+ ereport(ERROR,
+ (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
+ errmsg("could not register background process"),
+ errhint("You may need to increase max_worker_processes.")));
+ MemoryContextSwitchTo(oldcontext);
+ shm_mq_set_handle(responseq, worker_handle);
+
+ /* Wait for the worker to start. */
+ switch (WaitForBackgroundWorkerStartup(worker_handle, &pid))
+ {
+ case BGWH_STARTED:
+ /* Success. */
+ break;
+ case BGWH_STOPPED:
+ pfree(worker_handle);
+ ereport(ERROR,
+ (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
+ errmsg("could not start background process"),
+ errhint("More details may be available in the server log.")));
+ break;
+ case BGWH_POSTMASTER_DIED:
+ pfree(worker_handle);
+ ereport(ERROR,
+ (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
+ errmsg("cannot start background processes without postmaster"),
+ errhint("Kill all remaining database processes and restart the database.")));
+ break;
+ default:
+ elog(ERROR, "unexpected bgworker handle status");
+ break;
+ }
+
+ /* Store the relevant details about this worker for future use. */
+ save_worker_info(pid, seg, worker_handle, responseq);
+
+ /*
+ * Now that the worker info is saved, we do not need to, and should not,
+ * automatically detach the segment at resource-owner cleanup time.
+ */
+ dsm_pin_mapping(seg);
+
+ /* Return the worker's PID. */
+ PG_RETURN_INT32(pid);
+}
+
+/*
+ * Retrieve the results of a background query previously launched in this
+ * session.
+ */
+Datum
+pg_background_result(PG_FUNCTION_ARGS)
+{
+ int32 pid = PG_GETARG_INT32(0);
+ shm_mq_result res;
+ FuncCallContext *funcctx;
+ TupleDesc tupdesc;
+ StringInfoData msg;
+ pg_background_result_state *state;
+
+ /* First-time setup. */
+ if (SRF_IS_FIRSTCALL())
+ {
+ MemoryContext oldcontext;
+ pg_background_worker_info *info;
+
+ funcctx = SRF_FIRSTCALL_INIT();
+ oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+ /* See if we have a connection to the specified PID. */
+ if ((info = find_worker_info(pid)) == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_UNDEFINED_OBJECT),
+ errmsg("PID %d is not attached to this session", pid)));
+ check_rights(info);
+
+ /* Can't read results twice. */
+ if (info->consumed)
+ ereport(ERROR,
+ (errcode(ERRCODE_UNDEFINED_OBJECT),
+ errmsg("results for PID %d have already been consumed", pid)));
+ info->consumed = true;
+
+ /*
+ * Whether we succeed or fail, a future invocation of this function
+ * may not try to read from the DSM once we've begun to do so.
+ * Accordingly, make arrangements to clean things up at end of query.
+ */
+ dsm_unpin_mapping(info->seg);
+
+ /* Set up tuple-descriptor based on colum definition list. */
+ if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("function returning record called in context "
+ "that cannot accept type record"),
+ errhint("Try calling the function in the FROM clause "
+ "using a column definition list.")));
+ funcctx->tuple_desc = BlessTupleDesc(tupdesc);
+
+ /* Cache state that will be needed on every call. */
+ state = palloc0(sizeof(pg_background_result_state));
+ state->info = info;
+ if (funcctx->tuple_desc->natts > 0)
+ {
+ int natts = funcctx->tuple_desc->natts;
+ int i;
+
+ state->receive_functions = palloc(sizeof(FmgrInfo) * natts);
+ state->typioparams = palloc(sizeof(Oid) * natts);
+
+ for (i = 0; i < natts; ++i)
+ {
+ Oid receive_function_id;
+
+ getTypeBinaryInputInfo(funcctx->tuple_desc->attrs[i]->atttypid,
+ &receive_function_id,
+ &state->typioparams[i]);
+ fmgr_info(receive_function_id, &state->receive_functions[i]);
+ }
+ }
+ funcctx->user_fctx = state;
+
+ MemoryContextSwitchTo(oldcontext);
+ }
+ funcctx = SRF_PERCALL_SETUP();
+ tupdesc = funcctx->tuple_desc;
+ state = funcctx->user_fctx;
+
+ /* Initialize message buffer. */
+ initStringInfo(&msg);
+
+ /* Read and processes messages from the shared memory queue. */
+ for (;;)
+ {
+ char msgtype;
+ Size nbytes;
+ void *data;
+
+ /* Get next message. */
+ res = shm_mq_receive(state->info->responseq, &nbytes, &data, false);
+ if (res != SHM_MQ_SUCCESS)
+ break;
+
+ /*
+ * Message-parsing routines operate on a null-terminated StringInfo,
+ * so we must construct one.
+ */
+ resetStringInfo(&msg);
+ enlargeStringInfo(&msg, nbytes);
+ msg.len = nbytes;
+ memcpy(msg.data, data, nbytes);
+ msg.data[nbytes] = '\0';
+ msgtype = pq_getmsgbyte(&msg);
+
+ /* Dispatch on message type. */
+ switch (msgtype)
+ {
+ case 'E':
+ case 'N':
+ {
+ ErrorData edata;
+ ErrorContextCallback context;
+
+ /* Parse ErrorResponse or NoticeResponse. */
+ pq_parse_errornotice(&msg, &edata);
+
+ /*
+ * Limit the maximum error level to ERROR. We don't want
+ * a FATAL inside the background worker to kill the user
+ * session.
+ */
+ if (edata.elevel > ERROR)
+ edata.elevel = ERROR;
+
+ /*
+ * Rethrow the error with an appropriate context method.
+ */
+ context.callback = pg_background_error_callback;
+ context.arg = (void *) &pid;
+ context.previous = error_context_stack;
+ error_context_stack = &context;
+ ThrowErrorData(&edata);
+ error_context_stack = context.previous;
+
+ break;
+ }
+ case 'A':
+ {
+ /* Propagate NotifyResponse. */
+ pq_putmessage(msg.data[0], &msg.data[1], nbytes - 1);
+ break;
+ }
+ case 'T':
+ {
+ int16 natts = pq_getmsgint(&msg, 2);
+ int16 i;
+
+ if (state->has_row_description)
+ elog(ERROR, "multiple RowDescription messages");
+ state->has_row_description = true;
+ if (natts != tupdesc->natts)
+ ereport(ERROR,
+ (errcode(ERRCODE_DATATYPE_MISMATCH),
+ errmsg("remote query result rowtype does not match "
+ "the specified FROM clause rowtype")));
+
+ for (i = 0; i < natts; ++i)
+ {
+ Oid type_id;
+
+ (void) pq_getmsgstring(&msg); /* name */
+ (void) pq_getmsgint(&msg, 4); /* table OID */
+ (void) pq_getmsgint(&msg, 2); /* table attnum */
+ type_id = pq_getmsgint(&msg, 4); /* type OID */
+ (void) pq_getmsgint(&msg, 2); /* type length */
+ (void) pq_getmsgint(&msg, 4); /* typmod */
+ (void) pq_getmsgint(&msg, 2); /* format code */
+
+ if (type_id != tupdesc->attrs[i]->atttypid)
+ ereport(ERROR,
+ (errcode(ERRCODE_DATATYPE_MISMATCH),
+ errmsg("remote query result rowtype does not match "
+ "the specified FROM clause rowtype")));
+ }
+
+ pq_getmsgend(&msg);
+
+ break;
+ }
+ case 'D':
+ {
+ /* Handle DataRow message. */
+ HeapTuple result;
+
+ result = form_result_tuple(state, tupdesc, &msg);
+ SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(result));
+ }
+ case 'C':
+ {
+ /* Handle CommandComplete message. */
+ MemoryContext oldcontext;
+ const char *tag = pq_getmsgstring(&msg);
+
+ oldcontext =
+ MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+ state->command_tags = lappend(state->command_tags,
+ pstrdup(tag));
+ MemoryContextSwitchTo(oldcontext);
+ break;
+ }
+ case 'G':
+ case 'H':
+ case 'W':
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("COPY protocol not allowed in pg_background")));
+ }
+
+ case 'Z':
+ {
+ /* Handle ReadyForQuery message. */
+ state->complete = true;
+ break;
+ }
+ default:
+ elog(WARNING, "unknown message type: %c (%zu bytes)",
+ msg.data[0], nbytes);
+ break;
+ }
+ }
+
+ /* Check whether the connection was broken prematurely. */
+ if (!state->complete)
+ ereport(ERROR,
+ (errcode(ERRCODE_CONNECTION_FAILURE),
+ errmsg("lost connection to worker process with PID %d",
+ pid)));
+
+ /* If no data rows, return the command tags instead. */
+ if (!state->has_row_description)
+ {
+ if (tupdesc->natts != 1 || tupdesc->attrs[0]->atttypid != TEXTOID)
+ ereport(ERROR,
+ (errcode(ERRCODE_DATATYPE_MISMATCH),
+ errmsg("remote query did not return a result set, but "
+ "result rowtype is not a single text column")));
+ if (state->command_tags != NIL)
+ {
+ char *tag = linitial(state->command_tags);
+ Datum value;
+ bool isnull;
+ HeapTuple result;
+
+ state->command_tags = list_delete_first(state->command_tags);
+ value = PointerGetDatum(cstring_to_text(tag));
+ isnull = false;
+ result = heap_form_tuple(tupdesc, &value, &isnull);
+ SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(result));
+ }
+ }
+
+ /* We're done! */
+ dsm_detach(state->info->seg);
+ SRF_RETURN_DONE(funcctx);
+}
+
+/*
+ * Parse a DataRow message and form a result tuple.
+ */
+static HeapTuple
+form_result_tuple(pg_background_result_state *state, TupleDesc tupdesc,
+ StringInfo msg)
+{
+ /* Handle DataRow message. */
+ int16 natts = pq_getmsgint(msg, 2);
+ int16 i;
+ Datum *values = NULL;
+ bool *isnull = NULL;
+ StringInfoData buf;
+
+ if (!state->has_row_description)
+ elog(ERROR, "DataRow not preceded by RowDescription");
+ if (natts != tupdesc->natts)
+ elog(ERROR, "malformed DataRow");
+ if (natts > 0)
+ {
+ values = palloc(natts * sizeof(Datum));
+ isnull = palloc(natts * sizeof(bool));
+ }
+ initStringInfo(&buf);
+
+ for (i = 0; i < natts; ++i)
+ {
+ int32 bytes = pq_getmsgint(msg, 4);
+
+ if (bytes < 0)
+ {
+ values[i] = ReceiveFunctionCall(&state->receive_functions[i],
+ NULL,
+ state->typioparams[i],
+ tupdesc->attrs[i]->atttypmod);
+ isnull[i] = true;
+ }
+ else
+ {
+ resetStringInfo(&buf);
+ appendBinaryStringInfo(&buf, pq_getmsgbytes(msg, bytes), bytes);
+ values[i] = ReceiveFunctionCall(&state->receive_functions[i],
+ &buf,
+ state->typioparams[i],
+ tupdesc->attrs[i]->atttypmod);
+ isnull[i] = false;
+ }
+ }
+
+ pq_getmsgend(msg);
+
+ return heap_form_tuple(tupdesc, values, isnull);
+}
+
+/*
+ * Detach from the dynamic shared memory segment used for communication with
+ * a background worker. This prevents the worker from stalling waiting for
+ * us to read its results.
+ */
+Datum
+pg_background_detach(PG_FUNCTION_ARGS)
+{
+ int32 pid = PG_GETARG_INT32(0);
+ pg_background_worker_info *info;
+
+ info = find_worker_info(pid);
+ if (info == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_UNDEFINED_OBJECT),
+ errmsg("PID %d is not attached to this session", pid)));
+ check_rights(info);
+ dsm_detach(info->seg);
+
+ PG_RETURN_VOID();
+}
+
+/*
+ * When the dynamic shared memory segment associated with a worker is
+ * cleaned up, we need to clean up our associated private data structures.
+ */
+static void
+cleanup_worker_info(dsm_segment *seg, Datum pid_datum)
+{
+ pid_t pid = DatumGetInt32(pid_datum);
+ bool found;
+ pg_background_worker_info *info;
+
+ /* Find any worker info entry for this PID. If none, we're done. */
+ if ((info = find_worker_info(pid)) == NULL)
+ return;
+
+ /* Free memory used by the BackgroundWorkerHandle. */
+ if (info->handle != NULL)
+ {
+ pfree(info->handle);
+ info->handle = NULL;
+ }
+
+ /* Remove the hashtable entry. */
+ hash_search(worker_hash, (void *) &pid, HASH_REMOVE, &found);
+ if (!found)
+ elog(ERROR, "pg_background worker_hash table corrupted");
+}
+
+/*
+ * Find the background worker information for the worker with a given PID.
+ */
+static pg_background_worker_info *
+find_worker_info(pid_t pid)
+{
+ pg_background_worker_info *info = NULL;
+
+ if (worker_hash != NULL)
+ info = hash_search(worker_hash, (void *) &pid, HASH_FIND, NULL);
+
+ return info;
+}
+
+/*
+ * Check whether the current user has rights to manipulate the background
+ * worker with the given PID.
+ */
+static void
+check_rights(pg_background_worker_info *info)
+{
+ Oid current_user_id;
+ int sec_context;
+
+ GetUserIdAndSecContext(¤t_user_id, &sec_context);
+ if (!has_privs_of_role(current_user_id, info->current_user_id))
+ ereport(ERROR,
+ (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+ errmsg("permission denied for background worker with PID \"%d\"",
+ info->pid)));
+}
+
+/*
+ * Save worker information for future IPC.
+ */
+static void
+save_worker_info(pid_t pid, dsm_segment *seg, BackgroundWorkerHandle *handle,
+ shm_mq_handle *responseq)
+{
+ pg_background_worker_info *info;
+ Oid current_user_id;
+ int sec_context;
+
+ /* If the hash table hasn't been set up yet, do that now. */
+ if (worker_hash == NULL)
+ {
+ HASHCTL ctl;
+
+ ctl.keysize = sizeof(pid_t);
+ ctl.entrysize = sizeof(pg_background_worker_info);
+ worker_hash = hash_create("pg_background worker_hash", 8, &ctl,
+ HASH_ELEM);
+ }
+
+ /* Get current authentication information. */
+ GetUserIdAndSecContext(¤t_user_id, &sec_context);
+
+ /*
+ * In the unlikely event that there's an older worker with this PID,
+ * just detach it - unless it has a different user ID than the
+ * currently-active one, in which case someone might be trying to pull
+ * a fast one. Let's kill the backend to make sure we don't break
+ * anyone's expectations.
+ */
+ if ((info = find_worker_info(pid)) != NULL)
+ {
+ if (current_user_id != info->current_user_id)
+ ereport(FATAL,
+ (errcode(ERRCODE_DUPLICATE_OBJECT),
+ errmsg("background worker with PID \"%d\" already exists",
+ pid)));
+ dsm_detach(info->seg);
+ }
+
+ /* When the DSM is unmapped, clean everything up. */
+ on_dsm_detach(seg, cleanup_worker_info, Int32GetDatum(pid));
+
+ /* Create a new entry for this worker. */
+ info = hash_search(worker_hash, (void *) &pid, HASH_ENTER, NULL);
+ info->seg = seg;
+ info->handle = handle;
+ info->responseq = responseq;
+ info->consumed = false;
+}
+
+/*
+ * Indicate that an error came from a particular background worker.
+ */
+static void
+pg_background_error_callback(void *arg)
+{
+ pid_t pid = * (pid_t *) arg;
+
+ errcontext("background worker, pid %d", pid);
+}
+
+/*
+ * Background worker entrypoint.
+ */
+void
+pg_background_worker_main(Datum main_arg)
+{
+ dsm_segment *seg;
+ shm_toc *toc;
+ pg_background_fixed_data *fdata;
+ char *sql;
+ char *gucstate;
+ shm_mq *mq;
+ shm_mq_handle *responseq;
+
+ /* Establish signal handlers. */
+ pqsignal(SIGTERM, handle_sigterm);
+ BackgroundWorkerUnblockSignals();
+
+ /* Set up a memory context and resource owner. */
+ Assert(CurrentResourceOwner == NULL);
+ CurrentResourceOwner = ResourceOwnerCreate(NULL, "pg_background");
+ CurrentMemoryContext = AllocSetContextCreate(TopMemoryContext,
+ "pg_background session",
+ ALLOCSET_DEFAULT_MINSIZE,
+ ALLOCSET_DEFAULT_INITSIZE,
+ ALLOCSET_DEFAULT_MAXSIZE);
+
+ /* Connect to the dynamic shared memory segment. */
+ seg = dsm_attach(DatumGetInt32(main_arg));
+ if (seg == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("unable to map dynamic shared memory segment")));
+ toc = shm_toc_attach(PG_BACKGROUND_MAGIC, dsm_segment_address(seg));
+ if (toc == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("bad magic number in dynamic shared memory segment")));
+
+ /* Find data structures in dynamic shared memory. */
+ fdata = shm_toc_lookup(toc, PG_BACKGROUND_KEY_FIXED_DATA);
+ sql = shm_toc_lookup(toc, PG_BACKGROUND_KEY_SQL);
+ gucstate = shm_toc_lookup(toc, PG_BACKGROUND_KEY_GUC);
+ mq = shm_toc_lookup(toc, PG_BACKGROUND_KEY_QUEUE);
+ shm_mq_set_sender(mq, MyProc);
+ responseq = shm_mq_attach(mq, seg, NULL);
+
+ /* Redirect protocol messages to responseq. */
+ pq_redirect_to_shm_mq(mq, responseq);
+
+ /*
+ * Initialize our user and database ID based on the strings version of
+ * the data, and then go back and check that we actually got the database
+ * and user ID that we intended to get. We do this because it's not
+ * impossible for the process that started us to die before we get here,
+ * and the user or database could be renamed in the meantime. We don't
+ * want to latch on the wrong object by accident. There should probably
+ * be a variant of BackgroundWorkerInitializeConnection that accepts OIDs
+ * rather than strings.
+ */
+ BackgroundWorkerInitializeConnection(NameStr(fdata->database),
+ NameStr(fdata->authenticated_user));
+ if (fdata->database_id != MyDatabaseId ||
+ fdata->authenticated_user_id != GetAuthenticatedUserId())
+ ereport(ERROR,
+ (errmsg("user or database renamed during pg_background startup")));
+
+ /* Restore GUC values from launching backend. */
+ StartTransactionCommand();
+ RestoreGUCState(gucstate);
+ CommitTransactionCommand();
+
+ /* Handle local_preload_libraries and session_preload_libraries. */
+ process_session_preload_libraries();
+
+ /* Restore user ID and security context. */
+ SetUserIdAndSecContext(fdata->current_user_id, fdata->sec_context);
+
+ /* Prepare to execute the query. */
+ SetCurrentStatementStartTimestamp();
+ debug_query_string = sql;
+ pgstat_report_activity(STATE_RUNNING, sql);
+
+ /* Execute the query. */
+ execute_sql_string(sql);
+
+ /* Post-execution cleanup. */
+ ProcessCompletedNotifies();
+ pgstat_report_activity(STATE_IDLE, sql);
+ pgstat_report_stat(true);
+
+ /* Signal that we are done. */
+ ReadyForQuery(DestRemote);
+}
+
+/*
+ * Execute given SQL string.
+ *
+ * Using SPI here would preclude backgrounding commands like VACUUM which one
+ * might very well wish to launch in the background. So we do this instead.
+ */
+static void
+execute_sql_string(const char *sql)
+{
+ List *raw_parsetree_list;
+ ListCell *lc1;
+ bool isTopLevel;
+ int commands_remaining;
+ MemoryContext parsecontext;
+ MemoryContext oldcontext;
+
+ /* Start up a transaction command. */
+ start_xact_command();
+
+ /*
+ * Parse the SQL string into a list of raw parse trees.
+ *
+ * Because we allow statements that perform internal transaction control,
+ * we can't do this in TopTransactionContext; the parse trees might get
+ * blown away before we're done executing them.
+ */
+ parsecontext = AllocSetContextCreate(TopMemoryContext,
+ "pg_background parse/plan",
+ ALLOCSET_DEFAULT_MINSIZE,
+ ALLOCSET_DEFAULT_INITSIZE,
+ ALLOCSET_DEFAULT_MAXSIZE);
+ oldcontext = MemoryContextSwitchTo(parsecontext);
+ raw_parsetree_list = pg_parse_query(sql);
+ commands_remaining = list_length(raw_parsetree_list);
+ isTopLevel = commands_remaining == 1;
+ MemoryContextSwitchTo(oldcontext);
+
+ /*
+ * Do parse analysis, rule rewrite, planning, and execution for each raw
+ * parsetree. We must fully execute each query before beginning parse
+ * analysis on the next one, since there may be interdependencies.
+ */
+ foreach(lc1, raw_parsetree_list)
+ {
+ Node *parsetree = (Node *) lfirst(lc1);
+ const char *commandTag;
+ char completionTag[COMPLETION_TAG_BUFSIZE];
+ List *querytree_list,
+ *plantree_list;
+ bool snapshot_set = false;
+ Portal portal;
+ DestReceiver *receiver;
+ int16 format = 1;
+
+ /*
+ * We don't allow transaction-control commands like COMMIT and ABORT
+ * here. The entire SQL statement is executed as a single transaction
+ * which commits if no errors are encountered.
+ */
+ if (IsA(parsetree, TransactionStmt))
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("transaction control statements are not allowed in pg_background")));
+
+ /*
+ * Get the command name for use in status display (it also becomes the
+ * default completion tag, down inside PortalRun). Set ps_status and
+ * do any special start-of-SQL-command processing needed by the
+ * destination.
+ */
+ commandTag = CreateCommandTag(parsetree);
+ set_ps_display(commandTag, false);
+ BeginCommand(commandTag, DestNone);
+
+ /* Set up a snapshot if parse analysis/planning will need one. */
+ if (analyze_requires_snapshot(parsetree))
+ {
+ PushActiveSnapshot(GetTransactionSnapshot());
+ snapshot_set = true;
+ }
+
+ /*
+ * OK to analyze, rewrite, and plan this query.
+ *
+ * As with parsing, we need to make sure this data outlives the
+ * transaction, because of the possibility that the statement might
+ * perform internal transaction control.
+ */
+ oldcontext = MemoryContextSwitchTo(parsecontext);
+ querytree_list = pg_analyze_and_rewrite(parsetree, sql, NULL, 0);
+ plantree_list = pg_plan_queries(querytree_list, 0, NULL);
+
+ /* Done with the snapshot used for parsing/planning */
+ if (snapshot_set)
+ PopActiveSnapshot();
+
+ /* If we got a cancel signal in analysis or planning, quit */
+ CHECK_FOR_INTERRUPTS();
+
+ /*
+ * Execute the query using the unnamed portal.
+ */
+ portal = CreatePortal("", true, true);
+ /* Don't display the portal in pg_cursors */
+ portal->visible = false;
+ PortalDefineQuery(portal, NULL, sql, commandTag, plantree_list, NULL);
+ PortalStart(portal, NULL, 0, InvalidSnapshot);
+
+ /* We always use binary format, for efficiency. */
+ PortalSetResultFormat(portal, 1, &format);
+
+ /*
+ * Tuples returned by any command other than the last are simply
+ * discarded; but those returned by the last (or only) command are
+ * redirected to the shared memory queue we're using for communication
+ * with the launching backend. If the launching backend is gone or has
+ * detached us, these messages will just get dropped on the floor.
+ */
+ --commands_remaining;
+ if (commands_remaining > 0)
+ receiver = CreateDestReceiver(DestNone);
+ else
+ {
+ receiver = CreateDestReceiver(DestRemote);
+ SetRemoteDestReceiverParams(receiver, portal);
+ }
+
+ /*
+ * Only once the portal and destreceiver have been established can
+ * we return to the transaction context. All that stuff needs to
+ * survive an internal commit inside PortalRun!
+ */
+ MemoryContextSwitchTo(oldcontext);
+
+ /* Here's where we actually execute the command. */
+ (void) PortalRun(portal, FETCH_ALL, isTopLevel, receiver, receiver,
+ completionTag);
+
+ /* Clean up the receiver. */
+ (*receiver->rDestroy) (receiver);
+
+ /* Clean up the portal. */
+ PortalDrop(portal, false);
+
+ /*
+ * If this is the last parsetree, close down transaction statement
+ * before reporting CommandComplete. Otherwise, we need a
+ * CommandCounterIncrement.
+ */
+ if (lnext(lc1) == NULL)
+ finish_xact_command();
+ else
+ CommandCounterIncrement();
+
+ /*
+ * Send a CommandComplete message even if we suppressed the query
+ * results. The user backend will report the command tags in the
+ * absence of any true query results.
+ */
+ EndCommand(completionTag, DestRemote);
+ }
+
+ /* Make sure there's not still a transaction open. */
+ finish_xact_command();
+}
+
+/*
+ * When we receive a SIGTERM, we set InterruptPending and ProcDiePending just
+ * like a normal backend. The next CHECK_FOR_INTERRUPTS() will do the right
+ * thing.
+ */
+static void
+handle_sigterm(SIGNAL_ARGS)
+{
+ int save_errno = errno;
+
+ if (MyProc)
+ SetLatch(&MyProc->procLatch);
+
+ if (!proc_exit_inprogress)
+ {
+ InterruptPending = true;
+ ProcDiePending = true;
+ }
+
+ errno = save_errno;
+}
diff --git a/contrib/pg_background/pg_background.control b/contrib/pg_background/pg_background.control
new file mode 100644
index 0000000..733d0e1
--- /dev/null
+++ b/contrib/pg_background/pg_background.control
@@ -0,0 +1,4 @@
+comment = 'Run SQL queries in the background'
+default_version = '1.0'
+module_pathname = '$libdir/pg_background'
+relocatable = true
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index 61f17bf..e44f5fa 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -190,8 +190,6 @@ static int errdetail_execute(List *raw_parsetree_list);
static int errdetail_params(ParamListInfo params);
static int errdetail_abort(void);
static int errdetail_recovery_conflict(void);
-static void start_xact_command(void);
-static void finish_xact_command(void);
static bool IsTransactionExitStmt(Node *parsetree);
static bool IsTransactionExitStmtList(List *parseTrees);
static bool IsTransactionStmtList(List *parseTrees);
@@ -2373,7 +2371,7 @@ exec_describe_portal_message(const char *portal_name)
/*
* Convenience routines for starting/committing a single command.
*/
-static void
+void
start_xact_command(void)
{
if (!xact_started)
@@ -2393,7 +2391,7 @@ start_xact_command(void)
}
}
-static void
+void
finish_xact_command(void)
{
if (xact_started)
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index c23f4da..8e84ee7 100644
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -224,7 +224,7 @@ extern PGPROC *PreparedXactProcs;
/* configurable options */
extern int DeadlockTimeout;
-extern int StatementTimeout;
+extern PGDLLIMPORT int StatementTimeout;
extern int LockTimeout;
extern bool log_lock_waits;
diff --git a/src/include/tcop/tcopprot.h b/src/include/tcop/tcopprot.h
index 60f7532..fd3df58 100644
--- a/src/include/tcop/tcopprot.h
+++ b/src/include/tcop/tcopprot.h
@@ -69,6 +69,8 @@ extern void RecoveryConflictInterrupt(ProcSignalReason reason); /* called from S
* handler */
extern void prepare_for_client_read(void);
extern void client_read_ended(void);
+extern void start_xact_command(void);
+extern void finish_xact_command(void);
extern void process_postgres_switches(int argc, char *argv[],
GucContext ctx, const char **dbname);
extern void PostgresMain(int argc, char *argv[],
--
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
