[CF 4291] v8 - Logging parallel worker draught

This branch was automatically generated by a robot using patches from an
email thread registered at:

https://commitfest.postgresql.org/patch/4291

The branch will be overwritten each time a new patch version is posted to
the thread, and also periodically to check for bitrot caused by changes
on the master branch.

Patch(es): https://www.postgresql.org/message-id/CAA5RZ0tHb_hpq472oMj6S8F5kgXiOE-mV9QYSJOjn4g+CvZXmA@mail.gmail.com
Author(s): Benoit Lobréau

Author: Commitfest Bot

doc/src/sgml/config.sgml

@@ -7912,6 +7912,24 @@ log_line_prefix = '%m [%p] %q%u@%d/%a '
       </listitem>
      </varlistentry>
 
+    <varlistentry id="guc-log-parallel-workers" xreflabel="log_parallel_workers">
+      <term><varname>log_parallel_workers</varname> (<type>enum</type>)
+      <indexterm>
+       <primary><varname>log_parallel_workers</varname> configuration parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Controls whether a log message about the number of workers is emitted during the
+        execution of a parallel query or utility statement. The default value is
+        <literal>none</literal> which disables logging. <literal>all</literal> emits
+        information for all parallel queries or utilities, whereas <literal>shortage</literal>
+        emits information only when the number of workers launched is lower than the number
+        of planned workers.
+       </para>
+      </listitem>
+     </varlistentry>
+
      <varlistentry id="guc-log-parameter-max-length" xreflabel="log_parameter_max_length">
       <term><varname>log_parameter_max_length</varname> (<type>integer</type>)
       <indexterm>

src/backend/access/brin/brin.c

@@ -2552,6 +2552,10 @@ _brin_end_parallel(BrinLeader *brinleader, BrinBuildState *state)
 	/* Shutdown worker processes */
 	WaitForParallelWorkersToFinish(brinleader->pcxt);
 
+	LogParallelWorkersIfNeeded(log_parallel_workers,
+							   brinleader->pcxt->nworkers_to_launch,
+							   brinleader->pcxt->nworkers_launched);
+
 	/*
 	 * Next, accumulate WAL usage.  (This must wait for the workers to finish,
 	 * or we might get incomplete data.)

src/backend/access/nbtree/nbtsort.c

@@ -1613,6 +1613,10 @@ _bt_end_parallel(BTLeader *btleader)
 	/* Shutdown worker processes */
 	WaitForParallelWorkersToFinish(btleader->pcxt);
 
+	LogParallelWorkersIfNeeded(log_parallel_workers,
+							   btleader->pcxt->nworkers_to_launch,
+							   btleader->pcxt->nworkers_launched);
+
 	/*
 	 * Next, accumulate WAL usage.  (This must wait for the workers to finish,
 	 * or we might get incomplete data.)

src/backend/access/transam/parallel.c

@@ -1663,3 +1663,22 @@ LookupParallelWorkerFunction(const char *libraryname, const char *funcname)
 	return (parallel_worker_main_type)
 		load_external_function(libraryname, funcname, true, NULL);
 }
+
+/*
+ * If required, emit information about parallel workers usage in
+ * the logs.
+ */
+void
+LogParallelWorkersIfNeeded(int log_parallel_workers,
+						   int parallel_workers_to_launch,
+						   int parallel_workers_launched)
+{
+	if ((log_parallel_workers == LOG_PARALLEL_WORKERS_ALL &&
+		parallel_workers_to_launch > 0) ||
+		(log_parallel_workers == LOG_PARALLEL_WORKERS_SHORTAGE &&
+		parallel_workers_to_launch != parallel_workers_launched))
+		ereport(LOG,
+				(errmsg("launched %i parallel workers (planned: %i)",
+				 parallel_workers_launched,
+				 parallel_workers_to_launch)));
+}

src/backend/commands/vacuumparallel.c

@@ -208,6 +208,9 @@ struct ParallelVacuumState
 	int			nindexes_parallel_cleanup;
 	int			nindexes_parallel_condcleanup;
 
+	int			nworkers_to_launch;
+	int			nworkers_launched;
+
 	/* Buffer access strategy used by leader process */
 	BufferAccessStrategy bstrategy;
 
@@ -362,6 +365,9 @@ parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes,
 		if ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) != 0)
 			pvs->nindexes_parallel_condcleanup++;
 	}
+	pvs->nworkers_to_launch = 0;
+	pvs->nworkers_launched = 0;
+
 	shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_INDEX_STATS, indstats);
 	pvs->indstats = indstats;
 
@@ -437,6 +443,10 @@ parallel_vacuum_end(ParallelVacuumState *pvs, IndexBulkDeleteResult **istats)
 {
 	Assert(!IsParallelWorker());
 
+	LogParallelWorkersIfNeeded(log_parallel_workers,
+							   pvs->nworkers_to_launch,
+							   pvs->nworkers_launched);
+
 	/* Copy the updated statistics */
 	for (int i = 0; i < pvs->nindexes; i++)
 	{
@@ -738,6 +748,9 @@ parallel_vacuum_process_all_indexes(ParallelVacuumState *pvs, int num_index_scan
 
 		for (int i = 0; i < pvs->pcxt->nworkers_launched; i++)
 			InstrAccumParallelQuery(&pvs->buffer_usage[i], &pvs->wal_usage[i]);
+
+		pvs->nworkers_to_launch += pvs->pcxt->nworkers_to_launch;
+		pvs->nworkers_launched += pvs->pcxt->nworkers_launched;
 	}
 
 	/*

src/backend/executor/execMain.c

@@ -560,6 +560,10 @@ standard_ExecutorEnd(QueryDesc *queryDesc)
 		pgstat_update_parallel_workers_stats((PgStat_Counter) estate->es_parallel_workers_to_launch,
 											 (PgStat_Counter) estate->es_parallel_workers_launched);
 
+	LogParallelWorkersIfNeeded(log_parallel_workers,
+							   estate->es_parallel_workers_to_launch,
+							   estate->es_parallel_workers_launched);
+
 	/*
 	 * Check that ExecutorFinish was called, unless in EXPLAIN-only mode or if
 	 * execution was aborted.

src/backend/utils/misc/guc_tables.c

@@ -28,6 +28,7 @@
 
 #include "access/commit_ts.h"
 #include "access/gin.h"
+#include "access/parallel.h"
 #include "access/slru.h"
 #include "access/toast_compression.h"
 #include "access/twophase.h"
@@ -428,6 +429,13 @@ static const struct config_enum_entry debug_logical_replication_streaming_option
 	{NULL, 0, false}
 };
 
+static const struct config_enum_entry log_parallel_workers_options[] = {
+	{"none", LOG_PARALLEL_WORKERS_NONE, false},
+	{"all", LOG_PARALLEL_WORKERS_ALL, false},
+	{"shortage", LOG_PARALLEL_WORKERS_SHORTAGE, false},
+	{NULL, 0, false}
+};
+
 StaticAssertDecl(lengthof(ssl_protocol_versions_info) == (PG_TLS1_3_VERSION + 2),
 				 "array length mismatch");
 
@@ -531,6 +539,7 @@ int			log_min_duration_statement = -1;
 int			log_parameter_max_length = -1;
 int			log_parameter_max_length_on_error = 0;
 int			log_temp_files = -1;
+int			log_parallel_workers = LOG_PARALLEL_WORKERS_NONE;
 double		log_statement_sample_rate = 1.0;
 double		log_xact_sample_rate = 0;
 char	   *backtrace_functions;
@@ -5340,6 +5349,16 @@ struct config_enum ConfigureNamesEnum[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"log_parallel_workers", PGC_SUSET, LOGGING_WHAT,
+			gettext_noop("Log information about parallel worker usage"),
+			NULL
+		},
+		&log_parallel_workers,
+		LOG_PARALLEL_WORKERS_NONE, log_parallel_workers_options,
+		NULL, NULL, NULL
+	},
+
 	{
 		{"ssl_min_protocol_version", PGC_SIGHUP, CONN_AUTH_SSL,
 			gettext_noop("Sets the minimum SSL/TLS protocol version to use."),

src/backend/utils/misc/postgresql.conf.sample

@@ -638,6 +638,7 @@
 #log_temp_files = -1			# log temporary files equal or larger
 					# than the specified size in kilobytes;
 					# -1 disables, 0 logs all temp files
+#log_parallel_workers = none		# none, all, shortage
 #log_timezone = 'GMT'
 
 # - Process Title -

src/include/access/parallel.h

@@ -53,6 +53,12 @@ typedef struct ParallelWorkerContext
 	shm_toc    *toc;
 } ParallelWorkerContext;
 
+typedef enum {
+	LOG_PARALLEL_WORKERS_NONE=0,
+	LOG_PARALLEL_WORKERS_ALL,
+	LOG_PARALLEL_WORKERS_SHORTAGE,
+}			log_parallel_workers_option_list;
+
 extern PGDLLIMPORT volatile sig_atomic_t ParallelMessagePending;
 extern PGDLLIMPORT int ParallelWorkerNumber;
 extern PGDLLIMPORT bool InitializingParallelWorker;
@@ -78,4 +84,8 @@ extern void ParallelWorkerReportLastRecEnd(XLogRecPtr last_xlog_end);
 
 extern void ParallelWorkerMain(Datum main_arg);
 
+extern void LogParallelWorkersIfNeeded(int log_parallel_workers,
+									   int parallel_workers_to_launch,
+									   int parallel_workers_launched);
+
 #endif							/* PARALLEL_H */

src/include/utils/guc.h

@@ -279,6 +279,7 @@ extern PGDLLIMPORT int log_temp_files;
 extern PGDLLIMPORT double log_statement_sample_rate;
 extern PGDLLIMPORT double log_xact_sample_rate;
 extern PGDLLIMPORT char *backtrace_functions;
+extern PGDLLIMPORT int log_parallel_workers;
 
 extern PGDLLIMPORT int temp_file_limit;