Rationalize parallel dump/restore's handling of worker cmd/status messages.

The existing APIs for creating and parsing command and status messages are
rather messy; for example, archive-format modules have to provide code
for constructing command messages, which is entirely pointless since
the code to read them is hard-wired in WaitForCommands() and hence
no format-specific variation is actually possible.  But there's little
foreseeable reason to need format-specific variation anyway.

The situation for status messages is no better; at least those are both
constructed and parsed by format-specific code, but said code is quite
redundant since there's no actual need for format-specific variation.

To add insult to injury, the first API involves returning pointers to
static buffers, which is bad, while the second involves returning pointers
to malloc'd strings, which is safer but randomly inconsistent.

Hence, get rid of the MasterStartParallelItem and MasterEndParallelItem
APIs, and instead write centralized functions that construct and parse
command and status messages.  If we ever do need more flexibility, these
functions can be the standard implementations of format-specific
callback methods, but that's a long way off if it ever happens.

Tom Lane, reviewed by Kevin Grittner

Discussion: <17340.1464465717@sss.pgh.pa.us>

1 files changed, 145 insertions, 87 deletions

diff --git a/src/bin/pg_dump/parallel.c b/src/bin/pg_dump/parallel.c
index 634b4444f9f..0e2bfa106a7 100644
--- a/src/bin/pg_dump/parallel.c
+++ b/src/bin/pg_dump/parallel.c
@@ -20,32 +20,25 @@
  * the desired number of worker processes, which each enter WaitForCommands().
  *
  * The master process dispatches an individual work item to one of the worker
- * processes in DispatchJobForTocEntry().  That calls
- * AH->MasterStartParallelItemPtr, a routine of the output format.  This
- * function's arguments are the parents archive handle AH (containing the full
- * catalog information), the TocEntry that the worker should work on and a
- * T_Action value indicating whether this is a backup or a restore task.  The
- * function simply converts the TocEntry assignment into a command string that
- * is then sent over to the worker process. In the simplest case that would be
- * something like "DUMP 1234", with 1234 being the TocEntry id.
- *
+ * processes in DispatchJobForTocEntry().  We send a command string such as
+ * "DUMP 1234" or "RESTORE 1234", where 1234 is the TocEntry ID.
  * The worker process receives and decodes the command and passes it to the
  * routine pointed to by AH->WorkerJobDumpPtr or AH->WorkerJobRestorePtr,
  * which are routines of the current archive format.  That routine performs
- * the required action (dump or restore) and returns a malloc'd status string.
- * The status string is passed back to the master where it is interpreted by
- * AH->MasterEndParallelItemPtr, another format-specific routine.  That
- * function can update format-specific information on the master's side,
- * depending on the reply from the worker process.  In the end it returns a
- * status code, which we pass to the ParallelCompletionPtr callback function
- * that was passed to DispatchJobForTocEntry().  The callback function does
- * state updating for the master control logic in pg_backup_archiver.c.
+ * the required action (dump or restore) and returns an integer status code.
+ * This is passed back to the master where we pass it to the
+ * ParallelCompletionPtr callback function that was passed to
+ * DispatchJobForTocEntry().  The callback function does state updating
+ * for the master control logic in pg_backup_archiver.c.
  *
- * Remember that we have forked off the workers only after we have read in
- * the catalog.  That's why our worker processes can also access the catalog
- * information.  (In the Windows case, the workers are threads in the same
- * process.  To avoid problems, they work with cloned copies of the Archive
- * data structure; see RunWorker().)
+ * In principle additional archive-format-specific information might be needed
+ * in commands or worker status responses, but so far that hasn't proved
+ * necessary, since workers have full copies of the ArchiveHandle/TocEntry
+ * data structures.  Remember that we have forked off the workers only after
+ * we have read in the catalog.  That's why our worker processes can also
+ * access the catalog information.  (In the Windows case, the workers are
+ * threads in the same process.  To avoid problems, they work with cloned
+ * copies of the Archive data structure; see RunWorker().)
  *
  * In the master process, the workerStatus field for each worker has one of
  * the following values:
@@ -1074,6 +1067,110 @@ ParallelBackupEnd(ArchiveHandle *AH, ParallelState *pstate)
 }
 
 /*
+ * These next four functions handle construction and parsing of the command
+ * strings and response strings for parallel workers.
+ *
+ * Currently, these can be the same regardless of which archive format we are
+ * processing.  In future, we might want to let format modules override these
+ * functions to add format-specific data to a command or response.
+ */
+
+/*
+ * buildWorkerCommand: format a command string to send to a worker.
+ *
+ * The string is built in the caller-supplied buffer of size buflen.
+ */
+static void
+buildWorkerCommand(ArchiveHandle *AH, TocEntry *te, T_Action act,
+				   char *buf, int buflen)
+{
+	if (act == ACT_DUMP)
+		snprintf(buf, buflen, "DUMP %d", te->dumpId);
+	else if (act == ACT_RESTORE)
+		snprintf(buf, buflen, "RESTORE %d", te->dumpId);
+	else
+		Assert(false);
+}
+
+/*
+ * parseWorkerCommand: interpret a command string in a worker.
+ */
+static void
+parseWorkerCommand(ArchiveHandle *AH, TocEntry **te, T_Action *act,
+				   const char *msg)
+{
+	DumpId		dumpId;
+	int			nBytes;
+
+	if (messageStartsWith(msg, "DUMP "))
+	{
+		*act = ACT_DUMP;
+		sscanf(msg, "DUMP %d%n", &dumpId, &nBytes);
+		Assert(nBytes == strlen(msg));
+		*te = getTocEntryByDumpId(AH, dumpId);
+		Assert(*te != NULL);
+	}
+	else if (messageStartsWith(msg, "RESTORE "))
+	{
+		*act = ACT_RESTORE;
+		sscanf(msg, "RESTORE %d%n", &dumpId, &nBytes);
+		Assert(nBytes == strlen(msg));
+		*te = getTocEntryByDumpId(AH, dumpId);
+		Assert(*te != NULL);
+	}
+	else
+		exit_horribly(modulename,
+					  "unrecognized command received from master: \"%s\"\n",
+					  msg);
+}
+
+/*
+ * buildWorkerResponse: format a response string to send to the master.
+ *
+ * The string is built in the caller-supplied buffer of size buflen.
+ */
+static void
+buildWorkerResponse(ArchiveHandle *AH, TocEntry *te, T_Action act, int status,
+					char *buf, int buflen)
+{
+	snprintf(buf, buflen, "OK %d %d %d",
+			 te->dumpId,
+			 status,
+			 status == WORKER_IGNORED_ERRORS ? AH->public.n_errors : 0);
+}
+
+/*
+ * parseWorkerResponse: parse the status message returned by a worker.
+ *
+ * Returns the integer status code, and may update fields of AH and/or te.
+ */
+static int
+parseWorkerResponse(ArchiveHandle *AH, TocEntry *te,
+					const char *msg)
+{
+	DumpId		dumpId;
+	int			nBytes,
+				n_errors;
+	int			status = 0;
+
+	if (messageStartsWith(msg, "OK "))
+	{
+		sscanf(msg, "OK %d %d %d%n", &dumpId, &status, &n_errors, &nBytes);
+
+		Assert(dumpId == te->dumpId);
+		Assert(nBytes == strlen(msg));
+
+		AH->public.n_errors += n_errors;
+	}
+	else
+		exit_horribly(modulename,
+					  "invalid message received from worker: \"%s\"\n",
+					  msg);
+
+	return status;
+}
+
+/*
  * Dispatch a job to some free worker.
  *
  * te is the TocEntry to be processed, act is the action to be taken on it.
@@ -1091,18 +1188,16 @@ DispatchJobForTocEntry(ArchiveHandle *AH,
 					   void *callback_data)
 {
 	int			worker;
-	char	   *arg;
+	char		buf[256];
 
 	/* Get a worker, waiting if none are idle */
 	while ((worker = GetIdleWorker(pstate)) == NO_SLOT)
 		WaitForWorkers(AH, pstate, WFW_ONE_IDLE);
 
 	/* Construct and send command string */
-	arg = (AH->MasterStartParallelItemPtr) (AH, te, act);
-
-	sendMessageToWorker(pstate, worker, arg);
+	buildWorkerCommand(AH, te, act, buf, sizeof(buf));
 
-	/* XXX aren't we leaking string here? (no, because it's static. Ick.) */
+	sendMessageToWorker(pstate, worker, buf);
 
 	/* Remember worker is busy, and which TocEntry it's working on */
 	pstate->parallelSlot[worker].workerStatus = WRKR_WORKING;
@@ -1220,10 +1315,10 @@ static void
 WaitForCommands(ArchiveHandle *AH, int pipefd[2])
 {
 	char	   *command;
-	DumpId		dumpId;
-	int			nBytes;
-	char	   *str;
 	TocEntry   *te;
+	T_Action	act;
+	int			status = 0;
+	char		buf[256];
 
 	for (;;)
 	{
@@ -1233,47 +1328,29 @@ WaitForCommands(ArchiveHandle *AH, int pipefd[2])
 			return;
 		}
 
-		if (messageStartsWith(command, "DUMP "))
-		{
-			/* Decode the command */
-			sscanf(command + strlen("DUMP "), "%d%n", &dumpId, &nBytes);
-			Assert(nBytes == strlen(command) - strlen("DUMP "));
-			te = getTocEntryByDumpId(AH, dumpId);
-			Assert(te != NULL);
+		/* Decode the command */
+		parseWorkerCommand(AH, &te, &act, command);
 
+		if (act == ACT_DUMP)
+		{
 			/* Acquire lock on this table within the worker's session */
 			lockTableForWorker(AH, te);
 
 			/* Perform the dump command */
-			str = (AH->WorkerJobDumpPtr) (AH, te);
-
-			/* Return status to master */
-			sendMessageToMaster(pipefd, str);
-
-			/* we are responsible for freeing the status string */
-			free(str);
+			status = (AH->WorkerJobDumpPtr) (AH, te);
 		}
-		else if (messageStartsWith(command, "RESTORE "))
+		else if (act == ACT_RESTORE)
 		{
-			/* Decode the command */
-			sscanf(command + strlen("RESTORE "), "%d%n", &dumpId, &nBytes);
-			Assert(nBytes == strlen(command) - strlen("RESTORE "));
-			te = getTocEntryByDumpId(AH, dumpId);
-			Assert(te != NULL);
-
 			/* Perform the restore command */
-			str = (AH->WorkerJobRestorePtr) (AH, te);
-
-			/* Return status to master */
-			sendMessageToMaster(pipefd, str);
-
-			/* we are responsible for freeing the status string */
-			free(str);
+			status = (AH->WorkerJobRestorePtr) (AH, te);
 		}
 		else
-			exit_horribly(modulename,
-					   "unrecognized command received from master: \"%s\"\n",
-						  command);
+			Assert(false);
+
+		/* Return status to master */
+		buildWorkerResponse(AH, te, act, status, buf, sizeof(buf));
+
+		sendMessageToMaster(pipefd, buf);
 
 		/* command was pg_malloc'd and we are responsible for free()ing it. */
 		free(command);
@@ -1286,9 +1363,9 @@ WaitForCommands(ArchiveHandle *AH, int pipefd[2])
  * If do_wait is true, wait to get a status message; otherwise, just return
  * immediately if there is none available.
  *
- * When we get a status message, we let MasterEndParallelItemPtr process it,
- * then pass the resulting status code to the callback function that was
- * specified to DispatchJobForTocEntry, then reset the worker status to IDLE.
+ * When we get a status message, we pass the status code to the callback
+ * function that was specified to DispatchJobForTocEntry, then reset the
+ * worker status to IDLE.
  *
  * Returns true if we collected a status message, else false.
  *
@@ -1318,29 +1395,10 @@ ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait)
 	{
 		ParallelSlot *slot = &pstate->parallelSlot[worker];
 		TocEntry   *te = slot->te;
-		char	   *statusString;
 		int			status;
 
-		if (messageStartsWith(msg, "OK RESTORE "))
-		{
-			statusString = msg + strlen("OK RESTORE ");
-			status =
-				(AH->MasterEndParallelItemPtr)
-				(AH, te, statusString, ACT_RESTORE);
-			slot->callback(AH, te, status, slot->callback_data);
-		}
-		else if (messageStartsWith(msg, "OK DUMP "))
-		{
-			statusString = msg + strlen("OK DUMP ");
-			status =
-				(AH->MasterEndParallelItemPtr)
-				(AH, te, statusString, ACT_DUMP);
-			slot->callback(AH, te, status, slot->callback_data);
-		}
-		else
-			exit_horribly(modulename,
-						  "invalid message received from worker: \"%s\"\n",
-						  msg);
+		status = parseWorkerResponse(AH, te, msg);
+		slot->callback(AH, te, status, slot->callback_data);
 		slot->workerStatus = WRKR_IDLE;
 		slot->te = NULL;
 	}
@@ -1364,8 +1422,8 @@ ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait)
  * WFW_ONE_IDLE: wait for at least one worker to be idle
  * WFW_ALL_IDLE: wait for all workers to be idle
  *
- * Any received results are passed to MasterEndParallelItemPtr and then
- * to the callback specified to DispatchJobForTocEntry.
+ * Any received results are passed to the callback specified to
+ * DispatchJobForTocEntry.
  *
  * This function is executed in the master process.
  */