From 27bdac00ef8da7a01c401f2f10b4017c9ee0b1ba Mon Sep 17 00:00:00 2001 From: Andres Freund Date: Mon, 17 Feb 2025 17:57:00 -0500 Subject: [PATCH v2.4 11/29] aio: Skeleton IO worker infrastructure This doesn't do anything useful on its own, but the code that needs to be touched is independent of other changes. Remarks: - dynamic increase / decrease of workers based on IO load Author: Reviewed-by: Discussion: https://postgr.es/m/ Backpatch: --- src/include/miscadmin.h | 2 + src/include/postmaster/postmaster.h | 1 + src/include/storage/aio_subsys.h | 4 + src/include/storage/io_worker.h | 22 +++ src/include/storage/proc.h | 4 +- src/backend/postmaster/launch_backend.c | 2 + src/backend/postmaster/pmchild.c | 1 + src/backend/postmaster/postmaster.c | 165 ++++++++++++++++-- src/backend/storage/aio/Makefile | 1 + src/backend/storage/aio/meson.build | 1 + src/backend/storage/aio/method_worker.c | 88 ++++++++++ src/backend/tcop/postgres.c | 2 + src/backend/utils/activity/pgstat_backend.c | 1 + src/backend/utils/activity/pgstat_io.c | 1 + .../utils/activity/wait_event_names.txt | 1 + src/backend/utils/init/miscinit.c | 3 + 16 files changed, 286 insertions(+), 13 deletions(-) create mode 100644 src/include/storage/io_worker.h create mode 100644 src/backend/storage/aio/method_worker.c diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h index a2b63495eec..54429e046a9 100644 --- a/src/include/miscadmin.h +++ b/src/include/miscadmin.h @@ -360,6 +360,7 @@ typedef enum BackendType B_ARCHIVER, B_BG_WRITER, B_CHECKPOINTER, + B_IO_WORKER, B_STARTUP, B_WAL_RECEIVER, B_WAL_SUMMARIZER, @@ -389,6 +390,7 @@ extern PGDLLIMPORT BackendType MyBackendType; #define AmWalReceiverProcess() (MyBackendType == B_WAL_RECEIVER) #define AmWalSummarizerProcess() (MyBackendType == B_WAL_SUMMARIZER) #define AmWalWriterProcess() (MyBackendType == B_WAL_WRITER) +#define AmIoWorkerProcess() (MyBackendType == B_IO_WORKER) #define AmSpecialWorkerProcess() \ (AmAutoVacuumLauncherProcess() || \ diff --git a/src/include/postmaster/postmaster.h b/src/include/postmaster/postmaster.h index 188a06e2379..253dc98c50e 100644 --- a/src/include/postmaster/postmaster.h +++ b/src/include/postmaster/postmaster.h @@ -98,6 +98,7 @@ extern void InitProcessGlobals(void); extern int MaxLivePostmasterChildren(void); extern bool PostmasterMarkPIDForWorkerNotify(int); +extern void assign_io_workers(int newval, void *extra); #ifdef WIN32 extern void pgwin32_register_deadchild_callback(HANDLE procHandle, DWORD procId); diff --git a/src/include/storage/aio_subsys.h b/src/include/storage/aio_subsys.h index e4faf692a38..ed00d5c47cd 100644 --- a/src/include/storage/aio_subsys.h +++ b/src/include/storage/aio_subsys.h @@ -30,4 +30,8 @@ extern void pgaio_init_backend(void); extern void pgaio_error_cleanup(void); extern void AtEOXact_Aio(bool is_commit); + +/* aio_worker.c */ +extern bool pgaio_workers_enabled(void); + #endif /* AIO_SUBSYS_H */ diff --git a/src/include/storage/io_worker.h b/src/include/storage/io_worker.h new file mode 100644 index 00000000000..223d614dc4a --- /dev/null +++ b/src/include/storage/io_worker.h @@ -0,0 +1,22 @@ +/*------------------------------------------------------------------------- + * + * io_worker.h + * IO worker for implementing AIO "ourselves" + * + * + * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/include/storage/io.h + * + *------------------------------------------------------------------------- + */ +#ifndef IO_WORKER_H +#define IO_WORKER_H + + +extern void IoWorkerMain(char *startup_data, size_t startup_data_len) pg_attribute_noreturn(); + +extern int io_workers; + +#endif /* IO_WORKER_H */ diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h index 20777f7d5ae..64e9b8ff8c5 100644 --- a/src/include/storage/proc.h +++ b/src/include/storage/proc.h @@ -448,7 +448,9 @@ extern PGDLLIMPORT PGPROC *PreparedXactProcs; * 2 slots, but WAL writer is launched only after startup has exited, so we * only need 6 slots. */ -#define NUM_AUXILIARY_PROCS 6 +#define MAX_IO_WORKERS 32 +#define NUM_AUXILIARY_PROCS (6 + MAX_IO_WORKERS) + /* configurable options */ extern PGDLLIMPORT int DeadlockTimeout; diff --git a/src/backend/postmaster/launch_backend.c b/src/backend/postmaster/launch_backend.c index a97a1eda6da..54b4c22bd63 100644 --- a/src/backend/postmaster/launch_backend.c +++ b/src/backend/postmaster/launch_backend.c @@ -48,6 +48,7 @@ #include "replication/slotsync.h" #include "replication/walreceiver.h" #include "storage/dsm.h" +#include "storage/io_worker.h" #include "storage/pg_shmem.h" #include "tcop/backend_startup.h" #include "utils/memutils.h" @@ -197,6 +198,7 @@ static child_process_kind child_process_kinds[] = { [B_ARCHIVER] = {"archiver", PgArchiverMain, true}, [B_BG_WRITER] = {"bgwriter", BackgroundWriterMain, true}, [B_CHECKPOINTER] = {"checkpointer", CheckpointerMain, true}, + [B_IO_WORKER] = {"io_worker", IoWorkerMain, true}, [B_STARTUP] = {"startup", StartupProcessMain, true}, [B_WAL_RECEIVER] = {"wal_receiver", WalReceiverMain, true}, [B_WAL_SUMMARIZER] = {"wal_summarizer", WalSummarizerMain, true}, diff --git a/src/backend/postmaster/pmchild.c b/src/backend/postmaster/pmchild.c index 0d473226c3a..cde1d23a4ca 100644 --- a/src/backend/postmaster/pmchild.c +++ b/src/backend/postmaster/pmchild.c @@ -101,6 +101,7 @@ InitPostmasterChildSlots(void) pmchild_pools[B_AUTOVAC_WORKER].size = autovacuum_worker_slots; pmchild_pools[B_BG_WORKER].size = max_worker_processes; + pmchild_pools[B_IO_WORKER].size = MAX_IO_WORKERS; /* * There can be only one of each of these running at a time. They each diff --git a/src/backend/postmaster/postmaster.c b/src/backend/postmaster/postmaster.c index bb22b13adef..abef9d941c0 100644 --- a/src/backend/postmaster/postmaster.c +++ b/src/backend/postmaster/postmaster.c @@ -108,9 +108,12 @@ #include "replication/logicallauncher.h" #include "replication/slotsync.h" #include "replication/walsender.h" +#include "storage/aio_subsys.h" #include "storage/fd.h" +#include "storage/io_worker.h" #include "storage/ipc.h" #include "storage/pmsignal.h" +#include "storage/proc.h" #include "tcop/backend_startup.h" #include "tcop/tcopprot.h" #include "utils/datetime.h" @@ -341,6 +344,7 @@ typedef enum * ckpt */ PM_WAIT_XLOG_ARCHIVAL, /* waiting for archiver and walsenders to * finish */ + PM_WAIT_IO_WORKERS, /* waiting for io workers to exit */ PM_WAIT_CHECKPOINTER, /* waiting for checkpointer to shut down */ PM_WAIT_DEAD_END, /* waiting for dead-end children to exit */ PM_NO_CHILDREN, /* all important children have exited */ @@ -403,6 +407,10 @@ bool LoadedSSL = false; static DNSServiceRef bonjour_sdref = NULL; #endif +/* State for IO worker management. */ +static int io_worker_count = 0; +static PMChild *io_worker_children[MAX_IO_WORKERS]; + /* * postmaster.c - function prototypes */ @@ -437,6 +445,8 @@ static void TerminateChildren(int signal); static int CountChildren(BackendTypeMask targetMask); static void LaunchMissingBackgroundProcesses(void); static void maybe_start_bgworkers(void); +static bool maybe_reap_io_worker(int pid); +static void maybe_adjust_io_workers(void); static bool CreateOptsFile(int argc, char *argv[], char *fullprogname); static PMChild *StartChildProcess(BackendType type); static void StartSysLogger(void); @@ -1366,6 +1376,11 @@ PostmasterMain(int argc, char *argv[]) */ AddToDataDirLockFile(LOCK_FILE_LINE_PM_STATUS, PM_STATUS_STARTING); + UpdatePMState(PM_STARTUP); + + /* Make sure we can perform I/O while starting up. */ + maybe_adjust_io_workers(); + /* Start bgwriter and checkpointer so they can help with recovery */ if (CheckpointerPMChild == NULL) CheckpointerPMChild = StartChildProcess(B_CHECKPOINTER); @@ -1378,7 +1393,6 @@ PostmasterMain(int argc, char *argv[]) StartupPMChild = StartChildProcess(B_STARTUP); Assert(StartupPMChild != NULL); StartupStatus = STARTUP_RUNNING; - UpdatePMState(PM_STARTUP); /* Some workers may be scheduled to start now */ maybe_start_bgworkers(); @@ -2503,6 +2517,16 @@ process_pm_child_exit(void) continue; } + /* Was it an IO worker? */ + if (maybe_reap_io_worker(pid)) + { + if (!EXIT_STATUS_0(exitstatus) && !EXIT_STATUS_1(exitstatus)) + HandleChildCrash(pid, exitstatus, _("io worker")); + + maybe_adjust_io_workers(); + continue; + } + /* * Was it a backend or a background worker? */ @@ -2724,6 +2748,7 @@ HandleFatalError(QuitSignalReason reason, bool consider_sigabrt) case PM_WAIT_XLOG_SHUTDOWN: case PM_WAIT_XLOG_ARCHIVAL: case PM_WAIT_CHECKPOINTER: + case PM_WAIT_IO_WORKERS: /* * NB: Similar code exists in PostmasterStateMachine()'s handling @@ -2906,20 +2931,21 @@ PostmasterStateMachine(void) /* * If we are doing crash recovery or an immediate shutdown then we - * expect archiver, checkpointer and walsender to exit as well, - * otherwise not. + * expect archiver, checkpointer, io workers and walsender to exit as + * well, otherwise not. */ if (FatalError || Shutdown >= ImmediateShutdown) targetMask = btmask_add(targetMask, B_CHECKPOINTER, B_ARCHIVER, + B_IO_WORKER, B_WAL_SENDER); /* - * Normally walsenders and archiver will continue running; they will - * be terminated later after writing the checkpoint record. We also - * let dead-end children to keep running for now. The syslogger - * process exits last. + * Normally archiver, checkpointer, IO workers and walsenders will + * continue running; they will be terminated later after writing the + * checkpoint record. We also let dead-end children to keep running + * for now. The syslogger process exits last. * * This assertion checks that we have covered all backend types, * either by including them in targetMask, or by noting here that they @@ -2934,12 +2960,13 @@ PostmasterStateMachine(void) B_LOGGER); /* - * Archiver, checkpointer and walsender may or may not be in - * targetMask already. + * Archiver, checkpointer, IO workers, and walsender may or may + * not be in targetMask already. */ remainMask = btmask_add(remainMask, B_ARCHIVER, B_CHECKPOINTER, + B_IO_WORKER, B_WAL_SENDER); /* these are not real postmaster children */ @@ -3040,11 +3067,25 @@ PostmasterStateMachine(void) { /* * PM_WAIT_XLOG_ARCHIVAL state ends when there are no children other - * than checkpointer, dead-end children and logger left. There + * than checkpointer, io workers and dead-end children left. There * shouldn't be any regular backends left by now anyway; what we're * really waiting for is for walsenders and archiver to exit. */ - if (CountChildren(btmask_all_except(B_CHECKPOINTER, B_LOGGER, B_DEAD_END_BACKEND)) == 0) + if (CountChildren(btmask_all_except(B_CHECKPOINTER, B_IO_WORKER, + B_LOGGER, B_DEAD_END_BACKEND)) == 0) + { + UpdatePMState(PM_WAIT_IO_WORKERS); + SignalChildren(SIGUSR2, btmask(B_IO_WORKER)); + } + } + + if (pmState == PM_WAIT_IO_WORKERS) + { + /* + * PM_WAIT_IO_WORKERS state ends when there's only checkpointer and + * dead_end children left. + */ + if (io_worker_count == 0) { UpdatePMState(PM_WAIT_CHECKPOINTER); @@ -3172,10 +3213,14 @@ PostmasterStateMachine(void) /* re-create shared memory and semaphores */ CreateSharedMemoryAndSemaphores(); + UpdatePMState(PM_STARTUP); + + /* Make sure we can perform I/O while starting up. */ + maybe_adjust_io_workers(); + StartupPMChild = StartChildProcess(B_STARTUP); Assert(StartupPMChild != NULL); StartupStatus = STARTUP_RUNNING; - UpdatePMState(PM_STARTUP); /* crash recovery started, reset SIGKILL flag */ AbortStartTime = 0; @@ -3199,6 +3244,7 @@ pmstate_name(PMState state) PM_TOSTR_CASE(PM_WAIT_BACKENDS); PM_TOSTR_CASE(PM_WAIT_XLOG_SHUTDOWN); PM_TOSTR_CASE(PM_WAIT_XLOG_ARCHIVAL); + PM_TOSTR_CASE(PM_WAIT_IO_WORKERS); PM_TOSTR_CASE(PM_WAIT_DEAD_END); PM_TOSTR_CASE(PM_WAIT_CHECKPOINTER); PM_TOSTR_CASE(PM_NO_CHILDREN); @@ -4115,6 +4161,7 @@ bgworker_should_start_now(BgWorkerStartTime start_time) case PM_WAIT_DEAD_END: case PM_WAIT_XLOG_ARCHIVAL: case PM_WAIT_XLOG_SHUTDOWN: + case PM_WAIT_IO_WORKERS: case PM_WAIT_BACKENDS: case PM_STOP_BACKENDS: break; @@ -4265,6 +4312,100 @@ maybe_start_bgworkers(void) } } +static bool +maybe_reap_io_worker(int pid) +{ + for (int id = 0; id < MAX_IO_WORKERS; ++id) + { + if (io_worker_children[id] && + io_worker_children[id]->pid == pid) + { + ReleasePostmasterChildSlot(io_worker_children[id]); + + --io_worker_count; + io_worker_children[id] = NULL; + return true; + } + } + return false; +} + +static void +maybe_adjust_io_workers(void) +{ + if (!pgaio_workers_enabled()) + return; + + /* + * If we're in final shutting down state, then we're just waiting for all + * processes to exit. + */ + if (pmState >= PM_WAIT_IO_WORKERS) + return; + + /* Don't start new workers during an immediate shutdown either. */ + if (Shutdown >= ImmediateShutdown) + return; + + /* + * Don't start new workers if we're in the shutdown phase of a crash + * restart. But we *do* need to start if we're already starting up again. + */ + if (FatalError && pmState >= PM_STOP_BACKENDS) + return; + + Assert(pmState < PM_WAIT_IO_WORKERS); + + /* Not enough running? */ + while (io_worker_count < io_workers) + { + PMChild *child; + int id; + + /* find unused entry in io_worker_children array */ + for (id = 0; id < MAX_IO_WORKERS; ++id) + { + if (io_worker_children[id] == NULL) + break; + } + if (id == MAX_IO_WORKERS) + elog(ERROR, "could not find a free IO worker ID"); + + /* Try to launch one. */ + child = StartChildProcess(B_IO_WORKER); + if (child != NULL) + { + io_worker_children[id] = child; + ++io_worker_count; + } + else + break; /* XXX try again soon? */ + } + + /* Too many running? */ + if (io_worker_count > io_workers) + { + /* ask the IO worker in the highest slot to exit */ + for (int id = MAX_IO_WORKERS - 1; id >= 0; --id) + { + if (io_worker_children[id] != NULL) + { + kill(io_worker_children[id]->pid, SIGUSR2); + break; + } + } + } +} + +void +assign_io_workers(int newval, void *extra) +{ + io_workers = newval; + if (!IsUnderPostmaster && pmState > PM_INIT) + maybe_adjust_io_workers(); +} + + /* * When a backend asks to be notified about worker state changes, we * set a flag in its backend entry. The background worker machinery needs diff --git a/src/backend/storage/aio/Makefile b/src/backend/storage/aio/Makefile index 89f821ea7e1..f51c34a37f8 100644 --- a/src/backend/storage/aio/Makefile +++ b/src/backend/storage/aio/Makefile @@ -15,6 +15,7 @@ OBJS = \ aio_io.o \ aio_target.o \ method_sync.o \ + method_worker.o \ read_stream.o include $(top_srcdir)/src/backend/common.mk diff --git a/src/backend/storage/aio/meson.build b/src/backend/storage/aio/meson.build index 2c26089d52e..74f94c6e40b 100644 --- a/src/backend/storage/aio/meson.build +++ b/src/backend/storage/aio/meson.build @@ -7,5 +7,6 @@ backend_sources += files( 'aio_io.c', 'aio_target.c', 'method_sync.c', + 'method_worker.c', 'read_stream.c', ) diff --git a/src/backend/storage/aio/method_worker.c b/src/backend/storage/aio/method_worker.c new file mode 100644 index 00000000000..942f1609af2 --- /dev/null +++ b/src/backend/storage/aio/method_worker.c @@ -0,0 +1,88 @@ +/*------------------------------------------------------------------------- + * + * method_worker.c + * AIO - perform AIO using worker processes + * + * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * IDENTIFICATION + * src/backend/storage/aio/method_worker.c + * + *------------------------------------------------------------------------- + */ + +#include "postgres.h" + +#include "libpq/pqsignal.h" +#include "miscadmin.h" +#include "postmaster/auxprocess.h" +#include "postmaster/interrupt.h" +#include "storage/aio_subsys.h" +#include "storage/io_worker.h" +#include "storage/ipc.h" +#include "storage/latch.h" +#include "storage/proc.h" +#include "tcop/tcopprot.h" +#include "utils/wait_event.h" + + +/* GUCs */ +int io_workers = 3; + + +void +IoWorkerMain(char *startup_data, size_t startup_data_len) +{ + sigjmp_buf local_sigjmp_buf; + + MyBackendType = B_IO_WORKER; + AuxiliaryProcessMainCommon(); + + /* TODO review all signals */ + pqsignal(SIGHUP, SignalHandlerForConfigReload); + pqsignal(SIGINT, die); /* to allow manually triggering worker restart */ + + /* + * Ignore SIGTERM, will get explicit shutdown via SIGUSR2 later in the + * shutdown sequence, similar to checkpointer. + */ + pqsignal(SIGTERM, SIG_IGN); + /* SIGQUIT handler was already set up by InitPostmasterChild */ + pqsignal(SIGALRM, SIG_IGN); + pqsignal(SIGPIPE, SIG_IGN); + pqsignal(SIGUSR1, procsignal_sigusr1_handler); + pqsignal(SIGUSR2, SignalHandlerForShutdownRequest); + sigprocmask(SIG_SETMASK, &UnBlockSig, NULL); + + /* see PostgresMain() */ + if (sigsetjmp(local_sigjmp_buf, 1) != 0) + { + error_context_stack = NULL; + HOLD_INTERRUPTS(); + + EmitErrorReport(); + + proc_exit(1); + } + + /* We can now handle ereport(ERROR) */ + PG_exception_stack = &local_sigjmp_buf; + + while (!ShutdownRequestPending) + { + WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, -1, + WAIT_EVENT_IO_WORKER_MAIN); + ResetLatch(MyLatch); + CHECK_FOR_INTERRUPTS(); + } + + proc_exit(0); +} + +bool +pgaio_workers_enabled(void) +{ + /* placeholder for future commit */ + return false; +} diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c index 1149d89d7a1..bf6f204e71c 100644 --- a/src/backend/tcop/postgres.c +++ b/src/backend/tcop/postgres.c @@ -3313,6 +3313,8 @@ ProcessInterrupts(void) (errcode(ERRCODE_ADMIN_SHUTDOWN), errmsg("terminating background worker \"%s\" due to administrator command", MyBgworkerEntry->bgw_type))); + else if (AmIoWorkerProcess()) + proc_exit(0); else ereport(FATAL, (errcode(ERRCODE_ADMIN_SHUTDOWN), diff --git a/src/backend/utils/activity/pgstat_backend.c b/src/backend/utils/activity/pgstat_backend.c index 4a667e7019c..d7f7bc8d2e3 100644 --- a/src/backend/utils/activity/pgstat_backend.c +++ b/src/backend/utils/activity/pgstat_backend.c @@ -238,6 +238,7 @@ pgstat_tracks_backend_bktype(BackendType bktype) case B_LOGGER: case B_BG_WRITER: case B_CHECKPOINTER: + case B_IO_WORKER: case B_STARTUP: return false; diff --git a/src/backend/utils/activity/pgstat_io.c b/src/backend/utils/activity/pgstat_io.c index 28a431084b8..713b1dc6cd0 100644 --- a/src/backend/utils/activity/pgstat_io.c +++ b/src/backend/utils/activity/pgstat_io.c @@ -375,6 +375,7 @@ pgstat_tracks_io_bktype(BackendType bktype) case B_BG_WORKER: case B_BG_WRITER: case B_CHECKPOINTER: + case B_IO_WORKER: case B_SLOTSYNC_WORKER: case B_STANDALONE_BACKEND: case B_STARTUP: diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt index 6f3ca878bd1..d18b788b8a1 100644 --- a/src/backend/utils/activity/wait_event_names.txt +++ b/src/backend/utils/activity/wait_event_names.txt @@ -57,6 +57,7 @@ BGWRITER_HIBERNATE "Waiting in background writer process, hibernating." BGWRITER_MAIN "Waiting in main loop of background writer process." CHECKPOINTER_MAIN "Waiting in main loop of checkpointer process." CHECKPOINTER_SHUTDOWN "Waiting for checkpointer process to be terminated." +IO_WORKER_MAIN "Waiting in main loop of IO Worker process." LOGICAL_APPLY_MAIN "Waiting in main loop of logical replication apply process." LOGICAL_LAUNCHER_MAIN "Waiting in main loop of logical replication launcher process." LOGICAL_PARALLEL_APPLY_MAIN "Waiting in main loop of logical replication parallel apply process." diff --git a/src/backend/utils/init/miscinit.c b/src/backend/utils/init/miscinit.c index 0347fc11092..cbca090d2b0 100644 --- a/src/backend/utils/init/miscinit.c +++ b/src/backend/utils/init/miscinit.c @@ -293,6 +293,9 @@ GetBackendTypeDesc(BackendType backendType) case B_CHECKPOINTER: backendDesc = gettext_noop("checkpointer"); break; + case B_IO_WORKER: + backendDesc = "io worker"; + break; case B_LOGGER: backendDesc = gettext_noop("logger"); break; -- 2.48.1.76.g4e746b1a31.dirty