public inbox for [email protected]help / color / mirror / Atom feed
[PATCH v45 3/7] Make archiver process an auxiliary process 2+ messages / 2 participants [nested] [flat]
* [PATCH v45 3/7] Make archiver process an auxiliary process @ 2020-03-13 07:59 Kyotaro Horiguchi <[email protected]> 0 siblings, 0 replies; 2+ messages in thread From: Kyotaro Horiguchi @ 2020-03-13 07:59 UTC (permalink / raw) This is a preliminary patch for shared-memory based stats collector. Archiver process must be a auxiliary process since it uses shared memory after stats data was moved into shared-memory. Make the process an auxiliary process in order to make it work. --- src/backend/access/transam/xlogarchive.c | 6 +- src/backend/bootstrap/bootstrap.c | 22 ++-- src/backend/postmaster/pgarch.c | 130 +++-------------------- src/backend/postmaster/postmaster.c | 50 +++++---- src/backend/storage/lmgr/proc.c | 1 + src/include/access/xlog.h | 3 + src/include/access/xlogarchive.h | 1 + src/include/miscadmin.h | 2 + src/include/postmaster/pgarch.h | 4 +- src/include/storage/pmsignal.h | 1 - src/include/storage/proc.h | 3 + 11 files changed, 69 insertions(+), 154 deletions(-) diff --git a/src/backend/access/transam/xlogarchive.c b/src/backend/access/transam/xlogarchive.c index 1c5a4f8b5a..d01859bde5 100644 --- a/src/backend/access/transam/xlogarchive.c +++ b/src/backend/access/transam/xlogarchive.c @@ -29,7 +29,9 @@ #include "storage/fd.h" #include "storage/ipc.h" #include "storage/lwlock.h" +#include "storage/latch.h" #include "storage/pmsignal.h" +#include "storage/proc.h" /* * Attempt to retrieve the specified file from off-line archival storage. @@ -490,8 +492,8 @@ XLogArchiveNotify(const char *xlog) } /* Notify archiver that it's got something to do */ - if (IsUnderPostmaster) - SendPostmasterSignal(PMSIGNAL_WAKEN_ARCHIVER); + if (IsUnderPostmaster && ProcGlobal->archiverLatch) + SetLatch(ProcGlobal->archiverLatch); } /* diff --git a/src/backend/bootstrap/bootstrap.c b/src/backend/bootstrap/bootstrap.c index 6f615e6622..41da0c5059 100644 --- a/src/backend/bootstrap/bootstrap.c +++ b/src/backend/bootstrap/bootstrap.c @@ -317,6 +317,9 @@ AuxiliaryProcessMain(int argc, char *argv[]) case StartupProcess: MyBackendType = B_STARTUP; break; + case ArchiverProcess: + MyBackendType = B_ARCHIVER; + break; case BgWriterProcess: MyBackendType = B_BG_WRITER; break; @@ -437,30 +440,29 @@ AuxiliaryProcessMain(int argc, char *argv[]) proc_exit(1); /* should never return */ case StartupProcess: - /* don't set signals, startup process has its own agenda */ StartupProcessMain(); - proc_exit(1); /* should never return */ + proc_exit(1); + + case ArchiverProcess: + PgArchiverMain(); + proc_exit(1); case BgWriterProcess: - /* don't set signals, bgwriter has its own agenda */ BackgroundWriterMain(); - proc_exit(1); /* should never return */ + proc_exit(1); case CheckpointerProcess: - /* don't set signals, checkpointer has its own agenda */ CheckpointerMain(); - proc_exit(1); /* should never return */ + proc_exit(1); case WalWriterProcess: - /* don't set signals, walwriter has its own agenda */ InitXLOGAccess(); WalWriterMain(); - proc_exit(1); /* should never return */ + proc_exit(1); case WalReceiverProcess: - /* don't set signals, walreceiver has its own agenda */ WalReceiverMain(); - proc_exit(1); /* should never return */ + proc_exit(1); default: elog(PANIC, "unrecognized process type: %d", (int) MyAuxProcType); diff --git a/src/backend/postmaster/pgarch.c b/src/backend/postmaster/pgarch.c index edec311f12..9a2e21bf86 100644 --- a/src/backend/postmaster/pgarch.c +++ b/src/backend/postmaster/pgarch.c @@ -48,6 +48,7 @@ #include "storage/latch.h" #include "storage/pg_shmem.h" #include "storage/pmsignal.h" +#include "storage/procsignal.h" #include "utils/guc.h" #include "utils/ps_status.h" @@ -78,13 +79,11 @@ * Local data * ---------- */ -static time_t last_pgarch_start_time; static time_t last_sigterm_time = 0; /* * Flags set by interrupt handlers for later service in the main loop. */ -static volatile sig_atomic_t wakened = false; static volatile sig_atomic_t ready_to_stop = false; /* ---------- @@ -95,8 +94,6 @@ static volatile sig_atomic_t ready_to_stop = false; static pid_t pgarch_forkexec(void); #endif -NON_EXEC_STATIC void PgArchiverMain(int argc, char *argv[]) pg_attribute_noreturn(); -static void pgarch_waken(SIGNAL_ARGS); static void pgarch_waken_stop(SIGNAL_ARGS); static void pgarch_MainLoop(void); static void pgarch_ArchiverCopyLoop(void); @@ -110,75 +107,6 @@ static void pgarch_archiveDone(char *xlog); * ------------------------------------------------------------ */ -/* - * pgarch_start - * - * Called from postmaster at startup or after an existing archiver - * died. Attempt to fire up a fresh archiver process. - * - * Returns PID of child process, or 0 if fail. - * - * Note: if fail, we will be called again from the postmaster main loop. - */ -int -pgarch_start(void) -{ - time_t curtime; - pid_t pgArchPid; - - /* - * Do nothing if no archiver needed - */ - if (!XLogArchivingActive()) - return 0; - - /* - * Do nothing if too soon since last archiver start. This is a safety - * valve to protect against continuous respawn attempts if the archiver is - * dying immediately at launch. Note that since we will be re-called from - * the postmaster main loop, we will get another chance later. - */ - curtime = time(NULL); - if ((unsigned int) (curtime - last_pgarch_start_time) < - (unsigned int) PGARCH_RESTART_INTERVAL) - return 0; - last_pgarch_start_time = curtime; - -#ifdef EXEC_BACKEND - switch ((pgArchPid = pgarch_forkexec())) -#else - switch ((pgArchPid = fork_process())) -#endif - { - case -1: - ereport(LOG, - (errmsg("could not fork archiver: %m"))); - return 0; - -#ifndef EXEC_BACKEND - case 0: - /* in postmaster child ... */ - InitPostmasterChild(); - - /* Close the postmaster's sockets */ - ClosePostmasterPorts(false); - - /* Drop our connection to postmaster's shared memory, as well */ - dsm_detach_all(); - PGSharedMemoryDetach(); - - PgArchiverMain(0, NULL); - break; -#endif - - default: - return (int) pgArchPid; - } - - /* shouldn't get here */ - return 0; -} - /* ------------------------------------------------------------ * Local functions called by archiver follow * ------------------------------------------------------------ @@ -212,14 +140,9 @@ pgarch_forkexec(void) #endif /* EXEC_BACKEND */ -/* - * PgArchiverMain - * - * The argc/argv parameters are valid only in EXEC_BACKEND case. However, - * since we don't use 'em, it hardly matters... - */ -NON_EXEC_STATIC void -PgArchiverMain(int argc, char *argv[]) +/* Main entry point for archiver process */ +void +PgArchiverMain(void) { /* * Ignore all signals usually bound to some action in the postmaster, @@ -231,33 +154,26 @@ PgArchiverMain(int argc, char *argv[]) /* SIGQUIT handler was already set up by InitPostmasterChild */ pqsignal(SIGALRM, SIG_IGN); pqsignal(SIGPIPE, SIG_IGN); - pqsignal(SIGUSR1, pgarch_waken); + pqsignal(SIGUSR1, procsignal_sigusr1_handler); pqsignal(SIGUSR2, pgarch_waken_stop); + /* Reset some signals that are accepted by postmaster but not here */ pqsignal(SIGCHLD, SIG_DFL); + + /* Unblock signals (they were blocked when the postmaster forked us) */ PG_SETMASK(&UnBlockSig); - MyBackendType = B_ARCHIVER; - init_ps_display(NULL); + /* + * Advertise our latch that backends can use to wake us up while we're + * sleeping. + */ + ProcGlobal->archiverLatch = &MyProc->procLatch; pgarch_MainLoop(); exit(0); } -/* SIGUSR1 signal handler for archiver process */ -static void -pgarch_waken(SIGNAL_ARGS) -{ - int save_errno = errno; - - /* set flag that there is work to be done */ - wakened = true; - SetLatch(MyLatch); - - errno = save_errno; -} - /* SIGUSR2 signal handler for archiver process */ static void pgarch_waken_stop(SIGNAL_ARGS) @@ -282,14 +198,6 @@ pgarch_MainLoop(void) pg_time_t last_copy_time = 0; bool time_to_stop; - /* - * We run the copy loop immediately upon entry, in case there are - * unarchived files left over from a previous database run (or maybe the - * archiver died unexpectedly). After that we wait for a signal or - * timeout before doing more. - */ - wakened = true; - /* * There shouldn't be anything for the archiver to do except to wait for a * signal ... however, the archiver exists to protect our data, so she @@ -328,12 +236,8 @@ pgarch_MainLoop(void) } /* Do what we're here for */ - if (wakened || time_to_stop) - { - wakened = false; - pgarch_ArchiverCopyLoop(); - last_copy_time = time(NULL); - } + pgarch_ArchiverCopyLoop(); + last_copy_time = time(NULL); /* * Sleep until a signal is received, or until a poll is forced by @@ -354,13 +258,9 @@ pgarch_MainLoop(void) WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, timeout * 1000L, WAIT_EVENT_ARCHIVER_MAIN); - if (rc & WL_TIMEOUT) - wakened = true; if (rc & WL_POSTMASTER_DEATH) time_to_stop = true; } - else - wakened = true; } /* diff --git a/src/backend/postmaster/postmaster.c b/src/backend/postmaster/postmaster.c index 7de27ee4e0..af91c313e2 100644 --- a/src/backend/postmaster/postmaster.c +++ b/src/backend/postmaster/postmaster.c @@ -548,6 +548,7 @@ static void ShmemBackendArrayRemove(Backend *bn); #endif /* EXEC_BACKEND */ #define StartupDataBase() StartChildProcess(StartupProcess) +#define StartArchiver() StartChildProcess(ArchiverProcess) #define StartBackgroundWriter() StartChildProcess(BgWriterProcess) #define StartCheckpointer() StartChildProcess(CheckpointerProcess) #define StartWalWriter() StartChildProcess(WalWriterProcess) @@ -1788,7 +1789,7 @@ ServerLoop(void) /* If we have lost the archiver, try to start a new one. */ if (PgArchPID == 0 && PgArchStartupAllowed()) - PgArchPID = pgarch_start(); + PgArchPID = StartArchiver(); /* If we need to signal the autovacuum launcher, do so now */ if (avlauncher_needs_signal) @@ -3027,7 +3028,7 @@ reaper(SIGNAL_ARGS) if (!IsBinaryUpgrade && AutoVacuumingActive() && AutoVacPID == 0) AutoVacPID = StartAutoVacLauncher(); if (PgArchStartupAllowed() && PgArchPID == 0) - PgArchPID = pgarch_start(); + PgArchPID = StartArchiver(); if (PgStatPID == 0) PgStatPID = pgstat_start(); @@ -3162,20 +3163,16 @@ reaper(SIGNAL_ARGS) } /* - * Was it the archiver? If so, just try to start a new one; no need - * to force reset of the rest of the system. (If fail, we'll try - * again in future cycles of the main loop.). Unless we were waiting - * for it to shut down; don't restart it in that case, and - * PostmasterStateMachine() will advance to the next shutdown step. + * Was it the archiver? Normal exit can be ignored; we'll start a new + * one at the next iteration of the postmaster's main loop, if + * necessary. Any other exit condition is treated as a crash. */ if (pid == PgArchPID) { PgArchPID = 0; if (!EXIT_STATUS_0(exitstatus)) - LogChildExit(LOG, _("archiver process"), - pid, exitstatus); - if (PgArchStartupAllowed()) - PgArchPID = pgarch_start(); + HandleChildCrash(pid, exitstatus, + _("archiver process")); continue; } @@ -3423,7 +3420,7 @@ CleanupBackend(int pid, /* * HandleChildCrash -- cleanup after failed backend, bgwriter, checkpointer, - * walwriter, autovacuum, or background worker. + * walwriter, autovacuum, archiver or background worker. * * The objectives here are to clean up our local state about the child * process, and to signal all other remaining children to quickdie. @@ -3629,6 +3626,18 @@ HandleChildCrash(int pid, int exitstatus, const char *procname) signal_child(AutoVacPID, (SendStop ? SIGSTOP : SIGQUIT)); } + /* Take care of the archiver too */ + if (pid == PgArchPID) + PgArchPID = 0; + else if (PgArchPID != 0 && take_action) + { + ereport(DEBUG2, + (errmsg_internal("sending %s to process %d", + (SendStop ? "SIGSTOP" : "SIGQUIT"), + (int) PgArchPID))); + signal_child(PgArchPID, (SendStop ? SIGSTOP : SIGQUIT)); + } + /* * Force a power-cycle of the pgarch process too. (This isn't absolutely * necessary, but it seems like a good idea for robustness, and it @@ -3932,6 +3941,7 @@ PostmasterStateMachine(void) Assert(CheckpointerPID == 0); Assert(WalWriterPID == 0); Assert(AutoVacPID == 0); + Assert(PgArchPID == 0); /* syslogger is not considered here */ pmState = PM_NO_CHILDREN; } @@ -5160,7 +5170,7 @@ sigusr1_handler(SIGNAL_ARGS) */ Assert(PgArchPID == 0); if (XLogArchivingAlways()) - PgArchPID = pgarch_start(); + PgArchPID = StartArchiver(); /* * If we aren't planning to enter hot standby mode later, treat @@ -5214,16 +5224,6 @@ sigusr1_handler(SIGNAL_ARGS) if (StartWorkerNeeded || HaveCrashedWorker) maybe_start_bgworkers(); - if (CheckPostmasterSignal(PMSIGNAL_WAKEN_ARCHIVER) && - PgArchPID != 0) - { - /* - * Send SIGUSR1 to archiver process, to wake it up and begin archiving - * next WAL file. - */ - signal_child(PgArchPID, SIGUSR1); - } - /* Tell syslogger to rotate logfile if requested */ if (SysLoggerPID != 0) { @@ -5465,6 +5465,10 @@ StartChildProcess(AuxProcType type) ereport(LOG, (errmsg("could not fork startup process: %m"))); break; + case ArchiverProcess: + ereport(LOG, + (errmsg("could not fork archiver process: %m"))); + break; case BgWriterProcess: ereport(LOG, (errmsg("could not fork background writer process: %m"))); diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c index db0cfaa360..aabf9d73eb 100644 --- a/src/backend/storage/lmgr/proc.c +++ b/src/backend/storage/lmgr/proc.c @@ -182,6 +182,7 @@ InitProcGlobal(void) ProcGlobal->startupBufferPinWaitBufId = -1; ProcGlobal->walwriterLatch = NULL; ProcGlobal->checkpointerLatch = NULL; + ProcGlobal->archiverLatch = NULL; pg_atomic_init_u32(&ProcGlobal->procArrayGroupFirst, INVALID_PGPROCNO); pg_atomic_init_u32(&ProcGlobal->clogGroupFirst, INVALID_PGPROCNO); diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index 75ec1073bd..551f518cc2 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -339,6 +339,9 @@ extern XLogRecPtr GetRedoRecPtr(void); extern XLogRecPtr GetInsertRecPtr(void); extern XLogRecPtr GetFlushRecPtr(void); extern XLogRecPtr GetLastImportantRecPtr(void); +extern void XLogArchiveWakeupStart(void); +extern void XLogArchiveWakeupEnd(void); +extern void XLogArchiveWakeup(void); extern void RemovePromoteSignalFiles(void); extern bool PromoteIsTriggered(void); diff --git a/src/include/access/xlogarchive.h b/src/include/access/xlogarchive.h index 3edd1a976c..1a59181cf9 100644 --- a/src/include/access/xlogarchive.h +++ b/src/include/access/xlogarchive.h @@ -25,6 +25,7 @@ extern void ExecuteRecoveryCommand(const char *command, const char *commandName, extern void KeepFileRestoredFromArchive(const char *path, const char *xlogfname); extern void XLogArchiveNotify(const char *xlog); extern void XLogArchiveNotifySeg(XLogSegNo segno); +extern void XLogArchiveWakeup(void); extern void XLogArchiveForceDone(const char *xlog); extern bool XLogArchiveCheckDone(const char *xlog); extern bool XLogArchiveIsBusy(const char *xlog); diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h index 1bdc97e308..adb9f819bb 100644 --- a/src/include/miscadmin.h +++ b/src/include/miscadmin.h @@ -419,6 +419,7 @@ typedef enum BootstrapProcess, StartupProcess, BgWriterProcess, + ArchiverProcess, CheckpointerProcess, WalWriterProcess, WalReceiverProcess, @@ -431,6 +432,7 @@ extern AuxProcType MyAuxProcType; #define AmBootstrapProcess() (MyAuxProcType == BootstrapProcess) #define AmStartupProcess() (MyAuxProcType == StartupProcess) #define AmBackgroundWriterProcess() (MyAuxProcType == BgWriterProcess) +#define AmArchiverProcess() (MyAuxProcType == ArchiverProcess) #define AmCheckpointerProcess() (MyAuxProcType == CheckpointerProcess) #define AmWalWriterProcess() (MyAuxProcType == WalWriterProcess) #define AmWalReceiverProcess() (MyAuxProcType == WalReceiverProcess) diff --git a/src/include/postmaster/pgarch.h b/src/include/postmaster/pgarch.h index d102a21ab7..385b002dfe 100644 --- a/src/include/postmaster/pgarch.h +++ b/src/include/postmaster/pgarch.h @@ -32,8 +32,6 @@ */ extern int pgarch_start(void); -#ifdef EXEC_BACKEND -extern void PgArchiverMain(int argc, char *argv[]) pg_attribute_noreturn(); -#endif +extern void PgArchiverMain(void) pg_attribute_noreturn(); #endif /* _PGARCH_H */ diff --git a/src/include/storage/pmsignal.h b/src/include/storage/pmsignal.h index dbbed18f61..8ed4d87ae6 100644 --- a/src/include/storage/pmsignal.h +++ b/src/include/storage/pmsignal.h @@ -34,7 +34,6 @@ typedef enum { PMSIGNAL_RECOVERY_STARTED, /* recovery has started */ PMSIGNAL_BEGIN_HOT_STANDBY, /* begin Hot Standby */ - PMSIGNAL_WAKEN_ARCHIVER, /* send a NOTIFY signal to xlog archiver */ PMSIGNAL_ROTATE_LOGFILE, /* send SIGUSR1 to syslogger to rotate logfile */ PMSIGNAL_START_AUTOVAC_LAUNCHER, /* start an autovacuum launcher */ PMSIGNAL_START_AUTOVAC_WORKER, /* start an autovacuum worker */ diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h index 0786fcf103..430d438303 100644 --- a/src/include/storage/proc.h +++ b/src/include/storage/proc.h @@ -354,6 +354,9 @@ typedef struct PROC_HDR int startupProcPid; /* Buffer id of the buffer that Startup process waits for pin on, or -1 */ int startupBufferPinWaitBufId; + /* Archiver process's latch */ + Latch *archiverLatch; + /* Current shared estimate of appropriate spins_per_delay value */ } PROC_HDR; extern PGDLLIMPORT PROC_HDR *ProcGlobal; -- 2.27.0 ----Next_Part(Fri_Jan__8_10_24_34_2021_185)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v45-0004-Shared-memory-based-stats-collector.patch" ^ permalink raw reply [nested|flat] 2+ messages in thread
* Re: pg_upgrade parallelism @ 2022-01-12 04:51 Jaime Casanova <[email protected]> 0 siblings, 0 replies; 2+ messages in thread From: Jaime Casanova @ 2022-01-12 04:51 UTC (permalink / raw) To: Jacob Champion <[email protected]>; +Cc: pgsql-hackers On Wed, Nov 17, 2021 at 08:04:41PM +0000, Jacob Champion wrote: > On Wed, 2021-11-17 at 14:44 -0500, Jaime Casanova wrote: > > I'm trying to add more parallelism by copying individual segments > > of a relfilenode in different processes. Does anyone one see a big > > problem in trying to do that? I'm asking because no one did it before, > > that could not be a good sign. > > I looked into speeding this up a while back, too. For the use case I > was looking at -- Greenplum, which has huge numbers of relfilenodes -- > spinning disk I/O was absolutely the bottleneck and that is typically > not easily parallelizable. (In fact I felt at the time that Andres' > work on async I/O might be a better way forward, at least for some > filesystems.) > > But you mentioned that you were seeing disks that weren't saturated, so > maybe some CPU optimization is still valuable? I am a little skeptical > that more parallelism is the way to do that, but numbers trump my > skepticism. > Sorry for being unresponsive too long. I did add a new --jobs-per-disk option, this is a simple patch I made for the customer and ignored all WIN32 parts because I don't know anything about that part. I was wanting to complete that part but it has been in the same state two months now. AFAIU, it seems there is a different struct for the parameters of the function that will be called on the thread. I also decided to create a new reap_*_child() function for using with the new parameter. Now, the customer went from copy 25Tb in 6 hours to 4h 45min, which is an improvement of 20%! > > - why we read()/write() at all? is not a faster way of copying the file? > > i'm asking that because i don't actually know. > > I have idly wondered if something based on splice() would be faster, > but I haven't actually tried it. > I tried and got no better result. > But there is now support for copy-on-write with the clone mode, isn't > there? Or are you not able to take advantage of it? > That's sadly not possible because those are different disks, and yes I know that's something that pg_upgrade normally doesn't allow but is not difficult to make it happen. -- Jaime Casanova Director de Servicios Profesionales SystemGuards - Consultores de PostgreSQL Attachments: [text/x-diff] 0001-Add-jobs-per-disk-option-to-allow-multiple-processes.patch (8.9K, ../../Yd5eOwB6uBQdA11T@ahch-to/2-0001-Add-jobs-per-disk-option-to-allow-multiple-processes.patch) download | inline diff: From 0d04f79cb51d6be0ced9c6561cfca5bfe18c4bdd Mon Sep 17 00:00:00 2001 From: Jaime Casanova <[email protected]> Date: Wed, 15 Dec 2021 12:14:44 -0500 Subject: [PATCH] Add --jobs-per-disk option to allow multiple processes per tablespace This option is independent of the --jobs one. It's will fork new processes to copy the different segments of a relfilenode in parallel. --- src/bin/pg_upgrade/option.c | 8 ++- src/bin/pg_upgrade/parallel.c | 93 ++++++++++++++++++++++++++++++++ src/bin/pg_upgrade/pg_upgrade.h | 4 ++ src/bin/pg_upgrade/relfilenode.c | 59 +++++++++++--------- 4 files changed, 139 insertions(+), 25 deletions(-) diff --git a/src/bin/pg_upgrade/option.c b/src/bin/pg_upgrade/option.c index 66fe16964e..46b1913a42 100644 --- a/src/bin/pg_upgrade/option.c +++ b/src/bin/pg_upgrade/option.c @@ -54,6 +54,7 @@ parseCommandLine(int argc, char *argv[]) {"link", no_argument, NULL, 'k'}, {"retain", no_argument, NULL, 'r'}, {"jobs", required_argument, NULL, 'j'}, + {"jobs-per-disks", required_argument, NULL, 'J'}, {"socketdir", required_argument, NULL, 's'}, {"verbose", no_argument, NULL, 'v'}, {"clone", no_argument, NULL, 1}, @@ -103,7 +104,7 @@ parseCommandLine(int argc, char *argv[]) if (os_user_effective_id == 0) pg_fatal("%s: cannot be run as root\n", os_info.progname); - while ((option = getopt_long(argc, argv, "d:D:b:B:cj:kNo:O:p:P:rs:U:v", + while ((option = getopt_long(argc, argv, "d:D:b:B:cj:J:kNo:O:p:P:rs:U:v", long_options, &optindex)) != -1) { switch (option) @@ -132,6 +133,10 @@ parseCommandLine(int argc, char *argv[]) user_opts.jobs = atoi(optarg); break; + case 'J': + user_opts.jobs_per_disk = atoi(optarg); + break; + case 'k': user_opts.transfer_mode = TRANSFER_MODE_LINK; break; @@ -291,6 +296,7 @@ usage(void) printf(_(" -d, --old-datadir=DATADIR old cluster data directory\n")); printf(_(" -D, --new-datadir=DATADIR new cluster data directory\n")); printf(_(" -j, --jobs=NUM number of simultaneous processes or threads to use\n")); + printf(_(" -J, --jobs_per_disk=NUM number of simultaneous processes or threads to use per tablespace\n")); printf(_(" -k, --link link instead of copying files to new cluster\n")); printf(_(" -N, --no-sync do not wait for changes to be written safely to disk\n")); printf(_(" -o, --old-options=OPTIONS old cluster options to pass to the server\n")); diff --git a/src/bin/pg_upgrade/parallel.c b/src/bin/pg_upgrade/parallel.c index ee7364da3b..82f698a9ab 100644 --- a/src/bin/pg_upgrade/parallel.c +++ b/src/bin/pg_upgrade/parallel.c @@ -17,6 +17,9 @@ #include "pg_upgrade.h" static int parallel_jobs; +static int current_jobs = 0; + +static bool reap_subchild(bool wait_for_child); #ifdef WIN32 /* @@ -277,6 +280,60 @@ win32_transfer_all_new_dbs(transfer_thread_arg *args) #endif + +/* + * parallel_process_relfile_segment() + * + * Copy or link file from old cluster to new one. If vm_must_add_frozenbit + * is true, visibility map forks are converted and rewritten, even in link + * mode. + */ +void +parallel_process_relfile_segment(FileNameMap *map, const char *type_suffix, bool vm_must_add_frozenbit, const char *old_file, const char *new_file) +{ +#ifndef WIN32 + pid_t child; +#else + HANDLE child; + transfer_thread_arg *new_arg; +#endif + if (user_opts.jobs <= 1 || user_opts.jobs_per_disk <= 1) + process_relfile_segment(map, type_suffix, vm_must_add_frozenbit, old_file, new_file); + else + { + /* parallel */ + + /* harvest any dead children */ + while (reap_subchild(false) == true) + ; + + /* must we wait for a dead child? use a maximum of 3 childs per tablespace */ + if (current_jobs >= user_opts.jobs_per_disk) + reap_subchild(true); + + /* set this before we start the job */ + current_jobs++; + + /* Ensure stdio state is quiesced before forking */ + fflush(NULL); + +#ifndef WIN32 + child = fork(); + if (child == 0) + { + process_relfile_segment(map, type_suffix, vm_must_add_frozenbit, old_file, new_file); + /* use _exit to skip atexit() functions */ + _exit(0); + } + else if (child < 0) + /* fork failed */ + pg_fatal("could not create worker process: %s\n", strerror(errno)); +#endif + } +} + + + /* * collect status from a completed worker child */ @@ -345,3 +402,39 @@ reap_child(bool wait_for_child) return true; } + + + + +/* + * collect status from a completed worker subchild + */ +static bool +reap_subchild(bool wait_for_child) +{ +#ifndef WIN32 + int work_status; + pid_t child; +#else + int thread_num; + DWORD res; +#endif + + if (user_opts.jobs <= 1 || current_jobs == 0) + return false; + +#ifndef WIN32 + child = waitpid(-1, &work_status, wait_for_child ? 0 : WNOHANG); + if (child == (pid_t) -1) + pg_fatal("waitpid() failed: %s\n", strerror(errno)); + if (child == 0) + return false; /* no children, or no dead children */ + if (work_status != 0) + pg_fatal("child process exited abnormally: status %d\n", work_status); +#endif + + /* do this after job has been removed */ + current_jobs--; + + return true; +} diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.h index 22169f1002..adcb24ffea 100644 --- a/src/bin/pg_upgrade/pg_upgrade.h +++ b/src/bin/pg_upgrade/pg_upgrade.h @@ -282,6 +282,7 @@ typedef struct bool do_sync; /* flush changes to disk */ transferMode transfer_mode; /* copy files or link them? */ int jobs; /* number of processes/threads to use */ + int jobs_per_disk; /* number of processes/threads to use */ char *socketdir; /* directory to use for Unix sockets */ } UserOpts; @@ -450,4 +451,7 @@ void parallel_exec_prog(const char *log_file, const char *opt_log_file, void parallel_transfer_all_new_dbs(DbInfoArr *old_db_arr, DbInfoArr *new_db_arr, char *old_pgdata, char *new_pgdata, char *old_tablespace); + +void process_relfile_segment(FileNameMap *map, const char *suffix, bool vm_must_add_frozenbit, const char *old_file, const char *new_file); +void parallel_process_relfile_segment(FileNameMap *map, const char *suffix, bool vm_must_add_frozenbit, const char *old_file, const char *new_file); bool reap_child(bool wait_for_child); diff --git a/src/bin/pg_upgrade/relfilenode.c b/src/bin/pg_upgrade/relfilenode.c index 5dbefbceaf..8a7c49efaa 100644 --- a/src/bin/pg_upgrade/relfilenode.c +++ b/src/bin/pg_upgrade/relfilenode.c @@ -17,6 +17,7 @@ static void transfer_single_new_db(FileNameMap *maps, int size, char *old_tablespace); static void transfer_relfile(FileNameMap *map, const char *suffix, bool vm_must_add_frozenbit); +void process_relfile_segment(FileNameMap *map, const char *suffix, bool vm_must_add_frozenbit, const char *old_file, const char *new_file); /* @@ -232,30 +233,40 @@ transfer_relfile(FileNameMap *map, const char *type_suffix, bool vm_must_add_fro /* Copying files might take some time, so give feedback. */ pg_log(PG_STATUS, "%s", old_file); - if (vm_must_add_frozenbit && strcmp(type_suffix, "_vm") == 0) + parallel_process_relfile_segment(map, type_suffix, vm_must_add_frozenbit, old_file, new_file); + } +} + + + +void +process_relfile_segment(FileNameMap *map, const char *type_suffix, bool vm_must_add_frozenbit, const char *old_file, const char *new_file) +{ + + if (vm_must_add_frozenbit && strcmp(type_suffix, "_vm") == 0) + { + /* Need to rewrite visibility map format */ + pg_log(PG_VERBOSE, "rewriting \"%s\" to \"%s\"\n", + old_file, new_file); + rewriteVisibilityMap(old_file, new_file, map->nspname, map->relname); + } + else + switch (user_opts.transfer_mode) { - /* Need to rewrite visibility map format */ - pg_log(PG_VERBOSE, "rewriting \"%s\" to \"%s\"\n", - old_file, new_file); - rewriteVisibilityMap(old_file, new_file, map->nspname, map->relname); + case TRANSFER_MODE_CLONE: + pg_log(PG_VERBOSE, "cloning \"%s\" to \"%s\"\n", + old_file, new_file); + cloneFile(old_file, new_file, map->nspname, map->relname); + break; + case TRANSFER_MODE_COPY: + pg_log(PG_VERBOSE, "copying \"%s\" to \"%s\"\n", + old_file, new_file); + copyFile(old_file, new_file, map->nspname, map->relname); + break; + case TRANSFER_MODE_LINK: + pg_log(PG_VERBOSE, "linking \"%s\" to \"%s\"\n", + old_file, new_file); + linkFile(old_file, new_file, map->nspname, map->relname); + break; } - else - switch (user_opts.transfer_mode) - { - case TRANSFER_MODE_CLONE: - pg_log(PG_VERBOSE, "cloning \"%s\" to \"%s\"\n", - old_file, new_file); - cloneFile(old_file, new_file, map->nspname, map->relname); - break; - case TRANSFER_MODE_COPY: - pg_log(PG_VERBOSE, "copying \"%s\" to \"%s\"\n", - old_file, new_file); - copyFile(old_file, new_file, map->nspname, map->relname); - break; - case TRANSFER_MODE_LINK: - pg_log(PG_VERBOSE, "linking \"%s\" to \"%s\"\n", - old_file, new_file); - linkFile(old_file, new_file, map->nspname, map->relname); - } - } } -- 2.20.1 ^ permalink raw reply [nested|flat] 2+ messages in thread
end of thread, other threads:[~2022-01-12 04:51 UTC | newest] Thread overview: 2+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2020-03-13 07:59 [PATCH v45 3/7] Make archiver process an auxiliary process Kyotaro Horiguchi <[email protected]> 2022-01-12 04:51 Re: pg_upgrade parallelism Jaime Casanova <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox