diff --git a/src/bin/pg_dump/parallel.c b/src/bin/pg_dump/parallel.c index e9e8698..509a312 100644 *** a/src/bin/pg_dump/parallel.c --- b/src/bin/pg_dump/parallel.c *************** *** 20,49 **** * 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. ! * * 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 state or catalog information on the master's side, ! * depending on the reply from the worker process. In the end it returns a ! * status code, which is 0 for successful execution. * ! * 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 init_spawned_worker_win32().) * * In the master process, the workerStatus field for each worker has one of * the following values: --- 20,41 ---- * 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(). 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 an integer status code. ! * This is passed back to the master which updates its state accordingly. * ! * 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 init_spawned_worker_win32().) * * In the master process, the workerStatus field for each worker has one of * the following values: *************** ParallelBackupEnd(ArchiveHandle *AH, Par *** 703,708 **** --- 695,804 ---- } /* + * 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 (caller must ensure there is one!) * * te is the TocEntry to be processed, act is the action to be taken on it. *************** DispatchJobForTocEntry(ArchiveHandle *AH *** 712,729 **** T_Action act) { int worker; ! char *arg; /* our caller makes sure that at least one worker is idle */ worker = GetIdleWorker(pstate); Assert(worker != NO_SLOT); /* Construct and send command string */ ! arg = (AH->MasterStartParallelItemPtr) (AH, te, act); ! ! sendMessageToWorker(pstate, worker, arg); ! /* XXX aren't we leaking string here? (no, because it's static. Ick.) */ /* Remember worker is busy, and which TocEntry it's working on */ pstate->parallelSlot[worker].workerStatus = WRKR_WORKING; --- 808,823 ---- T_Action act) { int worker; ! char buf[256]; /* our caller makes sure that at least one worker is idle */ worker = GetIdleWorker(pstate); Assert(worker != NO_SLOT); /* Construct and send command string */ ! buildWorkerCommand(AH, te, act, buf, sizeof(buf)); ! sendMessageToWorker(pstate, worker, buf); /* Remember worker is busy, and which TocEntry it's working on */ pstate->parallelSlot[worker].workerStatus = WRKR_WORKING; *************** static void *** 863,872 **** WaitForCommands(ArchiveHandle *AH, int pipefd[2]) { char *command; - DumpId dumpId; - int nBytes; - char *str; TocEntry *te; for (;;) { --- 957,966 ---- WaitForCommands(ArchiveHandle *AH, int pipefd[2]) { char *command; TocEntry *te; + T_Action act; + int status = 0; + char buf[256]; for (;;) { *************** WaitForCommands(ArchiveHandle *AH, int p *** 878,924 **** 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); /* 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); } ! else if (messageStartsWith(command, "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); } else ! exit_horribly(modulename, ! "unrecognized command received from master: \"%s\"\n", ! command); /* command was pg_malloc'd and we are responsible for free()ing it. */ free(command); --- 972,1000 ---- return; } ! /* 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 */ ! status = (AH->WorkerJobDumpPtr) (AH, te); } ! else if (act == ACT_RESTORE) { /* Perform the restore command */ ! status = (AH->WorkerJobRestorePtr) (AH, te); } else ! 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); *************** WaitForCommands(ArchiveHandle *AH, int p *** 931,940 **** * 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 save the resulting status code and switch the worker's state to ! * WRKR_FINISHED. Later, caller must call ReapWorkerStatus() to verify ! * that the status was "OK" and push the worker back to IDLE state. * * XXX Rube Goldberg would be proud of this API, but no one else should be. * --- 1007,1016 ---- * 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 save the resulting status code and ! * switch the worker's state to WRKR_FINISHED. Later, caller must call ! * ReapWorkerStatus() to verify that the status was "OK" and push the worker ! * back to IDLE state. * * XXX Rube Goldberg would be proud of this API, but no one else should be. * *************** ListenToWorkers(ArchiveHandle *AH, Paral *** 963,988 **** if (messageStartsWith(msg, "OK ")) { TocEntry *te = pstate->parallelSlot[worker].args->te; - char *statusString; ! if (messageStartsWith(msg, "OK RESTORE ")) ! { ! statusString = msg + strlen("OK RESTORE "); ! pstate->parallelSlot[worker].status = ! (AH->MasterEndParallelItemPtr) ! (AH, te, statusString, ACT_RESTORE); ! } ! else if (messageStartsWith(msg, "OK DUMP ")) ! { ! statusString = msg + strlen("OK DUMP "); ! pstate->parallelSlot[worker].status = ! (AH->MasterEndParallelItemPtr) ! (AH, te, statusString, ACT_DUMP); ! } ! else ! exit_horribly(modulename, ! "invalid message received from worker: \"%s\"\n", ! msg); pstate->parallelSlot[worker].workerStatus = WRKR_FINISHED; } else --- 1039,1046 ---- if (messageStartsWith(msg, "OK ")) { TocEntry *te = pstate->parallelSlot[worker].args->te; ! pstate->parallelSlot[worker].status = parseWorkerResponse(AH, te, msg); pstate->parallelSlot[worker].workerStatus = WRKR_FINISHED; } else diff --git a/src/bin/pg_dump/pg_backup_archiver.h b/src/bin/pg_dump/pg_backup_archiver.h index 4aa7190..dcf1183 100644 *** a/src/bin/pg_dump/pg_backup_archiver.h --- b/src/bin/pg_dump/pg_backup_archiver.h *************** typedef void (*PrintTocDataPtr) (Archive *** 162,173 **** typedef void (*ClonePtr) (ArchiveHandle *AH); typedef void (*DeClonePtr) (ArchiveHandle *AH); ! typedef char *(*WorkerJobRestorePtr) (ArchiveHandle *AH, TocEntry *te); ! typedef char *(*WorkerJobDumpPtr) (ArchiveHandle *AH, TocEntry *te); ! typedef char *(*MasterStartParallelItemPtr) (ArchiveHandle *AH, TocEntry *te, ! T_Action act); ! typedef int (*MasterEndParallelItemPtr) (ArchiveHandle *AH, TocEntry *te, ! const char *str, T_Action act); typedef size_t (*CustomOutPtr) (ArchiveHandle *AH, const void *buf, size_t len); --- 162,169 ---- typedef void (*ClonePtr) (ArchiveHandle *AH); typedef void (*DeClonePtr) (ArchiveHandle *AH); ! typedef int (*WorkerJobDumpPtr) (ArchiveHandle *AH, TocEntry *te); ! typedef int (*WorkerJobRestorePtr) (ArchiveHandle *AH, TocEntry *te); typedef size_t (*CustomOutPtr) (ArchiveHandle *AH, const void *buf, size_t len); *************** struct _archiveHandle *** 267,275 **** StartBlobPtr StartBlobPtr; EndBlobPtr EndBlobPtr; - MasterStartParallelItemPtr MasterStartParallelItemPtr; - MasterEndParallelItemPtr MasterEndParallelItemPtr; - SetupWorkerPtr SetupWorkerPtr; WorkerJobDumpPtr WorkerJobDumpPtr; WorkerJobRestorePtr WorkerJobRestorePtr; --- 263,268 ---- diff --git a/src/bin/pg_dump/pg_backup_custom.c b/src/bin/pg_dump/pg_backup_custom.c index 66329dc..1f29067 100644 *** a/src/bin/pg_dump/pg_backup_custom.c --- b/src/bin/pg_dump/pg_backup_custom.c *************** static void _LoadBlobs(ArchiveHandle *AH *** 61,69 **** static void _Clone(ArchiveHandle *AH); static void _DeClone(ArchiveHandle *AH); ! static char *_MasterStartParallelItem(ArchiveHandle *AH, TocEntry *te, T_Action act); ! static int _MasterEndParallelItem(ArchiveHandle *AH, TocEntry *te, const char *str, T_Action act); ! char *_WorkerJobRestoreCustom(ArchiveHandle *AH, TocEntry *te); typedef struct { --- 61,67 ---- static void _Clone(ArchiveHandle *AH); static void _DeClone(ArchiveHandle *AH); ! static int _WorkerJobRestoreCustom(ArchiveHandle *AH, TocEntry *te); typedef struct { *************** InitArchiveFmt_Custom(ArchiveHandle *AH) *** 133,141 **** AH->ClonePtr = _Clone; AH->DeClonePtr = _DeClone; - AH->MasterStartParallelItemPtr = _MasterStartParallelItem; - AH->MasterEndParallelItemPtr = _MasterEndParallelItem; - /* no parallel dump in the custom archive, only parallel restore */ AH->WorkerJobDumpPtr = NULL; AH->WorkerJobRestorePtr = _WorkerJobRestoreCustom; --- 131,136 ---- *************** _DeClone(ArchiveHandle *AH) *** 808,884 **** } /* ! * This function is executed in the child of a parallel backup for the ! * custom format archive and dumps the actual data. */ ! char * _WorkerJobRestoreCustom(ArchiveHandle *AH, TocEntry *te) { - /* - * short fixed-size string + some ID so far, this needs to be malloc'ed - * instead of static because we work with threads on windows - */ - const int buflen = 64; - char *buf = (char *) pg_malloc(buflen); ParallelArgs pargs; - int status; pargs.AH = AH; pargs.te = te; ! status = parallel_restore(&pargs); ! ! snprintf(buf, buflen, "OK RESTORE %d %d %d", te->dumpId, status, ! status == WORKER_IGNORED_ERRORS ? AH->public.n_errors : 0); ! ! return buf; ! } ! ! /* ! * This function is executed in the parent process. Depending on the desired ! * action (dump or restore) it creates a string that is understood by the ! * _WorkerJobDump /_WorkerJobRestore functions of the dump format. ! */ ! static char * ! _MasterStartParallelItem(ArchiveHandle *AH, TocEntry *te, T_Action act) ! { ! /* ! * A static char is okay here, even on Windows because we call this ! * function only from one process (the master). ! */ ! static char buf[64]; /* short fixed-size string + number */ ! ! /* no parallel dump in the custom archive format */ ! Assert(act == ACT_RESTORE); ! ! snprintf(buf, sizeof(buf), "RESTORE %d", te->dumpId); ! ! return buf; ! } ! ! /* ! * This function is executed in the parent process. It analyzes the response of ! * the _WorkerJobDump / _WorkerJobRestore functions of the dump format. ! */ ! static int ! _MasterEndParallelItem(ArchiveHandle *AH, TocEntry *te, const char *str, T_Action act) ! { ! DumpId dumpId; ! int nBytes, ! status, ! n_errors; ! ! /* no parallel dump in the custom archive */ ! Assert(act == ACT_RESTORE); ! ! sscanf(str, "%d %d %d%n", &dumpId, &status, &n_errors, &nBytes); ! ! Assert(nBytes == strlen(str)); ! Assert(dumpId == te->dumpId); ! ! AH->public.n_errors += n_errors; ! ! return status; } /*-------------------------------------------------- --- 803,820 ---- } /* ! * This function is executed in the child of a parallel restore from a ! * custom-format archive and restores the actual data for one TOC entry. */ ! static int _WorkerJobRestoreCustom(ArchiveHandle *AH, TocEntry *te) { ParallelArgs pargs; pargs.AH = AH; pargs.te = te; ! return parallel_restore(&pargs); } /*-------------------------------------------------- diff --git a/src/bin/pg_dump/pg_backup_directory.c b/src/bin/pg_dump/pg_backup_directory.c index 27c6190..acfcb8a 100644 *** a/src/bin/pg_dump/pg_backup_directory.c --- b/src/bin/pg_dump/pg_backup_directory.c *************** static void _LoadBlobs(ArchiveHandle *AH *** 89,99 **** static void _Clone(ArchiveHandle *AH); static void _DeClone(ArchiveHandle *AH); ! static char *_MasterStartParallelItem(ArchiveHandle *AH, TocEntry *te, T_Action act); ! static int _MasterEndParallelItem(ArchiveHandle *AH, TocEntry *te, ! const char *str, T_Action act); ! static char *_WorkerJobRestoreDirectory(ArchiveHandle *AH, TocEntry *te); ! static char *_WorkerJobDumpDirectory(ArchiveHandle *AH, TocEntry *te); static void setFilePath(ArchiveHandle *AH, char *buf, const char *relativeFilename); --- 89,96 ---- static void _Clone(ArchiveHandle *AH); static void _DeClone(ArchiveHandle *AH); ! static int _WorkerJobRestoreDirectory(ArchiveHandle *AH, TocEntry *te); ! static int _WorkerJobDumpDirectory(ArchiveHandle *AH, TocEntry *te); static void setFilePath(ArchiveHandle *AH, char *buf, const char *relativeFilename); *************** InitArchiveFmt_Directory(ArchiveHandle * *** 140,148 **** AH->WorkerJobRestorePtr = _WorkerJobRestoreDirectory; AH->WorkerJobDumpPtr = _WorkerJobDumpDirectory; - AH->MasterStartParallelItemPtr = _MasterStartParallelItem; - AH->MasterEndParallelItemPtr = _MasterEndParallelItem; - /* Set up our private context */ ctx = (lclContext *) pg_malloc0(sizeof(lclContext)); AH->formatData = (void *) ctx; --- 137,142 ---- *************** _DeClone(ArchiveHandle *AH) *** 766,886 **** } /* ! * This function is executed in the parent process. Depending on the desired ! * action (dump or restore) it creates a string that is understood by the ! * _WorkerJobDump /_WorkerJobRestore functions of the dump format. ! */ ! static char * ! _MasterStartParallelItem(ArchiveHandle *AH, TocEntry *te, T_Action act) ! { ! /* ! * A static char is okay here, even on Windows because we call this ! * function only from one process (the master). ! */ ! static char buf[64]; ! ! if (act == ACT_DUMP) ! snprintf(buf, sizeof(buf), "DUMP %d", te->dumpId); ! else if (act == ACT_RESTORE) ! snprintf(buf, sizeof(buf), "RESTORE %d", te->dumpId); ! ! return buf; ! } ! ! /* ! * This function is executed in the child of a parallel backup for the ! * directory archive and dumps the actual data. ! * ! * We are currently returning only the DumpId so theoretically we could ! * make this function returning an int (or a DumpId). However, to ! * facilitate further enhancements and because sooner or later we need to ! * convert this to a string and send it via a message anyway, we stick with ! * char *. It is parsed on the other side by the _EndMasterParallel() ! * function of the respective dump format. */ ! static char * _WorkerJobDumpDirectory(ArchiveHandle *AH, TocEntry *te) { /* - * short fixed-size string + some ID so far, this needs to be malloc'ed - * instead of static because we work with threads on windows - */ - const int buflen = 64; - char *buf = (char *) pg_malloc(buflen); - lclTocEntry *tctx = (lclTocEntry *) te->formatData; - - /* This should never happen */ - if (!tctx) - exit_horribly(modulename, "error during backup\n"); - - /* * This function returns void. We either fail and die horribly or * succeed... A failure will be detected by the parent when the child dies * unexpectedly. */ WriteDataChunksForTocEntry(AH, te); ! snprintf(buf, buflen, "OK DUMP %d", te->dumpId); ! ! return buf; } /* ! * This function is executed in the child of a parallel backup for the ! * directory archive and dumps the actual data. */ ! static char * _WorkerJobRestoreDirectory(ArchiveHandle *AH, TocEntry *te) { - /* - * short fixed-size string + some ID so far, this needs to be malloc'ed - * instead of static because we work with threads on windows - */ - const int buflen = 64; - char *buf = (char *) pg_malloc(buflen); ParallelArgs pargs; - int status; pargs.AH = AH; pargs.te = te; ! status = parallel_restore(&pargs); ! ! snprintf(buf, buflen, "OK RESTORE %d %d %d", te->dumpId, status, ! status == WORKER_IGNORED_ERRORS ? AH->public.n_errors : 0); ! ! return buf; ! } ! ! /* ! * This function is executed in the parent process. It analyzes the response of ! * the _WorkerJobDumpDirectory/_WorkerJobRestoreDirectory functions of the ! * respective dump format. ! */ ! static int ! _MasterEndParallelItem(ArchiveHandle *AH, TocEntry *te, const char *str, T_Action act) ! { ! DumpId dumpId; ! int nBytes, ! n_errors; ! int status = 0; ! ! if (act == ACT_DUMP) ! { ! sscanf(str, "%d%n", &dumpId, &nBytes); ! ! Assert(dumpId == te->dumpId); ! Assert(nBytes == strlen(str)); ! } ! else if (act == ACT_RESTORE) ! { ! sscanf(str, "%d %d %d%n", &dumpId, &status, &n_errors, &nBytes); ! ! Assert(dumpId == te->dumpId); ! Assert(nBytes == strlen(str)); ! ! AH->public.n_errors += n_errors; ! } ! ! return status; } --- 760,792 ---- } /* ! * This function is executed in the child of a parallel backup for a ! * directory-format archive and dumps the actual data for one TOC entry. */ ! static int _WorkerJobDumpDirectory(ArchiveHandle *AH, TocEntry *te) { /* * This function returns void. We either fail and die horribly or * succeed... A failure will be detected by the parent when the child dies * unexpectedly. */ WriteDataChunksForTocEntry(AH, te); ! return 0; } /* ! * This function is executed in the child of a parallel restore from a ! * directory-format archive and restores the actual data for one TOC entry. */ ! static int _WorkerJobRestoreDirectory(ArchiveHandle *AH, TocEntry *te) { ParallelArgs pargs; pargs.AH = AH; pargs.te = te; ! return parallel_restore(&pargs); } diff --git a/src/bin/pg_dump/pg_backup_tar.c b/src/bin/pg_dump/pg_backup_tar.c index 8dfc6a9..9cadd0c 100644 *** a/src/bin/pg_dump/pg_backup_tar.c --- b/src/bin/pg_dump/pg_backup_tar.c *************** InitArchiveFmt_Tar(ArchiveHandle *AH) *** 152,160 **** AH->ClonePtr = NULL; AH->DeClonePtr = NULL; - AH->MasterStartParallelItemPtr = NULL; - AH->MasterEndParallelItemPtr = NULL; - AH->WorkerJobDumpPtr = NULL; AH->WorkerJobRestorePtr = NULL; --- 152,157 ----